[enocean] Refactor transceiver receiver (#20449)

* [enocean] Refactor transceiver receiver

Signed-off-by: Ravi Nadahar <nadahar@rediffmail.com>

bundles/org.openhab.binding.enocean/src/main/java/org/openhab/binding/enocean/internal/handler/EnOceanBridgeHandler.java

@@ -383,8 +383,7 @@ public class EnOceanBridgeHandler extends ConfigStatusBridgeHandler implements T
         sendingThings[id] = null;
     }
 
-    public <T extends @Nullable Response> void sendMessage(BasePacket message,
-            @Nullable ResponseListener<T> responseListener) {
+    public <T extends Response> void sendMessage(BasePacket message, @Nullable ResponseListener<T> responseListener) {
         try {
             EnOceanTransceiver localTransceiver = transceiver;
             if (localTransceiver == null) {

bundles/org.openhab.binding.enocean/src/main/java/org/openhab/binding/enocean/internal/messages/ESP2Packet.java

@@ -14,8 +14,8 @@ package org.openhab.binding.enocean.internal.messages;
 
 import static org.openhab.binding.enocean.internal.EnOceanBindingConstants.ZERO;
 
-import java.security.InvalidParameterException;
 import java.util.Arrays;
+import java.util.HexFormat;
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
 import org.openhab.binding.enocean.internal.EnOceanException;
@@ -31,6 +31,7 @@ public class ESP2Packet {
 
     public static final byte ENOCEAN_ESP2_FIRSTSYNC_BYTE = (byte) 0xA5;
     public static final byte ENOCEAN_ESP2_SECONDSYNC_BYTE = 0x5A;
+    public static final byte ENOCEAN_ESP2_INTERNAL_COMMAND_BYTE = (byte) 0xFC;
 
     private static final int ESP2_SYNC_BYTE_LENGTH = 2;
     private static final int ESP2_HEADER_LENGTH = 1;
@@ -104,10 +105,10 @@ public class ESP2Packet {
     }
 
     public enum ESP2PacketType {
-        Receive_Radio_Telegram((byte) 0x00),
-        Transmit_Radio_Telegram((byte) 0x03),
-        Receive_Message_Telegram((byte) 0x04),
-        Transmit_Command_Telegram((byte) 0x05);
+        RECEIVE_RADIO_TELEGRAM((byte) 0x00),
+        TRANSMIT_RADIO_TELEGRAM((byte) 0x03),
+        RECEIVE_MESSAGE_TELEGRAM((byte) 0x04),
+        TRANSMIT_COMMAND_TELEGRAM((byte) 0x05);
 
         private byte value;
 
@@ -122,7 +123,8 @@ public class ESP2Packet {
                 }
             }
 
-            throw new InvalidParameterException("Unknown ESP2 PacketType value");
+            throw new IllegalArgumentException(
+                    "Unknown ESP2 PacketType value '0x" + HexFormat.of().toHexDigits(packetType) + '\'');
         }
     }
 
@@ -135,10 +137,10 @@ public class ESP2Packet {
     private ESP2PacketType convertToESP2PacketType(ESPPacketType espPacketType) {
         switch (espPacketType) {
             case COMMON_COMMAND:
-                return ESP2PacketType.Transmit_Command_Telegram;
+                return ESP2PacketType.TRANSMIT_COMMAND_TELEGRAM;
             case RADIO_ERP1:
             case RADIO_ERP2:
-                return ESP2PacketType.Transmit_Radio_Telegram;
+                return ESP2PacketType.TRANSMIT_RADIO_TELEGRAM;
             case RESPONSE: // Response is not intended for outbound data (at least for ESP2)
             default:
                 throw new IllegalArgumentException("ESPPacketType not supported");
@@ -154,7 +156,7 @@ public class ESP2Packet {
             case _4BS:
                 return ORG._4BS.value;
             default:
-                throw new InvalidParameterException("RORG is not supported by ESP2");
+                throw new IllegalArgumentException("RORG is not supported by ESP2");
         }
     }
 
@@ -181,7 +183,7 @@ public class ESP2Packet {
                     System.arraycopy(message.getPayload(ESP2_ORG_LENGTH, 4), 0, data, ESP2_ORG_LENGTH, 4);
                     break;
                 default:
-                    throw new InvalidParameterException("CCMessage is not supported by ESP2");
+                    throw new IllegalArgumentException("CCMessage is not supported by ESP2");
             }
         }
 
@@ -251,12 +253,16 @@ public class ESP2Packet {
         }
     }
 
-    public static boolean validateCheckSum(byte[] data, int length, byte checkSum) {
+    public static boolean validateCheckSum(byte[] data, int start, int length) {
         int sum = 0;
-        for (int i = 0; i < length; i++) {
+        int end = start + length;
+        if (data.length <= end) {
+            return false;
+        }
+        for (int i = start; i < end; i++) {
             sum += (data[i] & 0xff);
         }
 
-        return (sum & 0xff) == (checkSum & 0xff);
+        return (sum & 0xff) == (data[end] & 0xff);
     }
 }

bundles/org.openhab.binding.enocean/src/main/java/org/openhab/binding/enocean/internal/messages/ESP2PacketConverter.java

@@ -37,29 +37,29 @@ public class ESP2PacketConverter {
     private static final int ESP3PACKET_BASE_LENGTH = ESP3Packet.ESP3_RORG_LENGTH + ESP3Packet.ESP3_SENDERID_LENGTH
             + ESP3Packet.ESP3_STATUS_LENGTH;
 
-    private static @Nullable BasePacket handleRadioTelegram(int dataLength, byte packetType, byte[] payload) {
-        switch (ESP2Packet.ORG.getORG(payload[1])) {
+    private static @Nullable BasePacket handleRadioTelegram(byte[] bytes) {
+        switch (ESP2Packet.ORG.getORG(bytes[1])) {
             case _RPS:
                 return ESP3PacketFactory.buildPacket(ESP3PACKET_BASE_LENGTH + RORG.RPS.getDataLength(), 0,
-                        ESPPacketType.RADIO_ERP1.getValue(), new byte[] { RORG.RPS.getValue(), payload[2], payload[6],
-                                payload[7], payload[8], payload[9], payload[10] });
+                        ESPPacketType.RADIO_ERP1.getValue(), new byte[] { RORG.RPS.getValue(), bytes[2], bytes[6],
+                                bytes[7], bytes[8], bytes[9], bytes[10] });
             case _1BS:
                 return ESP3PacketFactory.buildPacket(ESP3PACKET_BASE_LENGTH + RORG._1BS.getDataLength(), 0,
-                        ESPPacketType.RADIO_ERP1.getValue(), new byte[] { RORG._1BS.getValue(), payload[2], payload[6],
-                                payload[7], payload[8], payload[9], payload[10] });
+                        ESPPacketType.RADIO_ERP1.getValue(), new byte[] { RORG._1BS.getValue(), bytes[2], bytes[6],
+                                bytes[7], bytes[8], bytes[9], bytes[10] });
 
             case _4BS:
                 return ESP3PacketFactory.buildPacket(ESP3PACKET_BASE_LENGTH + RORG._4BS.getDataLength(), 0,
-                        ESPPacketType.RADIO_ERP1.getValue(), new byte[] { RORG._4BS.getValue(), payload[2], payload[3],
-                                payload[4], payload[5], payload[6], payload[7], payload[8], payload[9], payload[10] });
+                        ESPPacketType.RADIO_ERP1.getValue(), new byte[] { RORG._4BS.getValue(), bytes[2], bytes[3],
+                                bytes[4], bytes[5], bytes[6], bytes[7], bytes[8], bytes[9], bytes[10] });
             default:
-                logger.debug("Received unsupported ORG: {}", payload[1]);
+                logger.debug("Received unsupported ORG: {}", bytes[1]);
                 return null;
         }
     }
 
-    private static @Nullable BasePacket handleMessageTelegram(int dataLength, byte packetType, byte[] payload) {
-        switch (ESP2Packet.ESP2Response.getResponse(payload[1])) {
+    private static @Nullable BasePacket handleMessageTelegram(byte[] bytes) {
+        switch (ESP2Packet.ESP2Response.getResponse(bytes[1])) {
             case OK:
                 return ESP3PacketFactory.buildPacket(1, 0, ESPPacketType.RESPONSE.getValue(),
                         new byte[] { ResponseType.RET_OK.getValue() });
@@ -70,14 +70,14 @@ public class ESP2PacketConverter {
                 byte[] data = new byte[33];
                 Arrays.fill(data, (byte) 0);
                 data[0] = ResponseType.RET_OK.getValue();
-                System.arraycopy(payload, 1, data, 1, 4);
+                System.arraycopy(bytes, 1, data, 1, 4);
 
                 byte[] description = "TCM 210".getBytes(Charset.forName("ASCII"));
                 System.arraycopy(description, 0, data, 17, description.length);
                 return ESP3PacketFactory.buildPacket(data.length, 0, ESPPacketType.RESPONSE.getValue(), data);
             }
             case UNKOWN: // try to interpret it as a radio telegram
-                return handleRadioTelegram(dataLength, packetType, payload);
+                return handleRadioTelegram(bytes);
 
             case ERR_IDRANGE:
             case ERR_MODEM_DUP_ID:
@@ -93,36 +93,47 @@ public class ESP2PacketConverter {
             case INF_RX_SENSITIVITY:
             default:
                 logger.debug("Received unsupported message telegram: {}",
-                        ESP2Packet.ESP2Response.getResponse(payload[1]).name());
+                        ESP2Packet.ESP2Response.getResponse(bytes[1]).name());
                 return null;
         }
     }
 
-    public static @Nullable BasePacket buildPacket(int dataLength, byte packetType, byte[] payload) {
-        ESP2PacketType type = ESP2PacketType.getPacketType(packetType);
+    public static @Nullable BasePacket buildPacket(ESP2PacketType type, byte[] bytes, int start, int length) {
+        if (start == 0 && length == bytes.length) {
+            return buildPacket(type, bytes);
+        }
+        if (start + length > bytes.length) {
+            throw new IndexOutOfBoundsException("Index out of range: start=" + start + ", length=" + length
+                    + " -> index=" + (start + length) + ", array length=" + bytes.length);
+        }
+        byte[] packetBytes = new byte[length];
+        System.arraycopy(bytes, start, packetBytes, 0, length);
+        return buildPacket(type, packetBytes);
+    }
 
+    public static @Nullable BasePacket buildPacket(ESP2PacketType type, byte[] bytes) {
         switch (type) {
-            case Receive_Radio_Telegram: // RRT
-                logger.debug("Received ESP2 radio telegram: {}", HexUtils.bytesToHex(payload));
-                return handleRadioTelegram(dataLength, packetType, payload);
+            case RECEIVE_RADIO_TELEGRAM: // RRT
+                logger.debug("Received ESP2 radio telegram: {}", HexUtils.bytesToHex(bytes));
+                return handleRadioTelegram(bytes);
 
-            case Receive_Message_Telegram: // RMT => Response
-                logger.debug("Received ESP2 message telegram: {}", HexUtils.bytesToHex(payload));
-                return handleMessageTelegram(dataLength, packetType, payload);
+            case RECEIVE_MESSAGE_TELEGRAM: // RMT => Response
+                logger.debug("Received ESP2 message telegram: {}", HexUtils.bytesToHex(bytes));
+                return handleMessageTelegram(bytes);
 
-            case Transmit_Radio_Telegram: // TRT
+            case TRANSMIT_RADIO_TELEGRAM: // TRT
                 // This should never happen, as this telegram is just for outbound data
-                logger.trace("Received Transmit_Radio_Telegram: {}", HexUtils.bytesToHex(payload));
+                logger.trace("Received Transmit_Radio_Telegram: {}", HexUtils.bytesToHex(bytes));
                 break;
 
-            case Transmit_Command_Telegram: // TCT => CommonCommand
+            case TRANSMIT_COMMAND_TELEGRAM: // TCT => CommonCommand
                 // this should also never happen, as this telegram is also just for outbound data
                 // however FAM14 receives periodically 0xABFC messages
-                if (payload[1] == (byte) 0xFC) {
+                if (bytes[1] == (byte) 0xFC) {
                     return null;
                 }
 
-                logger.trace("Received Transmit_Command_Telegram: {}", HexUtils.bytesToHex(payload));
+                logger.trace("Received Transmit_Command_Telegram: {}", HexUtils.bytesToHex(bytes));
                 break;
         }
 

bundles/org.openhab.binding.enocean/src/main/java/org/openhab/binding/enocean/internal/messages/ESP3Packet.java

@@ -112,12 +112,16 @@ public class ESP3Packet {
         }
     }
 
-    public static boolean checkCRC8(byte[] data, int length, byte crc8) {
+    public static boolean checkCRC8(byte[] data, int start, int length) {
         byte output = 0;
-        for (int i = 0; i < length; i++) {
+        int end = start + length;
+        if (data.length <= end) {
+            return false;
+        }
+        for (int i = start; i < end; i++) {
             int index = (output ^ data[i]) & 0xff;
             output = crc8Table[index];
         }
-        return output == crc8;
+        return output == data[end];
     }
 }

bundles/org.openhab.binding.enocean/src/main/java/org/openhab/binding/enocean/internal/messages/Response.java

@@ -36,7 +36,7 @@ public class Response extends BasePacket {
         RET_BASEID_OUT_OF_RANGE((byte) 0x90),
         RET_BASEID_MAX_REACHED((byte) 0x91);
 
-        private byte value;
+        private final byte value;
 
         ResponseType(byte value) {
             this.value = value;

bundles/org.openhab.binding.enocean/src/main/java/org/openhab/binding/enocean/internal/transceiver/EnOceanESP2Transceiver.java

@@ -12,11 +12,11 @@
  */
 package org.openhab.binding.enocean.internal.transceiver;
 
+import java.io.BufferedInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.Arrays;
-import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
 import org.eclipse.jdt.annotation.Nullable;
@@ -25,8 +25,10 @@ import org.openhab.binding.enocean.internal.messages.BasePacket;
 import org.openhab.binding.enocean.internal.messages.ERP1Message;
 import org.openhab.binding.enocean.internal.messages.ERP1Message.RORG;
 import org.openhab.binding.enocean.internal.messages.ESP2Packet;
+import org.openhab.binding.enocean.internal.messages.ESP2Packet.ESP2PacketType;
 import org.openhab.binding.enocean.internal.messages.ESP2PacketConverter;
 import org.openhab.binding.enocean.internal.messages.Response;
+import org.openhab.binding.enocean.internal.util.EnOceanUtil;
 import org.openhab.core.io.transport.serial.SerialPortManager;
 import org.openhab.core.util.HexUtils;
 
@@ -37,134 +39,247 @@ import org.openhab.core.util.HexUtils;
 @NonNullByDefault
 public class EnOceanESP2Transceiver extends EnOceanTransceiver {
 
+    private static final String RX_THREAD_NAME_PREFIX = "OH-binding-enocean-ESP2-RX-";
+    private static final AtomicInteger THREAD_NUM = new AtomicInteger();
+
+    // All access must be guarded by "this"
+    private @Nullable Thread worker;
+
     public EnOceanESP2Transceiver(String path, TransceiverErrorListener errorListener,
             ScheduledExecutorService scheduler, @Nullable SerialPortManager serialPortManager) {
         super(path, errorListener, scheduler, serialPortManager);
     }
 
     enum ReadingState {
-        WaitingForFirstSyncByte,
-        WaitingForSecondSyncByte,
-        ReadingHeader,
-        ReadingData
+        WAIT_FIRST_SYNCBYTE,
+        WAIT_SECOND_SYNCBYTE,
+        READ_HEADER,
+        READ_DATA
     }
 
-    byte[] dataBuffer = new byte[ESP2Packet.ESP_PACKET_LENGTH];
-    ReadingState state = ReadingState.WaitingForFirstSyncByte;
-    int currentPosition = 0;
-    int dataLength = -1;
-    byte packetType = -1;
+    @Override
+    public void startReceiving(ScheduledExecutorService scheduler) {
+        Thread worker;
+        InputStream is;
+        synchronized (this) {
+            worker = this.worker;
+            if (worker != null && worker.isAlive()) {
+                worker.interrupt();
+            }
+
+            is = inputStream;
+            if (is == null) {
+                this.worker = worker = null;
+            } else {
+                this.worker = worker = new Thread(new Receiver(is, errorListener, scheduler),
+                        RX_THREAD_NAME_PREFIX + THREAD_NUM.incrementAndGet());
+                worker.setUncaughtExceptionHandler((t, e) -> {
+                    logger.warn("Uncaught exception in EnOceanSerialTransceiver RX thread ({}): {}", t.getName(),
+                            e.getMessage());
+                    logger.trace("", e);
+                    TransceiverErrorListener listener = this.errorListener;
+                    if (listener != null) {
+                        scheduler.execute(() -> listener.errorOccurred(e));
+                    }
+                });
+            }
+        }
+        if (worker == null) {
+            logger.warn("Cannot read from null stream");
+            TransceiverErrorListener errorListener = this.errorListener;
+            if (errorListener != null) {
+                IOException e = new IOException("Cannot read from null stream");
+                scheduler.execute(() -> errorListener.errorOccurred(e));
+            }
+        } else {
+            worker.start();
+            logger.info("EnOceanSerialTransceiver RX thread ({}) started", worker.getName());
+        }
+    }
 
     @Override
-    protected void processMessage(byte firstByte) {
-        byte[] readingBuffer = new byte[ENOCEAN_MAX_DATA];
-        int bytesRead = -1;
-        byte byteBuffer;
+    protected void shutDownRx() {
+        Thread worker;
+        synchronized (this) {
+            worker = this.worker;
+            this.worker = null;
+        }
+        if (worker != null && worker.isAlive()) {
+            worker.interrupt();
+        }
+    }
 
-        try {
-            readingBuffer[0] = firstByte;
-            InputStream localInputStream = inputStream;
-            if (localInputStream == null) {
-                throw new IOException("could not read from inputstream, it was null");
-            }
-            bytesRead = localInputStream.read(readingBuffer, 1, localInputStream.available());
-            if (bytesRead == -1) {
-                throw new IOException("could not read from inputstream");
+    private class Receiver implements Runnable {
+
+        private final InputStream is;
+        private final @Nullable TransceiverErrorListener errorListener;
+        private final ScheduledExecutorService scheduler;
+
+        public Receiver(InputStream is, @Nullable TransceiverErrorListener errorListener,
+                ScheduledExecutorService scheduler) {
+            this.is = is;
+            this.errorListener = errorListener;
+            this.scheduler = scheduler;
+        }
+
+        @Override
+        public void run() {
+            byte[] bytes = new byte[64];
+            int read;
+            ReadingState state = ReadingState.WAIT_FIRST_SYNCBYTE;
+            int doRead = 1;
+            byte packetType = -1;
+            int packetLength = -1;
+            int packetStart = -1;
+            int pos = 0;
+            final Thread thread = Thread.currentThread();
+            InputStream is = this.is;
+            logger.trace("RX InputStream implementation: {}", is.getClass().getName());
+            if (!is.markSupported()) {
+                // Use this as a "rough indicator" that the stream isn't buffered
+                logger.trace("Wrapping {} in BufferedInputStream", is.getClass().getName());
+                is = new BufferedInputStream(is, ENOCEAN_MAX_DATA);
             }
+            byte[] packetBytes;
 
-            Future<?> localReadingTask = readingTask;
-            if (localReadingTask == null || localReadingTask.isCancelled()) {
-                return;
-            }
-
-            bytesRead++;
-            for (int p = 0; p < bytesRead; p++) {
-                byteBuffer = readingBuffer[p];
-
-                switch (state) {
-                    case WaitingForFirstSyncByte:
-                        if (byteBuffer == ESP2Packet.ENOCEAN_ESP2_FIRSTSYNC_BYTE) {
-                            state = ReadingState.WaitingForSecondSyncByte;
-                            logger.trace("Received First Sync Byte");
-                        }
-                        break;
-                    case WaitingForSecondSyncByte:
-                        if (byteBuffer == ESP2Packet.ENOCEAN_ESP2_SECONDSYNC_BYTE) {
-                            state = ReadingState.ReadingHeader;
-                            logger.trace("Received Second Sync Byte");
-                        }
-                        break;
-                    case ReadingHeader: {
-                        state = ReadingState.ReadingData;
-
-                        currentPosition = 0;
-                        dataBuffer[currentPosition++] = byteBuffer;
-                        dataLength = ((dataBuffer[0] & 0xFF) & 0b11111);
-                        packetType = (byte) ((dataBuffer[0] & 0xFF) >> 5);
-
-                        logger.trace(">> Received header, data length {} packet type {}", dataLength, packetType);
+            while (!thread.isInterrupted()) {
+                try {
+                    read = is.read(bytes, pos, doRead);
+                } catch (IOException e) {
+                    logger.debug("Unable to read from serial port: {}", e.getMessage());
+                    logger.trace("", e);
+                    TransceiverErrorListener errorListener = this.errorListener;
+                    if (errorListener != null && !thread.isInterrupted()) {
+                        // We don't want to take the Thing offline if an IOException is thrown when the port is closed,
+                        // which is why we check isInterrupted()
+                        errorListener.errorOccurred(e);
                     }
+                    break;
+                }
+                if (read <= 0) {
+                    // Unlike regular InputStreams, the serial port streams occasionally returns -1 even if the
+                    // stream is still "alive", so just accept it and try to read again. Add a short backoff to
+                    // avoid a tight loop and high CPU usage if this happens repeatedly.
+                    try {
+                        Thread.sleep(5);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
                         break;
-                    case ReadingData:
-                        if (currentPosition == dataLength) {
-                            if (ESP2Packet.validateCheckSum(dataBuffer, dataLength, byteBuffer)) {
-                                BasePacket packet = ESP2PacketConverter.buildPacket(dataLength, packetType, dataBuffer);
-                                if (packet != null) {
-                                    switch (packet.getPacketType()) {
-                                        case RADIO_ERP1: {
-                                            ERP1Message msg = (ERP1Message) packet;
-                                            logger.debug("Converted to: {} with RORG {} for {}",
-                                                    packet.getPacketType().name(), msg.getRORG().name(),
-                                                    HexUtils.bytesToHex(msg.getSenderId()));
+                    }
+                    continue;
+                }
+                doRead -= read;
+                if (doRead == 0) {
+                    switch (state) {
+                        case WAIT_FIRST_SYNCBYTE:
+                            if (bytes[pos] == ESP2Packet.ENOCEAN_ESP2_FIRSTSYNC_BYTE) {
+                                state = ReadingState.WAIT_SECOND_SYNCBYTE;
+                                logger.trace("Received first sync byte");
+                            }
+                            doRead = 1;
+                            break;
+                        case WAIT_SECOND_SYNCBYTE:
+                            if (bytes[pos] == ESP2Packet.ENOCEAN_ESP2_SECONDSYNC_BYTE) {
+                                state = ReadingState.READ_HEADER;
+                                logger.trace("Received second sync byte");
+                            } else {
+                                state = ReadingState.WAIT_FIRST_SYNCBYTE;
+                                logger.trace(
+                                        "Received non-matching second sync byte ({}) - first sync byte was a false positive",
+                                        EnOceanUtil.byteToHex(bytes[pos]));
+                            }
+                            doRead = 1;
+                            break;
+                        case READ_HEADER:
+                            doRead = bytes[pos] & 0x1f;
+                            if (doRead == 0) {
+                                state = ReadingState.WAIT_FIRST_SYNCBYTE;
+                                doRead = 1;
+                                logger.debug(">> Received header with zero length, ignoring packet");
+                            } else {
+                                state = ReadingState.READ_DATA;
+                                packetStart = pos;
+                                packetLength = (byte) doRead;
+                                packetType = (byte) ((bytes[pos] & 0xff) >> 5);
+                                logger.trace(">> Received header, data length {} packet type {}", doRead, packetType);
+                            }
+                            break;
+                        case READ_DATA:
+                            try {
+                                packetBytes = EnOceanUtil.subArray(bytes, packetStart, packetLength + 1);
+                                if (ESP2Packet.validateCheckSum(packetBytes, 0, packetLength)) {
+                                    if (packetBytes[1] == ESP2Packet.ENOCEAN_ESP2_INTERNAL_COMMAND_BYTE) {
+                                        // Internal commands have a structure that we can't decode,
+                                        // and they shouldn't be of interest to us
+                                        if (logger.isTraceEnabled()) {
+                                            logger.trace("Skipping internal command ESP2Packet: {}",
+                                                    HexUtils.bytesToHex(packetBytes));
+                                        }
+                                    } else {
+                                        BasePacket packet = ESP2PacketConverter
+                                                .buildPacket(ESP2PacketType.getPacketType(packetType), packetBytes);
+                                        if (packet != null) {
+                                            switch (packet.getPacketType()) {
+                                                case RADIO_ERP1:
+                                                    ERP1Message msg = (ERP1Message) packet;
+                                                    if (logger.isDebugEnabled()) {
+                                                        logger.debug("Converted to: {} with RORG {} for {}",
+                                                                packet.getPacketType().name(), msg.getRORG().name(),
+                                                                HexUtils.bytesToHex(msg.getSenderId()));
+                                                    }
 
-                                            if (msg.getRORG() != RORG.Unknown) {
-                                                informListeners(msg);
-                                            } else {
-                                                logger.debug("Received unknown RORG");
+                                                    if (msg.getRORG() != RORG.Unknown) {
+                                                        informListeners(msg, scheduler);
+                                                    } else {
+                                                        logger.debug("Received unknown RORG");
+                                                    }
+                                                    break;
+                                                case RESPONSE:
+                                                    Response response = (Response) packet;
+                                                    logger.debug("Converted to: {} with code {}",
+                                                            packet.getPacketType().name(),
+                                                            response.getResponseType().name());
+
+                                                    handleResponse(response, scheduler);
+                                                    break;
+                                                default:
+                                                    logger.debug("Not handling packet of type {}",
+                                                            packet.getPacketType());
+                                                    break;
                                             }
+                                        } else if (logger.isDebugEnabled()) {
+                                            logger.debug("Unknown/unsupported ESP2Packet: {}",
+                                                    HexUtils.bytesToHex(packetBytes));
                                         }
-                                            break;
-                                        case RESPONSE: {
-                                            Response response = (Response) packet;
-                                            logger.debug("Converted to: {} with code {}", packet.getPacketType().name(),
-                                                    response.getResponseType().name());
-
-                                            handleResponse(response);
-                                        }
-                                            break;
-                                        default:
-                                            break;
                                     }
                                 } else {
-                                    if (dataBuffer[1] != (byte) 0xFC) {
-                                        byte[] array = Arrays.copyOf(dataBuffer, dataLength);
-                                        String packetString = array != null ? HexUtils.bytesToHex(array) : "";
-                                        logger.debug("Unknown/unsupported ESP2Packet: {}", packetString);
-                                    }
+                                    logger.debug("Malformed ESP2Packet: {}", HexUtils.bytesToHex(packetBytes));
+                                }
+                            } catch (RuntimeException e) {
+                                logger.debug("Unable to process message: {}", e.getMessage());
+                                logger.trace("", e);
+                                TransceiverErrorListener errorListener = this.errorListener;
+                                if (errorListener != null && !thread.isInterrupted()) {
+                                    // We don't want to take the Thing offline if a RuntimeException is thrown while
+                                    // the Receiver is terminating, which is why we check isInterrupted()
+                                    errorListener.errorOccurred(e);
                                 }
-                            } else {
-                                logger.debug("ESP2Packet malformed: {}", HexUtils.bytesToHex(dataBuffer));
                             }
-
-                            state = byteBuffer == ESP2Packet.ENOCEAN_ESP2_FIRSTSYNC_BYTE
-                                    ? ReadingState.WaitingForSecondSyncByte
-                                    : ReadingState.WaitingForFirstSyncByte;
-
-                            currentPosition = 0;
-                            dataLength = packetType = -1;
-                        } else {
-                            dataBuffer[currentPosition++] = byteBuffer;
-                        }
-                        break;
+                            state = ReadingState.WAIT_FIRST_SYNCBYTE;
+                            doRead = 1;
+                            packetStart = -1;
+                            packetLength = -1;
+                            packetType = -1;
+                            break;
+                    }
                 }
+
+                // Not checking for overflow here because it should be impossible, and should it happen
+                // throwing an IndexOutOfBoundsException above is just as good as anything we can do here.
+                pos = state == ReadingState.WAIT_FIRST_SYNCBYTE ? 0 : pos + read;
             }
-        } catch (IOException ioexception) {
-            logger.trace("Unable to process message", ioexception);
-            TransceiverErrorListener localListener = errorListener;
-            if (localListener != null) {
-                localListener.errorOccurred(ioexception);
-            }
-            return;
+
+            logger.info("Shutting down EnOceanSerialTransceiver RX thread ({})", thread.getName());
         }
     }
 

bundles/org.openhab.binding.enocean/src/main/java/org/openhab/binding/enocean/internal/transceiver/EnOceanESP3Transceiver.java

@@ -12,11 +12,11 @@
  */
 package org.openhab.binding.enocean.internal.transceiver;
 
+import java.io.BufferedInputStream;
 import java.io.IOException;
 import java.io.InputStream;
-import java.util.Arrays;
-import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
 import org.eclipse.jdt.annotation.Nullable;
@@ -25,6 +25,7 @@ import org.openhab.binding.enocean.internal.messages.BasePacket;
 import org.openhab.binding.enocean.internal.messages.ESP3Packet;
 import org.openhab.binding.enocean.internal.messages.ESP3PacketFactory;
 import org.openhab.binding.enocean.internal.messages.Response;
+import org.openhab.binding.enocean.internal.util.EnOceanUtil;
 import org.openhab.core.io.transport.serial.SerialPortManager;
 import org.openhab.core.util.HexUtils;
 
@@ -35,167 +36,251 @@ import org.openhab.core.util.HexUtils;
 @NonNullByDefault
 public class EnOceanESP3Transceiver extends EnOceanTransceiver {
 
+    private static final String RX_THREAD_NAME_PREFIX = "OH-binding-enocean-ESP3-RX-";
+    private static final AtomicInteger THREAD_NUM = new AtomicInteger();
+
+    // All access must be guarded by "this"
+    private @Nullable Thread worker;
+
     public EnOceanESP3Transceiver(String path, TransceiverErrorListener errorListener,
             ScheduledExecutorService scheduler, @Nullable SerialPortManager serialPortManager) {
         super(path, errorListener, scheduler, serialPortManager);
     }
 
     enum ReadingState {
-        WaitingForSyncByte,
-        ReadingHeader,
-        ReadingData
+        WAIT_FIRST_SYNCBYTE,
+        READ_HEADER,
+        READ_DATA
     }
 
-    byte[] dataBuffer = new byte[ENOCEAN_MAX_DATA];
-    ReadingState state = ReadingState.WaitingForSyncByte;
-    int currentPosition = 0;
-    int dataLength = -1;
-    int optionalLength = -1;
-    byte packetType = -1;
+    @Override
+    public void startReceiving(ScheduledExecutorService scheduler) {
+        Thread worker;
+        InputStream is;
+        synchronized (this) {
+            worker = this.worker;
+            if (worker != null && worker.isAlive()) {
+                worker.interrupt();
+            }
+
+            is = inputStream;
+            if (is == null) {
+                this.worker = worker = null;
+            } else {
+                this.worker = worker = new Thread(new Receiver(is, errorListener, scheduler),
+                        RX_THREAD_NAME_PREFIX + THREAD_NUM.incrementAndGet());
+                worker.setUncaughtExceptionHandler((t, e) -> {
+                    logger.warn("Uncaught exception in EnOceanSerialTransceiver RX thread ({}): {}", t.getName(),
+                            e.getMessage());
+                    logger.trace("", e);
+                    TransceiverErrorListener listener = this.errorListener;
+                    if (listener != null) {
+                        scheduler.execute(() -> listener.errorOccurred(e));
+                    }
+                });
+            }
+        }
+        if (worker == null) {
+            logger.warn("Cannot read from null stream");
+            TransceiverErrorListener errorListener = this.errorListener;
+            if (errorListener != null) {
+                IOException e = new IOException("Cannot read from null stream");
+                scheduler.execute(() -> errorListener.errorOccurred(e));
+            }
+        } else {
+            worker.start();
+            logger.info("EnOceanSerialTransceiver RX thread ({}) started", worker.getName());
+        }
+    }
 
     @Override
-    protected void processMessage(byte firstByte) {
-        byte[] readingBuffer = new byte[ENOCEAN_MAX_DATA];
-        int bytesRead = -1;
-        byte byteBuffer;
+    protected void shutDownRx() {
+        Thread worker;
+        synchronized (this) {
+            worker = this.worker;
+            this.worker = null;
+        }
+        if (worker != null && worker.isAlive()) {
+            worker.interrupt();
+        }
+    }
 
-        try {
-            readingBuffer[0] = firstByte;
-            InputStream localInPutStream = this.inputStream;
-            if (localInPutStream == null) {
-                throw new IOException("could not read from inputstream");
-            }
-            bytesRead = localInPutStream.read(readingBuffer, 1, localInPutStream.available());
-            if (bytesRead == -1) {
-                throw new IOException("could not read from inputstream");
+    private class Receiver implements Runnable {
+
+        private final InputStream is;
+        private final @Nullable TransceiverErrorListener errorListener;
+        private final ScheduledExecutorService scheduler;
+
+        public Receiver(InputStream is, @Nullable TransceiverErrorListener errorListener,
+                ScheduledExecutorService scheduler) {
+            this.is = is;
+            this.errorListener = errorListener;
+            this.scheduler = scheduler;
+        }
+
+        @Override
+        public void run() {
+            byte[] buffer = new byte[ENOCEAN_MAX_DATA + 7]; // 7 = 1 (sync) + 4 (header) + 2 * 1 (CRC-8)
+            int read;
+            ReadingState state = ReadingState.WAIT_FIRST_SYNCBYTE;
+            int doRead = 1;
+            byte packetType = -1;
+            int length = -1;
+            int optionalLength = -1;
+            int start = -1;
+            int pos = 0;
+            final Thread thread = Thread.currentThread();
+            InputStream is = this.is;
+            logger.trace("RX InputStream implementation: {}", is.getClass().getName());
+            if (!is.markSupported()) {
+                // Use this as a "rough indicator" that the stream isn't buffered
+                logger.trace("Wrapping {} in BufferedInputStream", is.getClass().getName());
+                is = new BufferedInputStream(is, 0x40000);
             }
+            byte[] bytes;
 
-            Future<?> localReadingTask = readingTask;
-            if (localReadingTask == null || localReadingTask.isCancelled()) {
-                return;
-            }
-
-            bytesRead++;
-            for (int p = 0; p < bytesRead; p++) {
-                byteBuffer = readingBuffer[p];
-
-                switch (state) {
-                    case WaitingForSyncByte:
-                        if (byteBuffer == ESP3Packet.ESP3_SYNC_BYTE) {
-                            state = ReadingState.ReadingHeader;
-                            logger.trace("Received Sync Byte");
-                        }
-                        break;
-                    case ReadingHeader:
-                        if (currentPosition == ESP3Packet.ESP3_HEADER_LENGTH) {
-                            if (ESP3Packet.checkCRC8(dataBuffer, ESP3Packet.ESP3_HEADER_LENGTH, byteBuffer)
-                                    && ((dataBuffer[0] & 0xFF) << 8) + (dataBuffer[1] & 0xFF)
-                                            + (dataBuffer[2] & 0xFF) > 0) {
-                                state = ReadingState.ReadingData;
-
-                                dataLength = ((dataBuffer[0] & 0xFF << 8) | (dataBuffer[1] & 0xFF));
-                                optionalLength = dataBuffer[2] & 0xFF;
-                                packetType = dataBuffer[3];
-                                currentPosition = 0;
-
-                                if (packetType == 3) {
-                                    logger.trace("Received sub_msg");
-                                }
-
-                                logger.trace(">> Received header, data length {} optional length {} packet type {}",
-                                        dataLength, optionalLength, packetType);
-                            } else {
-                                // check if we find a sync byte in current buffer
-                                int copyFrom = -1;
-                                for (int i = 0; i < ESP3Packet.ESP3_HEADER_LENGTH; i++) {
-                                    if (dataBuffer[i] == ESP3Packet.ESP3_SYNC_BYTE) {
-                                        copyFrom = i + 1;
-                                        break;
-                                    }
-                                }
-
-                                if (copyFrom != -1) {
-                                    System.arraycopy(dataBuffer, copyFrom, dataBuffer, 0,
-                                            ESP3Packet.ESP3_HEADER_LENGTH - copyFrom);
-                                    state = ReadingState.ReadingHeader;
-                                    currentPosition = ESP3Packet.ESP3_HEADER_LENGTH - copyFrom;
-                                    dataBuffer[currentPosition++] = byteBuffer;
-                                } else {
-                                    currentPosition = 0;
-                                    state = byteBuffer == ESP3Packet.ESP3_SYNC_BYTE ? ReadingState.ReadingHeader
-                                            : ReadingState.WaitingForSyncByte;
-                                }
-                                logger.trace("CrC8 header check not successful");
-                            }
-                        } else {
-                            dataBuffer[currentPosition++] = byteBuffer;
-                        }
-                        break;
-                    case ReadingData:
-                        if (currentPosition == dataLength + optionalLength) {
-                            if (ESP3Packet.checkCRC8(dataBuffer, dataLength + optionalLength, byteBuffer)) {
-                                state = ReadingState.WaitingForSyncByte;
-                                BasePacket packet = ESP3PacketFactory.buildPacket(dataLength, optionalLength,
-                                        packetType, dataBuffer);
-
-                                if (packet != null) {
-                                    switch (packet.getPacketType()) {
-                                        case COMMON_COMMAND:
-                                            logger.debug("Common command: {}",
-                                                    HexUtils.bytesToHex(packet.getPayload()));
-                                            break;
-                                        case EVENT:
-                                        case RADIO_ERP1:
-                                            informListeners(packet);
-                                            break;
-                                        case RADIO_ERP2:
-                                            break;
-                                        case RADIO_MESSAGE:
-                                            break;
-                                        case RADIO_SUB_TEL:
-                                            break;
-                                        case REMOTE_MAN_COMMAND:
-                                            break;
-                                        case RESPONSE: {
-                                            Response response = (Response) packet;
-                                            logger.debug("{} with code {} payload {} received",
-                                                    packet.getPacketType().name(), response.getResponseType().name(),
-                                                    HexUtils.bytesToHex(packet.getPayload())); // Responses do not have
-                                                                                               // optional data
-                                            handleResponse(response);
-                                        }
-                                            break;
-                                        case SMART_ACK_COMMAND:
-                                            break;
-                                        default:
-                                            break;
-                                    }
-                                } else {
-                                    logger.trace("Unknown ESP3Packet: {}", HexUtils
-                                            .bytesToHex(Arrays.copyOf(dataBuffer, dataLength + optionalLength)));
-                                }
-                            } else {
-                                state = byteBuffer == ESP3Packet.ESP3_SYNC_BYTE ? ReadingState.ReadingHeader
-                                        : ReadingState.WaitingForSyncByte;
-                                logger.trace("ESP3Packet malformed: {}",
-                                        HexUtils.bytesToHex(Arrays.copyOf(dataBuffer, dataLength + optionalLength)));
-                            }
-
-                            currentPosition = 0;
-                            dataLength = optionalLength = packetType = -1;
-                        } else {
-                            dataBuffer[currentPosition++] = byteBuffer;
-                        }
-                        break;
+            while (!thread.isInterrupted()) {
+                try {
+                    read = is.read(buffer, pos, doRead);
+                } catch (IOException e) {
+                    logger.debug("Unable to read from serial port: {}", e.getMessage());
+                    logger.trace("", e);
+                    TransceiverErrorListener errorListener = this.errorListener;
+                    if (errorListener != null && !thread.isInterrupted()) {
+                        // We don't want to take the Thing offline if an IOException is thrown when the port is closed,
+                        // which is why we check isInterrupted()
+                        errorListener.errorOccurred(e);
+                    }
+                    break;
                 }
+                if (read <= 0) {
+                    // Unlike regular InputStreams, the serial port streams occasionally returns -1 even if the
+                    // stream is still "alive", so just accept it and try to read again. Add a short backoff to
+                    // avoid a tight loop and high CPU usage if this happens repeatedly.
+                    try {
+                        Thread.sleep(5);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        break;
+                    }
+                    continue;
+                }
+                doRead -= read;
+                if (doRead == 0) {
+                    switch (state) {
+                        case WAIT_FIRST_SYNCBYTE:
+                            if (buffer[pos] == ESP3Packet.ESP3_SYNC_BYTE) {
+                                state = ReadingState.READ_HEADER;
+                                doRead = ESP3Packet.ESP3_HEADER_LENGTH + 1;
+                                start = pos + 1;
+                                is.mark(5);
+                                logger.trace("Received sync byte");
+                            } else {
+                                doRead = 1;
+                            }
+                            break;
+                        case READ_HEADER:
+                            bytes = EnOceanUtil.subArray(buffer, start, ESP3Packet.ESP3_HEADER_LENGTH + 1);
+                            length = -1;
+                            if (ESP3Packet.checkCRC8(bytes, 0, ESP3Packet.ESP3_HEADER_LENGTH)
+                                    && (length = ((bytes[0] & 0xff) << 8) | (bytes[1] & 0xff)) > 0) {
+                                state = ReadingState.READ_DATA;
+                                optionalLength = bytes[2] & 0xff;
+                                packetType = bytes[3];
+                                doRead = length + optionalLength + 1;
+                                start = pos + read;
+                                logger.trace(">> Received header, data length {} optional length {} packet type {}",
+                                        length, optionalLength, packetType);
+                                if (packetType == 3) {
+                                    logger.trace(">> Received sub_msg");
+                                }
+                            } else {
+                                state = ReadingState.WAIT_FIRST_SYNCBYTE;
+                                doRead = 1;
+                                if (length == 0) {
+                                    logger.debug("Received header with zero length, assuming false sync byte");
+                                } else {
+                                    logger.debug("Received header with invalid CRC-8, assuming false sync byte");
+                                }
+                                start = -1;
+                                length = -1;
+                                try {
+                                    is.reset();
+                                } catch (IOException e) {
+                                    logger.debug(
+                                            "Failed to rewind inputstream after failed header, one packet might be lost: {}",
+                                            e.getMessage());
+                                }
+                            }
+                            break;
+                        case READ_DATA:
+                            try {
+                                bytes = EnOceanUtil.subArray(buffer, start, length + optionalLength + 1);
+                                if (ESP3Packet.checkCRC8(bytes, 0, length + optionalLength)) {
+                                    BasePacket packet = ESP3PacketFactory.buildPacket(length, optionalLength,
+                                            packetType, bytes);
+
+                                    if (packet != null) {
+                                        switch (packet.getPacketType()) {
+                                            case COMMON_COMMAND:
+                                                logger.debug("Common command: {}",
+                                                        HexUtils.bytesToHex(packet.getPayload()));
+                                                break;
+                                            case EVENT:
+                                            case RADIO_ERP1:
+                                                informListeners(packet, scheduler);
+                                                break;
+                                            case RESPONSE:
+                                                Response response = (Response) packet;
+                                                if (logger.isDebugEnabled()) {
+                                                    // Responses do not have optional data
+                                                    logger.debug("{} with code {} payload {} received",
+                                                            packet.getPacketType().name(),
+                                                            response.getResponseType().name(),
+                                                            HexUtils.bytesToHex(packet.getPayload()));
+                                                }
+                                                handleResponse(response, scheduler);
+                                                break;
+                                            case RADIO_ERP2:
+                                            case RADIO_MESSAGE:
+                                            case RADIO_SUB_TEL:
+                                            case REMOTE_MAN_COMMAND:
+                                            case SMART_ACK_COMMAND:
+                                            default:
+                                                break;
+                                        }
+                                    } else {
+                                        logger.debug("Unknown ESP3Packet: {}", HexUtils.bytesToHex(bytes));
+                                    }
+                                } else {
+                                    logger.debug("Malformed ESP3Packet: {}", HexUtils.bytesToHex(bytes));
+                                }
+                            } catch (RuntimeException e) {
+                                logger.debug("Unable to process message: {}", e.getMessage());
+                                logger.trace("", e);
+                                TransceiverErrorListener errorListener = this.errorListener;
+                                if (errorListener != null && !thread.isInterrupted()) {
+                                    // We don't want to take the Thing offline if a RuntimeException is thrown while
+                                    // the Receiver is terminating, which is why we check isInterrupted()
+                                    errorListener.errorOccurred(e);
+                                }
+                            }
+                            state = ReadingState.WAIT_FIRST_SYNCBYTE;
+                            doRead = 1;
+                            start = -1;
+                            length = -1;
+                            optionalLength = -1;
+                            packetType = -1;
+                            break;
+                    }
+                }
+
+                // Not checking for overflow here because it should be impossible, and should it happen
+                // throwing an IndexOutOfBoundsException above is just as good as anything we can do here.
+                pos = state == ReadingState.WAIT_FIRST_SYNCBYTE ? 0 : pos + read;
             }
-        } catch (IOException ioexception) {
-            logger.trace("Unable to process message", ioexception);
-            TransceiverErrorListener localListener = errorListener;
-            if (localListener != null) {
-                localListener.errorOccurred(ioexception);
-            }
-            return;
+
+            logger.info("Shutting down EnOceanSerialTransceiver RX thread ({})", thread.getName());
         }
     }
 

bundles/org.openhab.binding.enocean/src/main/java/org/openhab/binding/enocean/internal/transceiver/EnOceanTransceiver.java

@@ -19,6 +19,7 @@ import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Queue;
+import java.util.Set;
 import java.util.TooManyListenersException;
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -58,7 +59,6 @@ public abstract class EnOceanTransceiver implements SerialPortEventListener {
     public static final int ENOCEAN_MAX_DATA = 65790;
 
     // Thread management
-    protected @Nullable Future<?> readingTask = null;
     private @Nullable Future<?> timeOutTask = null;
 
     protected Logger logger = LoggerFactory.getLogger(EnOceanTransceiver.class);
@@ -74,12 +74,12 @@ public abstract class EnOceanTransceiver implements SerialPortEventListener {
         @Nullable
         Response responsePacket;
         @Nullable
-        ResponseListener<? extends @Nullable Response> responseListener;
+        ResponseListener<? extends Response> responseListener;
     }
 
     private class RequestQueue {
-        private Queue<Request> queue = new LinkedBlockingQueue<>();
-        private ScheduledExecutorService scheduler;
+        private final Queue<Request> queue = new LinkedBlockingQueue<>();
+        private final ScheduledExecutorService scheduler;
 
         public RequestQueue(ScheduledExecutorService scheduler) {
             this.scheduler = scheduler;
@@ -220,19 +220,7 @@ public abstract class EnOceanTransceiver implements SerialPortEventListener {
         logger.info("EnOceanSerialTransceiver initialized");
     }
 
-    public void startReceiving(ScheduledExecutorService scheduler) {
-        @Nullable
-        Future<?> readingTask = this.readingTask;
-        if (readingTask == null || readingTask.isCancelled()) {
-            this.readingTask = scheduler.submit(new Runnable() {
-                @Override
-                public void run() {
-                    receivePackets();
-                }
-            });
-        }
-        logger.info("EnOceanSerialTransceiver RX thread started");
-    }
+    public abstract void startReceiving(ScheduledExecutorService scheduler);
 
     public void shutDown() {
         logger.debug("shutting down transceiver");
@@ -244,20 +232,7 @@ public abstract class EnOceanTransceiver implements SerialPortEventListener {
             this.timeOutTask = null;
         }
 
-        Future<?> readingTask = this.readingTask;
-        if (readingTask != null) {
-            readingTask.cancel(true);
-
-            InputStream localInputStream = inputStream;
-            if (localInputStream != null) {
-                try {
-                    localInputStream.close();
-                } catch (IOException e) {
-                    logger.debug("IOException occurred while closing the stream", e);
-                }
-            }
-            this.readingTask = null;
-        }
+        shutDownRx();
 
         listeners.clear();
         eventListeners.clear();
@@ -295,45 +270,9 @@ public abstract class EnOceanTransceiver implements SerialPortEventListener {
         logger.info("Transceiver shutdown");
     }
 
-    private void receivePackets() {
-        byte[] buffer = new byte[1];
+    protected abstract void shutDownRx();
 
-        Future<?> readingTask = this.readingTask;
-        while (readingTask != null && !readingTask.isCancelled()) {
-            int bytesRead = read(buffer, 1);
-            if (bytesRead > 0) {
-                processMessage(buffer[0]);
-            }
-        }
-    }
-
-    protected abstract void processMessage(byte firstByte);
-
-    protected int read(byte[] buffer, int length) {
-        InputStream localInputStream = inputStream;
-        if (localInputStream != null) {
-            try {
-                return localInputStream.read(buffer, 0, length);
-            } catch (IOException e) {
-                logger.debug("IOException occurred while reading the input stream", e);
-                return 0;
-            }
-        } else {
-            logger.warn("Cannot read from null stream");
-            Future<?> readingTask = this.readingTask;
-            if (readingTask != null) {
-                readingTask.cancel(true);
-                this.readingTask = null;
-            }
-            TransceiverErrorListener localListener = errorListener;
-            if (localListener != null) {
-                localListener.errorOccurred(new IOException("Cannot read from null stream"));
-            }
-            return 0;
-        }
-    }
-
-    protected void informListeners(BasePacket packet) {
+    protected void informListeners(BasePacket packet, @Nullable ScheduledExecutorService scheduler) {
         try {
             if (packet.getPacketType() == ESPPacketType.RADIO_ERP1) {
                 ERP1Message msg = (ERP1Message) packet;
@@ -356,7 +295,11 @@ public abstract class EnOceanTransceiver implements SerialPortEventListener {
 
                             TeachInListener localListener = teachInListener;
                             if (localListener != null) {
-                                localListener.packetReceived(msg);
+                                if (scheduler != null) {
+                                    scheduler.execute(() -> localListener.packetReceived(msg));
+                                } else {
+                                    localListener.packetReceived(msg);
+                                }
                             }
                             return;
                         } else if (teachInListener == null && msg.getIsTeachIn()) {
@@ -369,7 +312,12 @@ public abstract class EnOceanTransceiver implements SerialPortEventListener {
                         synchronized (this) {
                             HashSet<PacketListener> pl = listeners.get(s);
                             if (pl != null) {
-                                pl.forEach(l -> l.packetReceived(msg));
+                                if (scheduler != null) {
+                                    Set<PacketListener> plc = Set.copyOf(pl);
+                                    scheduler.execute(() -> plc.forEach(l -> l.packetReceived(msg)));
+                                } else {
+                                    pl.forEach(l -> l.packetReceived(msg));
+                                }
                             }
                         }
                     }
@@ -390,7 +338,11 @@ public abstract class EnOceanTransceiver implements SerialPortEventListener {
                         logger.info("Received smart teach in from {}", HexUtils.bytesToHex(senderId));
                         TeachInListener localListener = teachInListener;
                         if (localListener != null) {
-                            localListener.eventReceived(event);
+                            if (scheduler != null) {
+                                scheduler.execute(() -> localListener.eventReceived(event));
+                            } else {
+                                localListener.eventReceived(event);
+                            }
                         }
                         return;
                     } else {
@@ -401,7 +353,14 @@ public abstract class EnOceanTransceiver implements SerialPortEventListener {
                 }
 
                 synchronized (this) {
-                    eventListeners.forEach(l -> l.eventReceived(event));
+                    if (!eventListeners.isEmpty()) {
+                        if (scheduler != null) {
+                            final Set<EventListener> eventListeners = Set.copyOf(this.eventListeners);
+                            scheduler.execute(() -> eventListeners.forEach(l -> l.eventReceived(event)));
+                        } else {
+                            eventListeners.forEach(l -> l.eventReceived(event));
+                        }
+                    }
                 }
             }
         } catch (Exception e) {
@@ -409,18 +368,30 @@ public abstract class EnOceanTransceiver implements SerialPortEventListener {
         }
     }
 
-    protected void handleResponse(Response response) throws IOException {
+    protected void handleResponse(Response response, @Nullable ScheduledExecutorService scheduler) {
         Request localCurrentRequest = currentRequest;
         if (localCurrentRequest != null) {
-            ResponseListener<? extends @Nullable Response> listener = localCurrentRequest.responseListener;
+            ResponseListener<? extends Response> listener = localCurrentRequest.responseListener;
+            localCurrentRequest.responsePacket = response;
             if (listener != null) {
-                localCurrentRequest.responsePacket = response;
-                try {
-                    listener.handleResponse(response);
-                } catch (Exception e) {
-                    logger.debug("Exception during response handling");
-                } finally {
-                    logger.trace("Response handled");
+                if (scheduler != null) {
+                    scheduler.execute(() -> {
+                        try {
+                            listener.handleResponse(response);
+                            logger.trace("Response handled");
+                        } catch (Exception e) {
+                            logger.debug("Exception during response handling: {}", e.getMessage());
+                            logger.trace("", e);
+                        }
+                    });
+                } else {
+                    try {
+                        listener.handleResponse(response);
+                        logger.trace("Response handled");
+                    } catch (Exception e) {
+                        logger.debug("Exception during response handling: {}", e.getMessage());
+                        logger.trace("", e);
+                    }
                 }
             } else {
                 logger.trace("Response without listener");
@@ -431,7 +402,7 @@ public abstract class EnOceanTransceiver implements SerialPortEventListener {
     }
 
     public void sendBasePacket(@Nullable BasePacket packet,
-            @Nullable ResponseListener<? extends @Nullable Response> responseCallback) throws IOException {
+            @Nullable ResponseListener<? extends Response> responseCallback) throws IOException {
         if (packet == null) {
             return;
         }

bundles/org.openhab.binding.enocean/src/main/java/org/openhab/binding/enocean/internal/transceiver/ResponseListener.java

@@ -15,7 +15,6 @@ package org.openhab.binding.enocean.internal.transceiver;
 import java.lang.reflect.ParameterizedType;
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
-import org.eclipse.jdt.annotation.Nullable;
 import org.openhab.binding.enocean.internal.EnOceanException;
 import org.openhab.binding.enocean.internal.messages.Response;
 
@@ -24,7 +23,7 @@ import org.openhab.binding.enocean.internal.messages.Response;
  * @author Daniel Weber - Initial contribution
  */
 @NonNullByDefault
-public abstract class ResponseListener<T extends @Nullable Response> {
+public abstract class ResponseListener<T extends Response> {
 
     protected Class<T> persistentClass;
 

bundles/org.openhab.binding.enocean/src/main/java/org/openhab/binding/enocean/internal/transceiver/ResponseListenerIgnoringTimeouts.java

@@ -15,7 +15,6 @@ package org.openhab.binding.enocean.internal.transceiver;
 import java.lang.reflect.ParameterizedType;
 
 import org.eclipse.jdt.annotation.NonNullByDefault;
-import org.eclipse.jdt.annotation.Nullable;
 import org.openhab.binding.enocean.internal.messages.Response;
 
 /**
@@ -23,7 +22,7 @@ import org.openhab.binding.enocean.internal.messages.Response;
  * @author Daniel Weber - Initial contribution
  */
 @NonNullByDefault
-public abstract class ResponseListenerIgnoringTimeouts<T extends @Nullable Response> extends ResponseListener<T> {
+public abstract class ResponseListenerIgnoringTimeouts<T extends Response> extends ResponseListener<T> {
 
     @SuppressWarnings({ "unchecked", "null" })
     public ResponseListenerIgnoringTimeouts() {

bundles/org.openhab.binding.enocean/src/main/java/org/openhab/binding/enocean/internal/util/EnOceanUtil.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright (c) 2010-2026 Contributors to the openHAB project
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.openhab.binding.enocean.internal.util;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A general utility class.
+ *
+ * @author Ravi Nadahar - Initial contribution
+ */
+public class EnOceanUtil {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(EnOceanUtil.class);
+    private static final char[] HEX_ARRAY = "0123456789ABCDEF".toCharArray();
+
+    private EnOceanUtil() {
+        // Not to be instantiated
+    }
+
+    public static String byteToHex(byte b) {
+        int i = b & 0xFF;
+        return new StringBuilder(4).append("0x").append(HEX_ARRAY[i >>> 4]).append(HEX_ARRAY[i & 0x0F]).toString();
+    }
+
+    public static byte[] subArray(byte[] bytes, int start, int length) {
+        if (start == 0 && length == bytes.length) {
+            return bytes;
+        }
+        int len = Math.min(length, bytes.length - start);
+        if (len < length) {
+            LOGGER.debug("Packet is shorter ({}) than expected: {}", len, length);
+        }
+        byte[] result = new byte[len];
+        System.arraycopy(bytes, start, result, 0, len);
+        return result;
+    }
+}

bundles/org.openhab.binding.enocean/src/test/java/org/openhab/binding/enocean/internal/transceiver/EnOceanESP2TransceiverTest.java

@@ -0,0 +1,253 @@
+/*
+ * Copyright (c) 2010-2026 Contributors to the openHAB project
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.openhab.binding.enocean.internal.transceiver;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+import static org.mockito.Mockito.*;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Objects;
+
+import org.eclipse.jdt.annotation.NonNullByDefault;
+import org.eclipse.jdt.annotation.Nullable;
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import org.openhab.binding.enocean.internal.messages.BasePacket;
+import org.openhab.binding.enocean.internal.messages.ERP1Message;
+import org.openhab.binding.enocean.internal.messages.ESP2Packet.ESP2PacketType;
+import org.openhab.binding.enocean.internal.messages.ESP2PacketConverter;
+import org.openhab.core.io.transport.serial.PortInUseException;
+import org.openhab.core.io.transport.serial.SerialPort;
+import org.openhab.core.io.transport.serial.SerialPortIdentifier;
+import org.openhab.core.io.transport.serial.SerialPortManager;
+import org.openhab.core.util.SameThreadExecutorService;
+
+/**
+ * Tests for {@link EnOceanESP2Transceiver}.
+ *
+ * @author Ravi Nadahar - Initial contribution
+ */
+@NonNullByDefault
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public class EnOceanESP2TransceiverTest {
+
+    private @Mock @NonNullByDefault({}) TransceiverErrorListener errorListener;
+    private @Mock @NonNullByDefault({}) SerialPortManager portManager;
+    private @Mock @NonNullByDefault({}) SerialPort port;
+    private SameThreadExecutorService scheduler = new SameThreadExecutorService();
+
+    @BeforeEach
+    void setUp() {
+        when(portManager.getIdentifier("")).thenReturn(new SerialPortIdentifier() {
+
+            @Override
+            public SerialPort open(String owner, int timeout) throws PortInUseException {
+                return port;
+            }
+
+            @Override
+            public boolean isCurrentlyOwned() {
+                return true;
+            }
+
+            @Override
+            public String getName() {
+                return "comx";
+            }
+
+            @Override
+            public @Nullable String getCurrentOwner() {
+                return "OH";
+            }
+        });
+    }
+
+    @Test
+    public void testReceiver1() throws Exception {
+        String hexBytes = new String(
+                Objects.requireNonNull(getClass().getResourceAsStream("/ESP2Stream1.txt")).readAllBytes(),
+                StandardCharsets.UTF_8);
+        ByteArrayInputStream bis = new ByteArrayInputStream(HexFormat.of().parseHex(hexBytes));
+        when(port.getInputStream()).thenReturn(bis);
+
+        CapturingPacketListener sender2Listener = new CapturingPacketListener(2L);
+        CapturingPacketListener sender5Listener = new CapturingPacketListener(5L);
+        EnOceanESP2Transceiver trans = new EnOceanESP2Transceiver("", errorListener, scheduler, portManager);
+        trans.addPacketListener(sender2Listener, 2L);
+        trans.addPacketListener(sender5Listener, 5L);
+        trans.initialize();
+        trans.startReceiving(scheduler);
+        try {
+            assertThat(sender2Listener.packets, waitUntil(hasSize(31), 10000L));
+            assertThat(sender5Listener.packets, waitUntil(hasSize(31), 10000L));
+        } finally {
+            trans.shutDownRx();
+        }
+        assertThat(sender2Listener.packets,
+                hasItem(equalPacket(
+                        Objects.requireNonNull(ESP2PacketConverter.buildPacket(ESP2PacketType.RECEIVE_MESSAGE_TELEGRAM,
+                                HexFormat.of().parseHex("8B0550000000000000023012"))))));
+        assertThat(sender5Listener.packets,
+                hasItem(equalPacket(
+                        Objects.requireNonNull(ESP2PacketConverter.buildPacket(ESP2PacketType.RECEIVE_MESSAGE_TELEGRAM,
+                                HexFormat.of().parseHex("8B0550000000000000053015"))))));
+    }
+
+    @Test
+    public void testReceiver2() throws Exception {
+        String hexBytes = new String(
+                Objects.requireNonNull(getClass().getResourceAsStream("/ESP2Stream2.txt")).readAllBytes(),
+                StandardCharsets.UTF_8);
+        ByteArrayInputStream bis = new ByteArrayInputStream(HexFormat.of().parseHex(hexBytes));
+        when(port.getInputStream()).thenReturn(bis);
+
+        CapturingPacketListener sender1Listener = new CapturingPacketListener(0x33221118L);
+        CapturingPacketListener sender2Listener = new CapturingPacketListener(0x33221320L);
+        EnOceanESP2Transceiver trans = new EnOceanESP2Transceiver("", errorListener, scheduler, portManager);
+        trans.addPacketListener(sender1Listener, 0x33221118L);
+        trans.addPacketListener(sender2Listener, 0x33221320L);
+        trans.initialize();
+        trans.startReceiving(scheduler);
+        try {
+            assertThat(sender1Listener.packets, waitUntil(hasSize(6), 10000L));
+            assertThat(sender2Listener.packets, waitUntil(hasSize(3), 10000L));
+        } finally {
+            trans.shutDownRx();
+        }
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP2PacketConverter.buildPacket(
+                ESP2PacketType.RECEIVE_RADIO_TELEGRAM, HexFormat.of().parseHex("0B05000000003322111820AE"))))));
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP2PacketConverter.buildPacket(
+                ESP2PacketType.RECEIVE_RADIO_TELEGRAM, HexFormat.of().parseHex("0B05100000003322111830CE"))))));
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP2PacketConverter.buildPacket(
+                ESP2PacketType.RECEIVE_RADIO_TELEGRAM, HexFormat.of().parseHex("0B05300000003322111830EE"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP2PacketConverter.buildPacket(
+                ESP2PacketType.RECEIVE_RADIO_TELEGRAM, HexFormat.of().parseHex("0B0700A0800F3322132000C9"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP2PacketConverter.buildPacket(
+                ESP2PacketType.RECEIVE_RADIO_TELEGRAM, HexFormat.of().parseHex("0B07009D800F3322132000C6"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP2PacketConverter.buildPacket(
+                ESP2PacketType.RECEIVE_RADIO_TELEGRAM, HexFormat.of().parseHex("0B07009B800F3322132000C4"))))));
+    }
+
+    public static <T> Matcher<T> waitUntil(Matcher<T> matcher, long timeoutMs) {
+        return new WaitUntil<>(matcher, timeoutMs);
+    }
+
+    public static <T extends BasePacket> Matcher<T> equalPacket(BasePacket expected) {
+        return new EqualPacket<>(expected);
+    }
+
+    public static class EqualPacket<T extends BasePacket> extends BaseMatcher<T> {
+
+        private final BasePacket expected;
+
+        public EqualPacket(BasePacket expected) {
+            this.expected = expected;
+        }
+
+        @Override
+        public boolean matches(@Nullable Object actual) {
+            if (actual instanceof ERP1Message message && expected instanceof ERP1Message other) {
+                return message.getIsTeachIn() == other.getIsTeachIn()
+                        && message.getPacketType() == other.getPacketType() && message.getRORG() == other.getRORG()
+                        && Arrays.equals(message.getSenderId(), other.getSenderId())
+                        && Arrays.equals(message.getPayload(), other.getPayload())
+                        && Arrays.equals(message.getOptionalPayload(), other.getOptionalPayload());
+            }
+            return false;
+        }
+
+        @Override
+        public void describeTo(@Nullable Description description) {
+            if (description != null) {
+                description.appendText("equalPacket(").appendValue(expected).appendText(")");
+            }
+        }
+    }
+
+    public static class WaitUntil<T> extends BaseMatcher<T> {
+
+        private final Matcher<T> matcher;
+        private final long timeoutMs;
+
+        public WaitUntil(Matcher<T> matcher, long timeoutMs) {
+            this.matcher = matcher;
+            this.timeoutMs = timeoutMs;
+        }
+
+        @Override
+        public boolean matches(@Nullable Object actual) {
+            long before = System.currentTimeMillis() + timeoutMs;
+            while (System.currentTimeMillis() < before) {
+                if (matcher.matches(actual)) {
+                    return true;
+                } else {
+                    try {
+                        Thread.sleep(100);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        return false;
+                    }
+                }
+            }
+            return false;
+        }
+
+        @Override
+        public void describeTo(@Nullable Description description) {
+            if (description != null) {
+                description.appendText("waitUntil ").appendDescriptionOf(matcher);
+            }
+        }
+
+        @Override
+        public void describeMismatch(@Nullable Object item, @Nullable Description mismatchDescription) {
+            matcher.describeMismatch(item, mismatchDescription);
+        }
+    }
+
+    private static class CapturingPacketListener implements PacketListener {
+
+        private final List<BasePacket> packets = Collections.synchronizedList(new ArrayList<>());
+        private final long senderId;
+
+        public CapturingPacketListener(long senderId) {
+            this.senderId = senderId;
+        }
+
+        @Override
+        public void packetReceived(BasePacket packet) {
+            packets.add(packet);
+        }
+
+        @Override
+        public long getEnOceanIdToListenTo() {
+            return senderId;
+        }
+    }
+}

bundles/org.openhab.binding.enocean/src/test/java/org/openhab/binding/enocean/internal/transceiver/EnOceanESP3TransceiverTest.java

@@ -0,0 +1,274 @@
+/*
+ * Copyright (c) 2010-2026 Contributors to the openHAB project
+ *
+ * See the NOTICE file(s) distributed with this work for additional
+ * information.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-2.0
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ */
+package org.openhab.binding.enocean.internal.transceiver;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.*;
+import static org.mockito.Mockito.*;
+
+import java.io.ByteArrayInputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Objects;
+
+import org.eclipse.jdt.annotation.NonNullByDefault;
+import org.eclipse.jdt.annotation.Nullable;
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import org.openhab.binding.enocean.internal.messages.BasePacket;
+import org.openhab.binding.enocean.internal.messages.ERP1Message;
+import org.openhab.binding.enocean.internal.messages.ESP3PacketFactory;
+import org.openhab.core.io.transport.serial.PortInUseException;
+import org.openhab.core.io.transport.serial.SerialPort;
+import org.openhab.core.io.transport.serial.SerialPortIdentifier;
+import org.openhab.core.io.transport.serial.SerialPortManager;
+import org.openhab.core.util.SameThreadExecutorService;
+
+/**
+ * Tests for {@link EnOceanESP3Transceiver}.
+ *
+ * @author Ravi Nadahar - Initial contribution
+ */
+@NonNullByDefault
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public class EnOceanESP3TransceiverTest {
+
+    private @Mock @NonNullByDefault({}) TransceiverErrorListener errorListener;
+    private @Mock @NonNullByDefault({}) SerialPortManager portManager;
+    private @Mock @NonNullByDefault({}) SerialPort port;
+    private SameThreadExecutorService scheduler = new SameThreadExecutorService();
+
+    @BeforeEach
+    void setUp() {
+        when(portManager.getIdentifier("")).thenReturn(new SerialPortIdentifier() {
+
+            @Override
+            public SerialPort open(String owner, int timeout) throws PortInUseException {
+                return port;
+            }
+
+            @Override
+            public boolean isCurrentlyOwned() {
+                return true;
+            }
+
+            @Override
+            public String getName() {
+                return "comx";
+            }
+
+            @Override
+            public @Nullable String getCurrentOwner() {
+                return "OH";
+            }
+        });
+    }
+
+    @Test
+    public void testReceiver1() throws Exception {
+        String hexBytes = new String(
+                Objects.requireNonNull(getClass().getResourceAsStream("/ESP3Stream1.txt")).readAllBytes(),
+                StandardCharsets.UTF_8);
+        ByteArrayInputStream bis = new ByteArrayInputStream(HexFormat.of().parseHex(hexBytes));
+        when(port.getInputStream()).thenReturn(bis);
+
+        CapturingPacketListener sender1Listener = new CapturingPacketListener(0x058DF435L);
+        CapturingPacketListener sender2Listener = new CapturingPacketListener(0x0582D29DL);
+        EnOceanESP3Transceiver trans = new EnOceanESP3Transceiver("", errorListener, scheduler, portManager);
+        trans.addPacketListener(sender1Listener, 0x058DF435L);
+        trans.addPacketListener(sender2Listener, 0x0582D29DL);
+        trans.initialize();
+        trans.startReceiving(scheduler);
+        try {
+            assertThat(sender1Listener.packets, waitUntil(hasSize(4), 10000L));
+            assertThat(sender2Listener.packets, waitUntil(hasSize(4), 10000L));
+        } finally {
+            trans.shutDownRx();
+        }
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(7,
+                7, (byte) 1, HexFormat.of().parseHex("F6E0058DF4352001FFFFFFFF5B00"))))));
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(7,
+                7, (byte) 1, HexFormat.of().parseHex("F6D0058DF4352001FFFFFFFF5900"))))));
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(7,
+                7, (byte) 1, HexFormat.of().parseHex("F6C0058DF4352001FFFFFFFF5B00"))))));
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(7,
+                7, (byte) 1, HexFormat.of().parseHex("F6F0058DF4352101FFFFFFFF5C00"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A59E16000F0582D29D0001FFFFFFFF5500"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A59E14000F0582D29D0001FFFFFFFF5500"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A59E13000F0582D29D0001FFFFFFFF5500"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A59E11000F0582D29D0001FFFFFFFF5500"))))));
+    }
+
+    @Test
+    public void testReceiver2() throws Exception {
+        String hexBytes = new String(
+                Objects.requireNonNull(getClass().getResourceAsStream("/ESP3Stream2.txt")).readAllBytes(),
+                StandardCharsets.UTF_8);
+        ByteArrayInputStream bis = new ByteArrayInputStream(HexFormat.of().parseHex(hexBytes));
+        when(port.getInputStream()).thenReturn(bis);
+
+        CapturingPacketListener sender1Listener = new CapturingPacketListener(0x0582D29DL);
+        CapturingPacketListener sender2Listener = new CapturingPacketListener(0x05194A1DL);
+        EnOceanESP3Transceiver trans = new EnOceanESP3Transceiver("", errorListener, scheduler, portManager);
+        trans.addPacketListener(sender1Listener, 0x0582D29DL);
+        trans.addPacketListener(sender2Listener, 0x05194A1DL);
+        trans.initialize();
+        trans.startReceiving(scheduler);
+        try {
+            assertThat(sender1Listener.packets, waitUntil(hasSize(35), 10000L));
+            assertThat(sender2Listener.packets, waitUntil(hasSize(11), 10000L));
+        } finally {
+            trans.shutDownRx();
+        }
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A59F27000F0582D29D0001FFFFFFFF5C00"))))));
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A59F27000F0582D29D0001FFFFFFFF5C00"))))));
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A59F27000F0582D29D0001FFFFFFFF5C00"))))));
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A5A029000F0582D29D0001FFFFFFFF5800"))))));
+        assertThat(sender1Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A5A02B000F0582D29D0001FFFFFFFF5800"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A56D01C30905194A1D0001FFFFFFFF5300"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A56901E20905194A1D0001FFFFFFFF5800"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A56901EB0905194A1D0001FFFFFFFF5600"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A56901F50905194A1D0001FFFFFFFF5600"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A56901FE0905194A1D0001FFFFFFFF5600"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A56902060905194A1D0001FFFFFFFF5900"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(10,
+                7, (byte) 1, HexFormat.of().parseHex("A569020D0905194A1D0001FFFFFFFF5C00"))))));
+        assertThat(sender2Listener.packets, hasItem(equalPacket(Objects.requireNonNull(ESP3PacketFactory.buildPacket(8,
+                7, (byte) 1, HexFormat.of().parseHex("D0066405194A1D0001FFFFFFFF5C00"))))));
+    }
+
+    public static <T> Matcher<T> waitUntil(Matcher<T> matcher, long timeoutMs) {
+        return new WaitUntil<>(matcher, timeoutMs);
+    }
+
+    public static <T extends BasePacket> Matcher<T> equalPacket(BasePacket expected) {
+        return new EqualPacket<>(expected);
+    }
+
+    public static class EqualPacket<T extends BasePacket> extends BaseMatcher<T> {
+
+        private final BasePacket expected;
+
+        public EqualPacket(BasePacket expected) {
+            this.expected = expected;
+        }
+
+        @Override
+        public boolean matches(@Nullable Object actual) {
+            if (actual instanceof ERP1Message message && expected instanceof ERP1Message other) {
+                return message.getIsTeachIn() == other.getIsTeachIn()
+                        && message.getPacketType() == other.getPacketType() && message.getRORG() == other.getRORG()
+                        && Arrays.equals(message.getSenderId(), other.getSenderId())
+                        && Arrays.equals(message.getPayload(), other.getPayload())
+                        && Arrays.equals(message.getOptionalPayload(), other.getOptionalPayload());
+            }
+            return false;
+        }
+
+        @Override
+        public void describeTo(@Nullable Description description) {
+            if (description != null) {
+                description.appendText("equalPacket(").appendValue(expected).appendText(")");
+            }
+        }
+    }
+
+    public static class WaitUntil<T> extends BaseMatcher<T> {
+
+        private final Matcher<T> matcher;
+        private final long timeoutMs;
+
+        public WaitUntil(Matcher<T> matcher, long timeoutMs) {
+            this.matcher = matcher;
+            this.timeoutMs = timeoutMs;
+        }
+
+        @Override
+        public boolean matches(@Nullable Object actual) {
+            long before = System.currentTimeMillis() + timeoutMs;
+            while (System.currentTimeMillis() < before) {
+                if (matcher.matches(actual)) {
+                    return true;
+                } else {
+                    try {
+                        Thread.sleep(100);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                        return false;
+                    }
+                }
+            }
+            return false;
+        }
+
+        @Override
+        public void describeTo(@Nullable Description description) {
+            if (description != null) {
+                description.appendText("waitUntil ").appendDescriptionOf(matcher);
+            }
+        }
+
+        @Override
+        public void describeMismatch(@Nullable Object item, @Nullable Description mismatchDescription) {
+            matcher.describeMismatch(item, mismatchDescription);
+        }
+    }
+
+    private static class CapturingPacketListener implements PacketListener {
+
+        private final List<BasePacket> packets = Collections.synchronizedList(new ArrayList<>());
+        private final long senderId;
+
+        public CapturingPacketListener(long senderId) {
+            this.senderId = senderId;
+        }
+
+        @Override
+        public void packetReceived(BasePacket packet) {
+            packets.add(packet);
+        }
+
+        @Override
+        public long getEnOceanIdToListenTo() {
+            return senderId;
+        }
+    }
+}

bundles/org.openhab.binding.enocean/src/test/resources/ESP2Stream2.txt

@@ -0,0 +1 @@
+A55A0B070091840F3322132100BFA55A0B070011000F3322142300BEA55A0B07000C000F3322142600BCA55A0B07007A840F33221317009EA55A0B07008E7D0F3322131A00AEA55A0B07000A000F3322141B00AFA55A0B07008F710F33221216009EA55A8B07008F710F000000343005A55A0B0700A07F0F3322132900D1A55A0B07008C820F3322131F00B6A55A0B07007F830F3322132500B0A55A0B07008E810F3322131B00B3A55A0B0700897E0F3322132600B6A55A0B0700967C0F3322132800C3A55A0B07007D840F3322132700B1A55A0B07000F000F3322142600BFA55A8B05700000000000002D305DA55A0B0700A0800F3322132000C9A55A0B07007F750F3322121F009BA55A8B07007F750F0000003730FCA55A8B05500000000000002D303DA55A0B070091840F3322132100BFA55A0B070013000F3322142600C3A55A0B07007A840F33221317009EA55A0B07008E7D0F3322131A00AEA55A0B0700A27F0F3322132900D3A55A0B07008E820F3322131F00B8A55A0B05300000003322101830EDA55A0B05000000003322101820ADA55A8B05700000000000001B304BA55A0B05100000003322101830CDA55A0B05000000003322101820ADA55A0B05100000003322111830CEA55A0B05000000003322111820AEA55A8B05500000000000001A302AA55A8B05500000000000001B302BA55A0B05300000003322111830EEA55A0B05000000003322111820AEA55A0B07007F830F3322132500B0A55A0B07008E810F3322131B00B3A55A8B05700000000000001A304AA55A0B05100000003322111830CEA55A0B05000000003322111820AEA55A8B05500000000000001A302AA55A0B0700897E0F3322132600B6A55A0B0700967C0F3322132800C3A55A0B07007D840F3322132700B1A55A0B070017000F3322142600C7A55A0B07009D800F3322132000C6A55A8B05700000000000002D305DA55A0B07000C000F3322141A00B0A55A0B070091840F3322132100BFA55A0B070014000F3322142300C1A55A0B070018000F3322142600C8A55A0B07007A840F33221317009EA55A0B0700917D0F3322131A00B1A55A8B05500000000000002D303DA55A0B070072800F3322121A0094A55A8B070072800F0000002B30EEA55A0B07000E000F3322141B00B3A55A0B0700A27F0F3322132900D3A55A0B07008C820F3322131F00B6A55A0B07007F830F3322132500B0A55A0B070012000F3322141A00B6A55A0B070091810F3322131B00B6A55A0B07008C7E0F3322132600B9A55A0B0700967C0F3322132800C3A55A0B07007D840F3322132700B1A55A0B070019000F3322142300C6A55A0B070017000F3322142600C7A55A0B07008C710F3322122700ACA55A8B07008C710F0000003030FEA55A0B070010000F3322141B00B5A55A0B07ADA1610F0582FD2730ABA55A0B07009B800F3322132000C4

bundles/org.openhab.binding.enocean/src/test/resources/ESP3Stream1.txt

@@ -0,0 +1 @@
+55000707017AF6E0058DF4352001FFFFFFFF5B008C5555000707017AF6D0058DF4352001FFFFFFFF59002755E455000707017AF6C0058DF4352001FFFFFFFF5B007255759A55000707017AF6F0058DF4352101FFFFFFFF5C008B55000707017AF6E0FED0892F2001FFFFFFFF5C00AD55000707017AF6D0FED0892F2001FFFFFFFF55009155000707017AF6C0FED0892F2001FFFFFFFF5500EE55000707017AF6F0FED0892F2001FFFFFFFF5E00F855000A0701EBA59E16000F0582D29D0001FFFFFFFF55007155000A0701EBA5D8021E09051947330001FFFFFFFF59006755000A0701EBA578020008052939F80001FFFFFFFF49004755000A0701EBA59E14000F0582D29D0001FFFFFFFF55001C55000807013DD0064605293A000001FFFFFFFF4300E655000A0701EBA5C5021C09051947330001FFFFFFFF59009055000707017AD50905A363E90101FFFFFFFF5C004155000A0701EBA59E13000F0582D29D0001FFFFFFFF55001E55000A0701EBA55F02270905194A1D0001FFFFFFFF55009F55000A0701EBA5AA021A09051947330001FFFFFFFF59001E55000A0701EBA59E11000F0582D29D0001FFFFFFFF55007355000A0701EBA5B9014D08059A08250001FFFFFFFF5300DE55000807013DD0066405194A1D0001FFFFFFFF550029

bundles/org.openhab.binding.enocean/src/test/resources/ESP3Stream2.txt

@@ -0,0 +1 @@
+55000707017AF6E00517E9102001FFFFFFFF50000455000707017AF6D00517E9102001FFFFFFFF4A005055000A0701EBA59F27000F0582D29D0001FFFFFFFF5C00D855000707017AD50905A363E90101FFFFFFFF5B002A55000A0701EBA599020E0805293A000001FFFFFFFF4400D955000A0701EBA56D01C30905194A1D0001FFFFFFFF53000255000A0701EBA57E01FD08052939F80001FFFFFFFF4700B855000A0701EBA59F27000F0582D29D0001FFFFFFFF5C00D855000707017AD50805A363E90101FFFFFFFF5E00FF55000707017AD50805A363E90001FFFFFFFF5C00C655000707017AD50905A363E90001FFFFFFFF5F006D55000A0701EBA5BD020909051947330001FFFFFFFF5C002355000A0701EBA5A027000F0582D29D0001FFFFFFFF5B00CD55000A0701EBA56C01CA0905194A1D0001FFFFFFFF5200CD55000A0701EBA5C5014809059A08250001FFFFFFFF5500B455000A0701EBA591020F0905293A000001FFFFFFFF4400B855000A0701EBA59E020B09051947330101FFFFFFFF5C00BB55000A0701EBA5A027000F0582D29D0001FFFFFFFF5B00CD55000A0701EBA58E020B09051947330001FFFFFFFF5C008855000A0701EBA5A028000F0582D29D0001FFFFFFFF5F002855000707017AF6E00517E6EE2002FFFFFFFF52006855000A0701EBA589020F0905293A000001FFFFFFFF44008855000A0701EBA587020B09051947330001FFFFFFFF5B00F155000A0701EBA5A028000F0582D29D0001FFFFFFFF56009555000A0701EBA5A028000F0582D29D0001FFFFFFFF58004355000A0701EBA5C4014C09059A08250001FFFFFFFF53005555000707017AD50905A363E90101FFFFFFFF5B002A55000A0701EBA5A028000F0582D29D0001FFFFFFFF59005655000A0701EBA57D01FC08052939F80001FFFFFFFF49008D55000A0701EBA5A028000F0582D29D0001FFFFFFFF59005655000A0701EBA56901DA0905194A1D0001FFFFFFFF5500D655000A0701EBA576020F08051946E00001FFFFFFFF5B005555000A0701EBA5A028000F0582D29D0001FFFFFFFF58004355000707017AF6E00510FA8F2001FFFFFFFF5F00A855000A0701EBA57B020C0905293A000001FFFFFFFF44004355000707017AD50805A363E90101FFFFFFFF5C00D555000707017AD50805A363E90101FFFFFFFF5B00BE55000A0701EBA5CE98000805A363E90101FFFFFFFF5B002655000707017AD50805A363E90101FFFFFFFF5F00EA55000A0701EBA5C4015009059A08250001FFFFFFFF5500F155000707017AD50905A363E90101FFFFFFFF5E006B55000A0701EBA5A028000F0582D29D0001FFFFFFFF58004355000707017AD50905A363E90101FFFFFFFF5C004155000A0701EBA56901E20905194A1D0001FFFFFFFF58008C55000A0701EBA5A029000F0582D29D0001FFFFFFFF5800F655000A0701EBA5A029000F0582D29D0001FFFFFFFF5800F655000A0701EBA575020C0905293A000001FFFFFFFF44005F55000A0701EBA56D020E09051946E00001FFFFFFFF5B001255000A0701EBA5A029000F0582D29D0001FFFFFFFF5800F655000A0701EBA56901EB0905194A1D0001FFFFFFFF56008255000A0701EBA5C3015409059A08250001FFFFFFFF55006255000A0701EBA5A029000F0582D29D0001FFFFFFFF5800F655000A0701EBA57F020C08051947330001FFFFFFFF5C004C55000707017AD50905A363E90101FFFFFFFF5B002A55000A0701EBA5A029000F0582D29D0001FFFFFFFF5800F655000A0701EBA5A029000F0582D29D0001FFFFFFFF5800F655000A0701EBA56F020C0905293A000001FFFFFFFF44006B55000A0701EBA5A02A000F0582D29D0001FFFFFFFF5600F855000A0701EBA5C2015809059A08250001FFFFFFFF5500C055000A0701EBA56901F50905194A1D0001FFFFFFFF56009555000A0701EBA5A02A000F0582D29D0001FFFFFFFF59003B55000A0701EBA57C01FD08052939F80001FFFFFFFF4700BC55000A0701EBA5A02A000F0582D29D0001FFFFFFFF5B001155000A0701EBA567020C09051946E00001FFFFFFFF5C00A055000A0701EBA5A02A000F0582D29D0001FFFFFFFF5B001155000807013DD00646051947330001FFFFFFFF5C001E55000A0701EBA56901FE0905194A1D0001FFFFFFFF56008055000A0701EBA5A02A000F0582D29D0001FFFFFFFF5B001155000A0701EBA5C1015C09059A08250001FFFFFFFF55005B55000A0701EBA56A02090905293A000001FFFFFFFF44001955000A0701EBA5A02A000F0582D29D0001FFFFFFFF5C007A55000A0701EBA57F020C08051947330001FFFFFFFF5C004C55000A0701EBA5A02A000F0582D29D0001FFFFFFFF5B001155000A0701EBA562020A09051946E00001FFFFFFFF5C00FA55000A0701EBA56902060905194A1D0001FFFFFFFF59001E55000A0701EBA5A02A000F0582D29D0001FFFFFFFF5B001155000707017AD50905A363E90101FFFFFFFF59000055000A0701EBA5A02A000F0582D29D0001FFFFFFFF5B001155000707017AF6E0058DF4352001FFFFFFFF5E00CD55000A0701EBA5BF016009059A08250001FFFFFFFF55008955000A0701EBA5A02A000F0582D29D0001FFFFFFFF5B001155000A0701EBA5A02B000F0582D29D0001FFFFFFFF59008E55000A0701EBA569020D0905194A1D0001FFFFFFFF5C004A55000A0701EBA57D01FF08052939F80001FFFFFFFF47007355000A0701EBA5A02B000F0582D29D0001FFFFFFFF56004D55000A0701EBA5A02B000F0582D29D0001FFFFFFFF58009B55000707017AD50905A363E90101FFFFFFFF5C004155000A0701EBA55D020B09051946E00001FFFFFFFF5F005E55000A0701EBA5BE016509059A08250001FFFFFFFF5500F355000A0701EBA56502060905293A000001FFFFFFFF44008F55000A0701EBA56902140905194A1D0001FFFFFFFF5C00E855000807013DD0066405194A1D0001FFFFFFFF5C009455000A0701EBA5A02B000F0582D29D0001FFFFFFFF56004D55000807013DD00664051946E00001FFFFFFFF5E00AD55000A0701EBA5A02B000F0582D29D0001FFFFFFFF58009B55000A0701EBA5A02B000F0582D29D0001FFFFFFFF59008E55000A0701EBA57E020E08051947330001FFFFFFFF5C008355000A0701EBA5A02B000F0582D29D0001FFFFFFFF59008E55000A0701EBA5A02B000F0582D29D0001FFFFFFFF58009B55000707017AF6F00510FA8F2001FFFFFFFF5B0083