rework ebus protocol engine (consume buffered data in a row, avoid starting arbitration when data is buffered, separate transport to/from device)

ChangeLog.md

@@ -10,13 +10,15 @@
 * fix potentially unusable SSL context
 * fix SYN generator timing
 * fix missing check for PB/SB validity
+* fix non-SSL build
 
 ## Features
 * add temperatures in Kelvin and ... to Home Assistant MQTT discovery integration
 * add options to turn off scanconfig and limit number of retries
 * remove dependency on argp
 * add time fields to Home Assistant MQTT discovery integration
-+ add templates endpoint to HTTP JSON
+* add templates endpoint to HTTP JSON
+* add reworked eBUS protocol engine that is especially useful for slow network issues
 
 
 # 23.2 (2023-07-08)

src/lib/ebus/CMakeLists.txt

@@ -7,6 +7,7 @@ set(libebus_a_SOURCES
     datatype.h datatype.cpp
     data.h data.cpp
     device.h device.cpp
+    transport.h transport.cpp
     protocol.h protocol.cpp
     protocol_direct.h protocol_direct.cpp
     message.h message.cpp

src/lib/ebus/Makefile.am

@@ -11,6 +11,7 @@ libebus_a_SOURCES = \
 		    datatype.h datatype.cpp \
 		    data.h data.cpp \
 		    device.h device.cpp \
+		    transport.h transport.cpp \
 		    protocol.h protocol.cpp \
 		    protocol_direct.h protocol_direct.cpp \
 		    message.h message.cpp \

src/lib/ebus/device.h

@@ -19,15 +19,9 @@
 #ifndef LIB_EBUS_DEVICE_H_
 #define LIB_EBUS_DEVICE_H_
 
-#include <unistd.h>
-#include <termios.h>
-#include <arpa/inet.h>
-#include <netinet/in.h>
-#include <netdb.h>
-#include <iostream>
-#include <fstream>
 #include <string>
 #include "lib/ebus/result.h"
+#include "lib/ebus/transport.h"
 #include "lib/ebus/symbol.h"
 
 namespace ebusd {
@@ -35,25 +29,12 @@ namespace ebusd {
 /** @file lib/ebus/device.h
  * Classes providing access to the eBUS.
  *
- * A @a Device is either a @a SerialDevice directly connected to a local tty
+ * A @a Device allows to send and receive data to/from a local or remote eBUS
- * port or a remote @a NetworkDevice handled via a TCP socket. It allows to
+ * device while optionally dumping the data to a file and/or forwarding it to
- * send and receive bytes to/from the eBUS while optionally dumping the data
+ * a logging function.
- * to a file and/or forwarding it to a logging function.
+ * The data transport itself is handled by a @a Transport instance.
  */
 
-/** the transfer latency of the network device [ms]. */
-#define NETWORK_LATENCY_MS 30
-
-/** the extra transfer latency to take into account for enhanced protocol. */
-#define ENHANCED_LATENCY_MS 10
-
-/** the latency of the host [ms]. */
-#if defined(__CYGWIN__) || defined(_WIN32)
-#define HOST_LATENCY_MS 20
-#else
-#define HOST_LATENCY_MS 10
-#endif
-
 /** the arbitration state handled by @a Device. */
 enum ArbitrationState {
   as_none,     //!< no arbitration in process
@@ -65,11 +46,6 @@ enum ArbitrationState {
   as_won,      //!< arbitration won
 };
 
-/** the sequence IDs as handled by @a FileDevice. */
-enum SequenceId {
-  sid_info,     //!< send/receive info
-};
-
 /**
  * Interface for listening to data received on/sent to a device.
  */
@@ -81,11 +57,12 @@ class DeviceListener {
   virtual ~DeviceListener() {}
 
   /**
-   * Listener method that is called when a symbol was received from/sent to eBUS.
+   * Listener method that is called when symbols were received from/sent to eBUS.
-   * @param symbol the received/sent symbol.
+   * @param data the received/sent data.
+   * @param len the length of received/sent data.
    * @param received @a true on reception, @a false on sending.
    */
-  virtual void notifyDeviceData(symbol_t symbol, bool received) = 0;  // abstract
+  virtual void notifyDeviceData(symbol_t* data, size_t len, bool received) = 0;  // abstract
 
   /**
    * Called to notify a status message from the device.
@@ -99,25 +76,32 @@ class DeviceListener {
 /**
  * The base class for accessing an eBUS.
  */
-class Device {
+class Device : public TransportListener {
  protected:
   /**
    * Construct a new instance.
-   * @param name the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
+   * @param transport the @a Transport to use.
    */
-  explicit Device(const char* name);
+  explicit Device(Transport* transport)
+  : m_transport(transport), m_listener(nullptr) {
+  }
 
  public:
   /**
    * Destructor.
    */
-  virtual ~Device() { }
+  virtual ~Device() {
+    if (m_transport) {
+      delete m_transport;
+      m_transport = nullptr;
+    }
+  }
 
   /**
    * Get the device name.
    * @return the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
    */
-  const char* getName() const { return m_name; }
+  const char* getName() const { return m_transport->getName(); }
 
   /**
    * Set the @a DeviceListener.
@@ -131,43 +115,75 @@ class Device {
    * @param verbose whether to add verbose infos.
    * @param prefix true for the synchronously retrievable prefix, false for the potentially asynchronous suffix.
    */
-  virtual void formatInfo(ostringstream* output, bool verbose, bool prefix) = 0;
+  virtual void formatInfo(ostringstream* output, bool verbose, bool prefix) {
+    if (prefix) {
+      *output << m_transport->getName() << ", " << m_transport->getTransportInfo();
+    } else if (!m_transport->isValid()) {
+      *output << ", invalid";
+    }
+  }
 
   /**
    * Format device infos in JSON format.
    * @param output the @a ostringstream to append the infos to.
    */
-  virtual void formatInfoJson(ostringstream* output) = 0;
+  virtual void formatInfoJson(ostringstream* output) const {}
+
+  /**
+   * @return whether the device supports checking for version updates.
+   */
+  virtual bool supportsUpdateCheck() const { return false; }
+
+  // @copydoc
+  virtual result_t notifyTransportStatus(bool opened) {
+    m_listener->notifyDeviceStatus(!opened, opened ? "transport opened" : "transport closed");
+    return RESULT_OK;
+  }
+
+  // @copydoc
+  virtual void notifyTransportMessage(bool error, const char* message) {
+    m_listener->notifyDeviceStatus(error, message);
+  }
 
   /**
    * Open the file descriptor.
    * @return the @a result_t code.
    */
-  virtual result_t open() = 0;
+  virtual result_t open() { return m_transport->open(); }
-
-  /**
-   * Has to be called by subclasses upon successful opening the device as last action in open().
-   * @return the @a result_t code.
-   */
-  virtual result_t afterOpen() { return RESULT_OK; }
-
-  /**
-   * Close the file descriptor if opened.
-   */
-  virtual void close() = 0;
 
   /**
    * Return whether the device is opened and available.
    * @return whether the device is opened and available.
    */
-  virtual bool isValid() = 0;
+  virtual bool isValid() { return m_transport->isValid(); }
 
+ protected:
+  /** the @a Transport to use. */
+  Transport* m_transport;
+
+  /** the @a DeviceListener, or nullptr. */
+  DeviceListener* m_listener;
+};
+
+
+class CharDevice : public Device {
+ protected:
+  /**
+   * Construct a new instance.
+   * @param transport the @a Transport to use.
+   */
+  explicit CharDevice(Transport* transport)
+  : Device(transport), m_arbitrationMaster(SYN), m_arbitrationCheck(0) {
+    transport->setListener(this);
+  }
+
+ public:
   /**
    * Write a single byte to the device.
    * @param value the byte value to write.
    * @return the @a result_t code.
    */
-  virtual result_t send(symbol_t value) = 0;
+  virtual result_t send(symbol_t value) = 0;  // abstract
 
   /**
    * Read a single byte from the device.
@@ -177,7 +193,7 @@ class Device {
    * @a as_won, the received byte is the master address that was successfully arbitrated with.
    * @return the result_t code.
    */
-  virtual result_t recv(unsigned int timeout, symbol_t* value, ArbitrationState* arbitrationState) = 0;
+  virtual result_t recv(unsigned int timeout, symbol_t* value, ArbitrationState* arbitrationState) = 0;  // abstract
 
   /**
    * Start the arbitration with the specified master address. A subsequent request while an arbitration is currently in
@@ -185,74 +201,65 @@ class Device {
    * @param masterAddress the master address, or @a SYN to cancel a previous arbitration request.
    * @return the result_t code.
    */
-  virtual result_t startArbitration(symbol_t masterAddress) = 0;
+  virtual result_t startArbitration(symbol_t masterAddress);
 
   /**
    * Return whether the device is currently in arbitration.
    * @return true when the device is currently in arbitration.
    */
-  virtual bool isArbitrating() const = 0;
+  virtual bool isArbitrating() const { return m_arbitrationMaster != SYN; }
 
   /**
-   * @return whether the device supports checking for version updates.
+   * Cancel a running arbitration.
+   * @param arbitrationState the reference in which @a as_error is stored when cancelled.
+   * @return true if it was cancelled, false if not.
    */
-  virtual bool supportsUpdateCheck() const = 0;
+  virtual bool cancelRunningArbitration(ArbitrationState* arbitrationState);
 
  protected:
-  /** the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network). */
+  /** the arbitration master address to send when in arbitration, or @a SYN. */
-  const char* m_name;
+  symbol_t m_arbitrationMaster;
 
-  /** the @a DeviceListener, or nullptr. */
+  /** >0 when in arbitration and the next received symbol needs to be checked against the sent master address,
-  DeviceListener* m_listener;
+   * incremented with each received SYN when arbitration was not performed as expected and needs to be stopped. */
+  size_t m_arbitrationCheck;
 };
 
 
-/** the possible enhanced protocol levels. */
+class PlainCharDevice : public CharDevice {
-enum EnhancedLevel {
-  el_none = 0,   //!< non-enhanced
-  el_basic = 1,  //!< enhanced basic
-  el_speed = 2,  //!< enhanced high-speed
-};
-
-
-/**
- * The common base class for devices using a file descriptor.
- */
-class FileDevice : public Device {
- protected:
-  /**
-   * Construct a new instance.
-   * @param name the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
-   * @param checkDevice whether to regularly check the device availability.
-   * @param latency the bus transfer latency in milliseconds.
-   * @param enhancedLevel whether to use the ebusd enhanced protocol.
-   */
-  FileDevice(const char* name, bool checkDevice, unsigned int latency,
-      EnhancedLevel enhancedLevel);
-
  public:
   /**
-   * Destructor.
+   * Construct a new instance.
+   * @param transport the @a Transport to use.
    */
-  virtual ~FileDevice();
+  explicit PlainCharDevice(Transport* transport)
+  : CharDevice(transport) {
+  }
+
+  // @copydoc
+  result_t send(symbol_t value) override;
+
+  // @copydoc
+  result_t recv(unsigned int timeout, symbol_t* value, ArbitrationState* arbitrationState) override;
+};
+
+
+class EnhancedCharDevice : public CharDevice {
+ public:
+  /**
+   * Construct a new instance.
+   * @param transport the @a Transport to use.
+   */
+  explicit EnhancedCharDevice(Transport* transport)
+  : CharDevice(transport), m_resetRequested(false),
+    m_extraFatures(0), m_infoReqTime(0), m_infoLen(0), m_infoPos(0) {
+  }
 
   // @copydoc
   void formatInfo(ostringstream* output, bool verbose, bool prefix) override;
 
   // @copydoc
-  void formatInfoJson(ostringstream* output) override;
+  void formatInfoJson(ostringstream* output) const override;
-
-  // @copydoc
-  result_t open() override;
-
-  // @copydoc
-  result_t afterOpen() override;
-
-  // @copydoc
-  void close() override;
-
-  // @copydoc
-  bool isValid() override;
 
   // @copydoc
   result_t send(symbol_t value) override;
@@ -264,50 +271,19 @@ class FileDevice : public Device {
   result_t startArbitration(symbol_t masterAddress) override;
 
   // @copydoc
-  bool isArbitrating() const override { return m_arbitrationMaster != SYN; }
+  virtual result_t notifyTransportStatus(bool opened);
-
-  /**
-   * Get the transfer latency of this device.
-   * @return the transfer latency in milliseconds.
-   */
-  virtual unsigned int getLatency() const { return m_latency; }
-
-  /**
-   * Return whether the device supports the ebusd enhanced protocol.
-   * @return whether the device supports the ebusd enhanced protocol.
-   */
-  bool isEnhancedProto() const { return m_enhancedLevel != el_none; }
-
-  /**
-   * Get info about enhanced protocol support as string.
-   * @return a @a string describing level of enhanced protocol support, or the empty string.
-   */
-  virtual string getEnhancedProtoInfo() const { return m_enhancedLevel ? "enhanced" : ""; }
 
   // @copydoc
-  bool supportsUpdateCheck() const override { return m_enhancedLevel && m_extraFatures & 0x01; }
+  bool supportsUpdateCheck() const override { return m_extraFatures & 0x01; }
-
-  /**
-   * @return whether the device supports the ebusd enhanced protocol and supports querying extra infos.
-   */
-  bool supportsEnhancedInfos() const { return m_enhancedLevel && m_extraFatures & 0x01; }
 
   /**
    * Check for a running extra infos request, wait for it to complete,
    * and then send a new request for extra infos to enhanced device.
    * @param infoId the ID of the info to request.
+   * @param wait true to wait for a running request to complete, false to send right away.
    * @return @a RESULT_OK on success, or an error code otherwise.
    */
-  result_t requestEnhancedInfo(symbol_t infoId);
+  result_t requestEnhancedInfo(symbol_t infoId, bool wait = true);
-
-  /**
-   * Write a sequence of bytes to the device.
-   * @param id the ID of the sequence.
-   * @param data the buffer with the data to send.
-   * @param len the length of the buffer.
-   * @return the @a result_t code.
-   */
-  virtual result_t sendSequence(SequenceId id, const uint8_t* data = nullptr, size_t len = 0);
 
   /**
    * Get the enhanced device version.
@@ -321,12 +297,7 @@ class FileDevice : public Device {
    */
   string getEnhancedInfos();
 
- protected:
+ private:
-  /**
-   * Check if the device is still available and close it if not.
-   */
-  virtual void checkDevice() = 0;  // abstract
-
   /**
    * Cancel a running arbitration.
    * @param arbitrationState the reference in which @a as_error is stored when cancelled.
@@ -334,83 +305,22 @@ class FileDevice : public Device {
    */
   bool cancelRunningArbitration(ArbitrationState* arbitrationState);
 
-  /**
-   * Write a single byte.
-   * @param value the byte value to write.
-   * @param startArbitration true to start arbitration.
-   * @return true on success, false on error.
-   */
-  virtual bool write(symbol_t value, bool startArbitration = false);
-
-  /**
-   * Check whether a symbol is available for reading immediately (without waiting).
-   * @return true when a symbol is available for reading immediately.
-   */
-  virtual bool available();
-
-  /**
-   * Read a single byte.
-   * @param value the reference in which the read byte value is stored.
-   * @param isAvailable the result of the immediately preceding call to @a available().
-   * @param arbitrationState the variable in which to store the current/received arbitration state (mandatory for enhanced proto).
-   * @param incomplete the variable in which to store when a partial transfer needs another poll.
-   * @return true on success, false on error.
-   */
-  virtual bool read(symbol_t* value, bool isAvailable, ArbitrationState* arbitrationState = nullptr,
-                    bool* incomplete = nullptr);
-
-  /** whether to regularly check the device availability. */
-  const bool m_checkDevice;
-
-  /** the bus transfer latency in milliseconds. */
-  const unsigned int m_latency;
-
-  /** whether the device supports the ebusd enhanced protocol. */
-  const EnhancedLevel m_enhancedLevel;
-
-  /** the opened file descriptor, or -1. */
-  int m_fd;
-
-  /** whether the reset of an enhanced device was already requested. */
-  bool m_resetRequested;
-
- private:
   /**
    * Handle the already buffered enhanced data.
    * @param value the reference in which the read byte value is stored.
-   * @param arbitrationState the variable in which to store the current/received arbitration state (mandatory for enhanced proto).
+   * @param arbitrationState the variable in which to store the current/received arbitration state.
-   * @return true if the value was set, false otherwise.
+   * @return the @a result_t code, especially RESULT_CONTINE if the value was set and more data is available immediately.
    */
-  bool handleEnhancedBufferedData(symbol_t* value, ArbitrationState* arbitrationState);
+  result_t handleEnhancedBufferedData(const uint8_t* data, size_t len, symbol_t* value,
+  ArbitrationState* arbitrationState);
 
   /**
    * Called when reception of an info ID was completed.
    */
   void notifyInfoRetrieved();
 
-  /** the arbitration master address to send when in arbitration, or @a SYN. */
+  /** whether the reset of the device was already requested. */
-  symbol_t m_arbitrationMaster;
+  bool m_resetRequested;
-
-  /** >0 when in arbitration and the next received symbol needs to be checked against the sent master address,
-   * incremented with each received SYN when arbitration was not performed as expected and needs to be stopped. */
-  size_t m_arbitrationCheck;
-
-  /** the read buffer. */
-  symbol_t* m_buffer;
-
-  /** the read buffer size (multiple of 4). */
-  size_t m_bufSize;
-
-  /** the read buffer fill length. */
-  size_t m_bufLen;
-
-  /** the read buffer read position. */
-  size_t m_bufPos;
-  /** the send buffer. */
-  uint8_t* m_sendBuf;
-
-  /** the send buffer size. */
-  size_t m_sendBufSize;
 
   /** the extra features supported by the device. */
   symbol_t m_extraFatures;
@@ -440,96 +350,6 @@ class FileDevice : public Device {
   string m_enhInfoBusVoltage;
 };
 
-
-/**
- * The @a Device for directly connected serial interfaces (tty).
- */
-class SerialDevice : public FileDevice {
- public:
-  /**
-   * Construct a new instance.
-   * @param name the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
-   * @param checkDevice whether to regularly check the device availability.
-   * @param extraLatency the extra bus transfer latency in milliseconds.
-   * @param enhancedLevel whether to use the ebusd enhanced protocol.
-   */
-  SerialDevice(const char* name, bool checkDevice, unsigned int extraLatency,
-               EnhancedLevel enhancedLevel)
-    : FileDevice(name, checkDevice, extraLatency, enhancedLevel) {
-  }
-
-  // @copydoc
-  string getEnhancedProtoInfo() const override {
-    return m_enhancedLevel == el_speed ? "enhanced high speed" : FileDevice::getEnhancedProtoInfo();
-  }
-
-  // @copydoc
-  result_t open() override;
-
-  // @copydoc
-  void close() override;
-
-
- protected:
-  // @copydoc
-  void checkDevice() override;
-
-
- private:
-  /** the previous settings of the device for restoring. */
-  termios m_oldSettings;
-};
-
-/**
- * The @a Device for remote network interfaces.
- */
-class NetworkDevice : public FileDevice {
- public:
-  /**
-   * Construct a new instance.
-   * @param name the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
-   * @param address the socket address of the device.
-   * @param hostOrIp the host name or IP address of the device.
-   * @param port the TCP or UDP port of the device.
-   * @param extraLatency the extra bus transfer latency in milliseconds.
-   * @param udp true for UDP, false to TCP.
-   * @param enhancedLevel whether to use the ebusd enhanced protocol.
-   */
-  NetworkDevice(const char* name, const char* hostOrIp, uint16_t port, unsigned int extraLatency,
-      bool udp, EnhancedLevel enhancedLevel)
-    : FileDevice(name, true, NETWORK_LATENCY_MS+extraLatency, enhancedLevel),
-    m_hostOrIp(hostOrIp), m_port(port), m_udp(udp) {}
-
-  /**
-   * Destructor.
-   */
-  ~NetworkDevice() override {
-    if (m_hostOrIp) {
-      free((void*)m_hostOrIp);
-      m_hostOrIp = nullptr;
-    }
-  }
-
-  // @copydoc
-  result_t open() override;
-
-
- protected:
-  // @copydoc
-  void checkDevice() override;
-
-
- private:
-  /** the host name or IP address of the device. */
-  const char* m_hostOrIp;
-
-  /** the TCP or UDP port of the device. */
-  const uint16_t m_port;
-
-  /** true for UDP, false to TCP. */
-  const bool m_udp;
-};
-
 }  // namespace ebusd
 
 #endif  // LIB_EBUS_DEVICE_H_

src/lib/ebus/protocol.cpp

@@ -57,21 +57,18 @@ bool ActiveBusRequest::notify(result_t result, const SlaveSymbolString& slave) {
 ProtocolHandler* ProtocolHandler::create(const ebus_protocol_config_t config,
   ProtocolListener* listener) {
   const char* name = config.device;
-  EnhancedLevel enhanced = el_none;
+  bool enhanced = false;
-  if (strncmp(name, "en", 2) == 0 && name[2] && name[3] == ':') {
+  uint8_t speed = 0;
-    switch (name[2]) {
+  if (name[0] == 'e' && name[1] && name[2] && name[3] == ':') {
-      case 's':
+    speed = name[2] == 's' ? 2 : name[2] == 'h' ? 1 : 0;
-        enhanced = el_speed;
+    enhanced = speed > 0 && name[1] == 'n';
-        break;
-      case 'h':
-        enhanced = el_basic;
-        break;
-    }
     if (enhanced) {
       name += 4;
+    } else {
+      speed = 0;
     }
   }
-  FileDevice* device = nullptr;
+  Transport* transport;
   if (strchr(name, '/') == nullptr && strchr(name, ':') != nullptr) {
     char* in = strdup(name);
     bool udp = false;
@@ -95,10 +92,16 @@ ProtocolHandler* ProtocolHandler::create(const ebus_protocol_config_t config,
     *portpos = 0;
     char* hostOrIp = strdup(addrpos);
     free(in);
-    device = new NetworkDevice(name, hostOrIp, port, config.extraLatency, udp, enhanced);
+    transport = new NetworkTransport(name, config.extraLatency, hostOrIp, port, udp);
   } else {
-    // support enx:/dev/<device>, ens:/dev/<device>, enh:/dev/<device>, and /dev/<device>
+    // support ens:/dev/<device>, enh:/dev/<device>, and /dev/<device>
-    device = new SerialDevice(name, !config.noDeviceCheck, config.extraLatency, enhanced);
+    transport = new SerialTransport(name, config.extraLatency, !config.noDeviceCheck, speed);
+  }
+  CharDevice* device;
+  if (enhanced) {
+    device = new EnhancedCharDevice(transport);
+  } else {
+    device = new PlainCharDevice(transport);
   }
   return new DirectProtocolHandler(config, device, listener);
 }
@@ -124,55 +127,56 @@ void ProtocolHandler::formatInfo(ostringstream* ostream, bool verbose, bool noWa
   m_device->formatInfo(ostream, verbose, false);
 }
 
-void ProtocolHandler::formatInfoJson(ostringstream* ostream) {
+void ProtocolHandler::formatInfoJson(ostringstream* ostream) const {
   m_device->formatInfoJson(ostream);
 }
 
-void ProtocolHandler::notifyDeviceData(symbol_t symbol, bool received) {
+void ProtocolHandler::notifyDeviceData(symbol_t* data, size_t len, bool received) {
   if (received && m_dumpFile) {
-    m_dumpFile->write(&symbol, 1);
+    m_dumpFile->write(data, len);
   }
   if (!m_logRawFile && !m_logRawEnabled) {
     return;
   }
   if (m_logRawBytes) {
     if (m_logRawFile) {
-      m_logRawFile->write(&symbol, 1, received);
+      m_logRawFile->write(data, len, received);
     } else if (m_logRawEnabled) {
-      if (received) {
+      for (size_t pos = 0; pos < len; pos++) {
-        logNotice(lf_bus, "<%02x", symbol);
+        logNotice(lf_bus, "%c%02x", received ? '<' : '>', data[pos]);
-      } else {
-        logNotice(lf_bus, ">%02x", symbol);
       }
     }
     return;
   }
-  if (symbol != SYN) {
+  for (size_t pos = 0; pos < len; pos++) {
-    if (received && !m_logRawLastReceived && symbol == m_logRawLastSymbol) {
+    symbol_t symbol = data[pos];
-      return;  // skip received echo of previously sent symbol
+    if (symbol != SYN) {
+      if (received && !m_logRawLastReceived && symbol == m_logRawLastSymbol) {
+        continue;  // skip received echo of previously sent symbol
+      }
+      if (m_logRawBuffer.tellp() == 0 || received != m_logRawLastReceived) {
+        m_logRawLastReceived = received;
+        if (m_logRawBuffer.tellp() == 0 && m_logRawLastSymbol != SYN) {
+          m_logRawBuffer << "...";
+        }
+        m_logRawBuffer << (received ? "<" : ">");
+      }
+      m_logRawBuffer << setw(2) << setfill('0') << hex << static_cast<unsigned>(symbol);
     }
-    if (m_logRawBuffer.tellp() == 0 || received != m_logRawLastReceived) {
+    m_logRawLastSymbol = symbol;
-      m_logRawLastReceived = received;
+    if (m_logRawBuffer.tellp() > (symbol == SYN ? 0 : 64)) {  // flush: direction+5 hdr+24 max data+crc+direction+ack+1
-      if (m_logRawBuffer.tellp() == 0 && m_logRawLastSymbol != SYN) {
+      if (symbol != SYN) {
         m_logRawBuffer << "...";
       }
-      m_logRawBuffer << (received ? "<" : ">");
+      const string bufStr = m_logRawBuffer.str();
+      const char* str = bufStr.c_str();
+      if (m_logRawFile) {
+        m_logRawFile->write((const unsigned char*)str, strlen(str), received, false);
+      } else {
+        logNotice(lf_bus, str);
+      }
+      m_logRawBuffer.str("");
     }
-    m_logRawBuffer << setw(2) << setfill('0') << hex << static_cast<unsigned>(symbol);
-  }
-    m_logRawLastSymbol = symbol;
-  if (m_logRawBuffer.tellp() > (symbol == SYN ? 0 : 64)) {  // flush: direction+5 hdr+24 max data+crc+direction+ack+1
-    if (symbol != SYN) {
-      m_logRawBuffer << "...";
-  }
-    const string bufStr = m_logRawBuffer.str();
-    const char* str = bufStr.c_str();
-    if (m_logRawFile) {
-      m_logRawFile->write((const unsigned char*)str, strlen(str), received, false);
-    } else {
-      logNotice(lf_bus, str);
-    }
-    m_logRawBuffer.str("");
   }
 }
 

src/lib/ebus/protocol.h

@@ -332,7 +332,7 @@ class ProtocolHandler : public WaitThread, DeviceListener {
    * Format device/protocol infos in JSON format.
    * @param output the @a ostringstream to append the infos to.
    */
-  virtual void formatInfoJson(ostringstream* output);
+  virtual void formatInfoJson(ostringstream* output) const;
 
   /**
    * @return whether to allow read access to the device only.
@@ -381,7 +381,7 @@ class ProtocolHandler : public WaitThread, DeviceListener {
   virtual bool supportsUpdateCheck() const { return m_device->supportsUpdateCheck(); }
 
   // @copydoc
-  void notifyDeviceData(symbol_t symbol, bool received) override;
+  virtual void notifyDeviceData(symbol_t* symbols, size_t len, bool received);
 
   // @copydoc
   void notifyDeviceStatus(bool error, const char* message) override;

src/lib/ebus/protocol_direct.cpp

@@ -82,27 +82,38 @@ void DirectProtocolHandler::run() {
   lastTime += 2;
   logNotice(lf_bus, "bus started with own address %2.2x/%2.2x%s", m_ownMasterAddress, m_ownSlaveAddress,
       m_config.answer?" in answer mode":"");
-
   do {
-    if (m_device->isValid() && !m_reconnect) {
+    bool valid = m_device->isValid();
-      result_t result = handleSymbol();
+    if (valid && !m_reconnect) {
-      time(&now);
+      unsigned int recvTimeout = 0;
-      if (result != RESULT_ERR_TIMEOUT && now >= lastTime) {
+      symbol_t sentSymbol = ESC;
-        symCount++;
+      struct timespec sentTime;
-      }
+      result_t result = handleSend(&recvTimeout, &sentSymbol, &sentTime);
-      if (now > lastTime) {
+      bool sent = result == RESULT_CONTINUE;
-        m_symPerSec = symCount / (unsigned int)(now-lastTime);
+      do {
-        if (m_symPerSec > m_maxSymPerSec) {
+        if (result >= RESULT_OK) {
-          m_maxSymPerSec = m_symPerSec;
+          result = handleReceive(recvTimeout, sent, sentSymbol, &sentTime);
-          if (m_maxSymPerSec > 100) {
-            logNotice(lf_bus, "max. symbols per second: %d", m_maxSymPerSec);
-          }
         }
-        lastTime = now;
+        time(&now);
-        symCount = 0;
+        if (result != RESULT_ERR_TIMEOUT && now >= lastTime) {
-      }
+          symCount++;
+        }
+        if (now > lastTime) {
+          m_symPerSec = symCount / (unsigned int)(now-lastTime);
+          if (m_symPerSec > m_maxSymPerSec) {
+            m_maxSymPerSec = m_symPerSec;
+            if (m_maxSymPerSec > 100) {
+              logNotice(lf_bus, "max. symbols per second: %d", m_maxSymPerSec);
+            }
+          }
+          lastTime = now;
+          symCount = 0;
+        }
+        recvTimeout = 0;  // for further buffered bytes
+        sent = false;
+      } while (result == RESULT_CONTINUE);
     } else {
-      if (!m_device->isValid()) {
+      if (!valid) {
         logNotice(lf_bus, "device invalid");
         setState(bs_noSignal, RESULT_ERR_DEVICE);
       }
@@ -136,7 +147,8 @@ void DirectProtocolHandler::run() {
 #endif
 #endif
 
-result_t DirectProtocolHandler::handleSymbol() {
+result_t DirectProtocolHandler::handleSend(unsigned int* recvTimeout, symbol_t* sentSymbol,
+struct timespec* sentTime) {
   unsigned int timeout = SYN_TIMEOUT;
   symbol_t sendSymbol = ESC;
   bool sending = false;
@@ -247,8 +259,6 @@ result_t DirectProtocolHandler::handleSymbol() {
   }
 
   // send symbol if necessary
-  result_t result;
-  struct timespec sentTime, recvTime;
   if (sending && !m_config.readOnly) {
     if (m_state != bs_sendSyn && (sendSymbol == ESC || sendSymbol == SYN)) {
       if (m_escape) {
@@ -258,43 +268,51 @@ result_t DirectProtocolHandler::handleSymbol() {
         sendSymbol = ESC;
       }
     }
-    result = m_device->send(sendSymbol);
+    result_t result = m_device->send(sendSymbol);
-    clockGettime(&sentTime);
+    clockGettime(sentTime);
     if (result == RESULT_OK) {
       if (m_state == bs_ready) {
         timeout = m_config.busAcquireTimeout;
       } else {
         timeout = SEND_TIMEOUT;
       }
+      *sentSymbol = sendSymbol;
     } else {
       sending = false;
       timeout = SYN_TIMEOUT;
       setState(bs_skip, result);
     }
+    *recvTimeout = timeout;
+    return sending ? RESULT_CONTINUE : result;
   } else {
-    clockGettime(&sentTime);  // for measuring arbitration delay in enhanced protocol
+    clockGettime(sentTime);  // for measuring arbitration delay in enhanced protocol
   }
+  *recvTimeout = timeout;
+  return RESULT_OK;
+}
 
+result_t DirectProtocolHandler::handleReceive(unsigned int timeout, bool sending, symbol_t sentSymbol,
+struct timespec* sentTime) {
   // receive next symbol (optionally check reception of sent symbol)
   symbol_t recvSymbol;
+  struct timespec recvTime;
   ArbitrationState arbitrationState = as_none;
-  result = m_device->recv(timeout, &recvSymbol, &arbitrationState);
+  result_t result = m_device->recv(timeout, &recvSymbol, &arbitrationState);
+  bool sentAutoSyn = false;
   if (sending) {
     clockGettime(&recvTime);
-  }
+  } else if (!m_config.readOnly && result == RESULT_ERR_TIMEOUT && m_generateSynInterval > 0
-  bool sentAutoSyn = false;
-  if (!sending && !m_config.readOnly && result == RESULT_ERR_TIMEOUT && m_generateSynInterval > 0
       && timeout >= m_generateSynInterval && (m_state == bs_noSignal || m_state == bs_skip)) {
     // check if acting as AUTO-SYN generator is required
     result = m_device->send(SYN);
     if (result != RESULT_OK) {
       return setState(bs_skip, result);
     }
-    clockGettime(&sentTime);
+    clockGettime(sentTime);
     recvSymbol = ESC;
     result = m_device->recv(SEND_TIMEOUT, &recvSymbol, &arbitrationState);
     clockGettime(&recvTime);
-    if (result != RESULT_OK) {
+    if (result < RESULT_OK) {
       logError(lf_bus, "unable to receive sent AUTO-SYN symbol: %s", getResultCode(result));
       return setState(bs_noSignal, result);
     }
@@ -302,7 +320,7 @@ result_t DirectProtocolHandler::handleSymbol() {
       logError(lf_bus, "received %2.2x instead of AUTO-SYN symbol", recvSymbol);
       return setState(bs_noSignal, result);
     }
-    measureLatency(&sentTime, &recvTime);
+    measureLatency(sentTime, &recvTime);
     if (m_generateSynInterval != SYN_INTERVAL) {
       // received own AUTO-SYN symbol back again: act as AUTO-SYN generator now
       m_generateSynInterval = SYN_INTERVAL;
@@ -337,7 +355,7 @@ result_t DirectProtocolHandler::handleSymbol() {
         } else {
           logDebug(lf_bus, "arbitration won");
           m_currentRequest = startRequest;
-          sendSymbol = m_currentRequest->getMaster()[0];
+          sentSymbol = m_currentRequest->getMaster()[0];
           sending = true;
         }
       }
@@ -361,11 +379,11 @@ result_t DirectProtocolHandler::handleSymbol() {
       break;
   }
   if (sentAutoSyn && !sending) {
-    return RESULT_OK;
+    return result;
   }
   time_t now;
   time(&now);
-  if (result != RESULT_OK) {
+  if (result < RESULT_OK) {
     if ((m_generateSynInterval != SYN_INTERVAL && difftime(now, m_lastReceive) > 1)
       // at least one full second has passed since last received symbol
       || m_state == bs_noSignal) {
@@ -376,20 +394,26 @@ result_t DirectProtocolHandler::handleSymbol() {
 
   m_lastReceive = now;
   if ((recvSymbol == SYN) && (m_state != bs_sendSyn)) {
-    if (!sending && m_remainLockCount > 0 && m_command.size() != 1) {
+    if (result == RESULT_CONTINUE) {
-      m_remainLockCount--;
+      if (m_remainLockCount == 0) {
-    } else if (!sending && m_remainLockCount == 0 && m_command.size() == 1) {
+        m_remainLockCount = 1;  // avoid starting arbitration when more data is already buffered
-      m_remainLockCount = 1;  // wait for next AUTO-SYN after SYN / address / SYN (bus locked for own priority)
+      }
+    } else if (!sending) {
+      if (m_remainLockCount > 0 && m_command.size() != 1) {
+        m_remainLockCount--;
+      } else if (m_remainLockCount == 0 && m_command.size() == 1) {
+        m_remainLockCount = 1;  // wait for next AUTO-SYN after SYN / address / SYN (bus locked for own priority)
+      }
     }
-    clockGettime(&m_lastSynReceiveTime);
+    m_lastSynReceiveTime = recvTime;
-    return setState(bs_ready, m_state == bs_skip ? RESULT_OK : RESULT_ERR_SYN);
+    return setState(bs_ready, m_state == bs_skip || m_remainLockCount > 0 ? result : RESULT_ERR_SYN);
   }
 
   if (sending && m_state != bs_ready) {  // check received symbol for equality if not in arbitration
-    if (recvSymbol != sendSymbol) {
+    if (recvSymbol != sentSymbol) {
       return setState(bs_skip, RESULT_ERR_SYMBOL);
     }
-    measureLatency(&sentTime, &recvTime);
+    measureLatency(sentTime, &recvTime);
   }
 
   switch (m_state) {
@@ -407,10 +431,10 @@ result_t DirectProtocolHandler::handleSymbol() {
   if (m_escape) {
     // check escape/unescape state
     if (sending) {
-      if (sendSymbol == ESC) {
+      if (sentSymbol == ESC) {
-        return RESULT_OK;
+        return result;
       }
-      sendSymbol = recvSymbol = m_escape;
+      sentSymbol = recvSymbol = m_escape;
     } else {
       if (recvSymbol > 0x01) {
         return setState(bs_skip, RESULT_ERR_ESC);
@@ -420,23 +444,23 @@ result_t DirectProtocolHandler::handleSymbol() {
     m_escape = 0;
   } else if (!sending && recvSymbol == ESC) {
     m_escape = ESC;
-    return RESULT_OK;
+    return result;
   }
 
   switch (m_state) {
   case bs_noSignal:
-    return setState(bs_skip, RESULT_OK);
+    return setState(bs_skip, result);
 
   case bs_skip:
-    return RESULT_OK;
+    return result;
 
   case bs_ready:
     if (m_currentRequest != nullptr && sending) {
       // check arbitration
-      if (recvSymbol == sendSymbol) {  // arbitration successful
+      if (recvSymbol == sentSymbol) {  // arbitration successful
         // measure arbitration delay
-        int64_t latencyLong = (sentTime.tv_sec*1000000000 + sentTime.tv_nsec
+        int64_t latencyLong = (sentTime->tv_sec*1000000000LL + sentTime->tv_nsec
-        - m_lastSynReceiveTime.tv_sec*1000000000 - m_lastSynReceiveTime.tv_nsec)/1000;
+        - m_lastSynReceiveTime.tv_sec*1000000000LL - m_lastSynReceiveTime.tv_nsec)/1000;
         if (latencyLong >= 0 && latencyLong <= 10000) {  // skip clock skew or out of reasonable range
           auto latency = static_cast<int>(latencyLong);
           logDebug(lf_bus, "arbitration delay %d micros", latency);
@@ -452,11 +476,11 @@ result_t DirectProtocolHandler::handleSymbol() {
         }
         m_nextSendPos = 1;
         m_repeat = false;
-        return setState(bs_sendCmd, RESULT_OK);
+        return setState(bs_sendCmd, result);
       }
       // arbitration lost. if same priority class found, try again after next AUTO-SYN
       m_remainLockCount = isMaster(recvSymbol) ? 2 : 1;  // number of SYN to wait for before next send try
-      if ((recvSymbol & 0x0f) != (sendSymbol & 0x0f) && m_lockCount > m_remainLockCount) {
+      if ((recvSymbol & 0x0f) != (sentSymbol & 0x0f) && m_lockCount > m_remainLockCount) {
         // if different priority class found, try again after N AUTO-SYN symbols (at least next AUTO-SYN)
         m_remainLockCount = m_lockCount;
       }
@@ -464,7 +488,7 @@ result_t DirectProtocolHandler::handleSymbol() {
     }
     m_command.push_back(recvSymbol);
     m_repeat = false;
-    return setState(bs_recvCmd, RESULT_OK);
+    return setState(bs_recvCmd, result);
 
   case bs_recvCmd:
     if ((m_command.size() == 0 && !isMaster(recvSymbol))
@@ -473,9 +497,9 @@ result_t DirectProtocolHandler::handleSymbol() {
     }
     m_command.push_back(recvSymbol);
     if (m_command.isComplete()) {  // all data received
-      return setState(bs_recvCmdCrc, RESULT_OK);
+      return setState(bs_recvCmdCrc, result);
     }
-    return RESULT_OK;
+    return result;
 
   case bs_recvCmdCrc:
     m_crcValid = recvSymbol == m_crc;
@@ -483,7 +507,7 @@ result_t DirectProtocolHandler::handleSymbol() {
       if (m_crcValid) {
         addSeenAddress(m_command[0]);
         messageCompleted();
-        return setState(bs_skip, RESULT_OK);
+        return setState(bs_skip, result);
       }
       return setState(bs_skip, RESULT_ERR_CRC);
     }
@@ -493,14 +517,14 @@ result_t DirectProtocolHandler::handleSymbol() {
         if (m_crcValid) {
           addSeenAddress(m_command[0]);
           m_currentAnswering = true;
-          return setState(bs_sendCmdAck, RESULT_OK);
+          return setState(bs_sendCmdAck, result);
         }
         return setState(bs_sendCmdAck, RESULT_ERR_CRC);
       }
     }
     if (m_crcValid) {
       addSeenAddress(m_command[0]);
-      return setState(bs_recvCmdAck, RESULT_OK);
+      return setState(bs_recvCmdAck, result);
     }
     if (m_repeat) {
       return setState(bs_skip, RESULT_ERR_CRC);
@@ -515,15 +539,15 @@ result_t DirectProtocolHandler::handleSymbol() {
       if (m_currentRequest != nullptr) {
         if (isMaster(m_currentRequest->getMaster()[1])) {
           messageCompleted();
-          return setState(bs_sendSyn, RESULT_OK);
+          return setState(bs_sendSyn, result);
         }
       } else if (isMaster(m_command[1])) {
         messageCompleted();
-        return setState(bs_skip, RESULT_OK);
+        return setState(bs_skip, result);
       }
 
       m_repeat = false;
-      return setState(bs_recvRes, RESULT_OK);
+      return setState(bs_recvRes, result);
     }
     if (recvSymbol == NAK) {
       if (!m_repeat) {
@@ -543,17 +567,17 @@ result_t DirectProtocolHandler::handleSymbol() {
   case bs_recvRes:
     m_response.push_back(recvSymbol);
     if (m_response.isComplete()) {  // all data received
-      return setState(bs_recvResCrc, RESULT_OK);
+      return setState(bs_recvResCrc, result);
     }
-    return RESULT_OK;
+    return result;
 
   case bs_recvResCrc:
     m_crcValid = recvSymbol == m_crc;
     if (m_crcValid) {
       if (m_currentRequest != nullptr) {
-        return setState(bs_sendResAck, RESULT_OK);
+        return setState(bs_sendResAck, result);
       }
-      return setState(bs_recvResAck, RESULT_OK);
+      return setState(bs_recvResAck, result);
     }
     if (m_repeat) {
       if (m_currentRequest != nullptr) {
@@ -572,7 +596,7 @@ result_t DirectProtocolHandler::handleSymbol() {
         return setState(bs_skip, RESULT_ERR_ACK);
       }
       messageCompleted();
-      return setState(bs_skip, RESULT_OK);
+      return setState(bs_skip, result);
     }
     if (recvSymbol == NAK) {
       if (!m_repeat) {
@@ -594,17 +618,17 @@ result_t DirectProtocolHandler::handleSymbol() {
     }
     m_nextSendPos++;
     if (m_nextSendPos >= m_currentRequest->getMaster().size()) {
-      return setState(bs_sendCmdCrc, RESULT_OK);
+      return setState(bs_sendCmdCrc, result);
     }
-    return RESULT_OK;
+    return result;
 
   case bs_sendCmdCrc:
     if (m_currentRequest->getMaster()[1] == BROADCAST) {
       messageCompleted();
-      return setState(bs_sendSyn, RESULT_OK);
+      return setState(bs_sendSyn, result);
     }
     m_crcValid = true;
-    return setState(bs_recvCmdAck, RESULT_OK);
+    return setState(bs_recvCmdAck, result);
 
   case bs_sendResAck:
     if (!sending || m_currentRequest == nullptr) {
@@ -619,7 +643,7 @@ result_t DirectProtocolHandler::handleSymbol() {
       return setState(bs_sendSyn, RESULT_ERR_ACK);
     }
     messageCompleted();
-    return setState(bs_sendSyn, RESULT_OK);
+    return setState(bs_sendSyn, result);
 
   case bs_sendCmdAck:
     if (!sending || !m_config.answer) {
@@ -636,18 +660,20 @@ result_t DirectProtocolHandler::handleSymbol() {
     }
     if (isMaster(m_command[1])) {
       messageCompleted();  // TODO decode command and store value into database of internal variables
-      return setState(bs_skip, RESULT_OK);
+      return setState(bs_skip, result);
     }
 
     m_nextSendPos = 0;
     m_repeat = false;
     // build response and store in m_response for sending back to requesting master
     m_response.clear();
-    result = m_listener->notifyProtocolAnswer(m_command, &m_response);
+    {
-    if (result != RESULT_OK) {
+      result_t result = m_listener->notifyProtocolAnswer(m_command, &m_response);
-      return setState(bs_skip, result);
+      if (result != RESULT_OK) {
+        return setState(bs_skip, result);
+      }
     }
-    return setState(bs_sendRes, RESULT_OK);
+    return setState(bs_sendRes, result);
 
   case bs_sendRes:
     if (!sending || !m_config.answer) {
@@ -656,23 +682,23 @@ result_t DirectProtocolHandler::handleSymbol() {
     m_nextSendPos++;
     if (m_nextSendPos >= m_response.size()) {
       // slave data completely sent
-      return setState(bs_sendResCrc, RESULT_OK);
+      return setState(bs_sendResCrc, result);
     }
-    return RESULT_OK;
+    return result;
 
   case bs_sendResCrc:
     if (!sending || !m_config.answer) {
       return setState(bs_skip, RESULT_ERR_INVALID_ARG);
     }
-    return setState(bs_recvResAck, RESULT_OK);
+    return setState(bs_recvResAck, result);
 
   case bs_sendSyn:
     if (!sending) {
       return setState(bs_ready, RESULT_ERR_INVALID_ARG);
     }
-    return setState(bs_ready, RESULT_OK);
+    return setState(bs_ready, result);
   }
-  return RESULT_OK;
+  return result;
 }
 
 result_t DirectProtocolHandler::setState(BusState state, result_t result, bool firstRepetition) {
@@ -682,7 +708,7 @@ result_t DirectProtocolHandler::setState(BusState state, result_t result, bool f
       m_currentRequest->incrementBusLostRetries();
       m_nextRequests.push(m_currentRequest);  // repeat
       m_currentRequest = nullptr;
-    } else if (state == bs_sendSyn || (result != RESULT_OK && !firstRepetition)) {
+    } else if (state == bs_sendSyn || (result < RESULT_OK && !firstRepetition)) {
       logDebug(lf_bus, "notify request: %s", getResultCode(result));
       bool restart = m_currentRequest->notify(
         result == RESULT_ERR_SYN && (m_state == bs_recvCmdAck || m_state == bs_recvRes)
@@ -716,13 +742,13 @@ result_t DirectProtocolHandler::setState(BusState state, result_t result, bool f
 
   m_escape = 0;
   if (state == m_state) {
-    if (m_listener && result != RESULT_OK) {
+    if (m_listener && result < RESULT_OK && state != bs_noSignal) {
       m_listener->notifyProtocolStatus(m_listenerState, result);
     }
     return result;
   }
   if ((result < RESULT_OK && !(result == RESULT_ERR_TIMEOUT && state == bs_skip && m_state == bs_ready))
-      || (result != RESULT_OK && state == bs_skip && m_state != bs_ready)) {
+      || (result < RESULT_OK && state == bs_skip && m_state != bs_ready)) {
     logDebug(lf_bus, "%s during %s, switching to %s", getResultCode(result), getStateCode(m_state),
         getStateCode(state));
   } else if (m_currentRequest != nullptr || state == bs_sendCmd || state == bs_sendCmdCrc || state == bs_sendCmdAck
@@ -744,7 +770,7 @@ result_t DirectProtocolHandler::setState(BusState state, result_t result, bool f
     if (pstate == ps_idle && m_generateSynInterval == SYN_INTERVAL) {
       pstate = ps_idleSYN;
     }
-    if (result != RESULT_OK || pstate != m_listenerState) {
+    if (result < RESULT_OK || pstate != m_listenerState) {
       m_listener->notifyProtocolStatus(pstate, result);
       m_listenerState = pstate;
     }

src/lib/ebus/protocol_direct.h

@@ -65,8 +65,8 @@ class DirectProtocolHandler : public ProtocolHandler {
    * @param listener the @a ProtocolListener.
    */
   DirectProtocolHandler(const ebus_protocol_config_t config,
-      Device* device, ProtocolListener* listener)
+      CharDevice* device, ProtocolListener* listener)
-    : ProtocolHandler(config, device, listener),
+    : ProtocolHandler(config, device, listener), m_device(device),
       m_lockCount(config.lockCount <= 3 ? 3 : config.lockCount),
       m_remainLockCount(config.lockCount == 0 ? 1 : 0),
       m_generateSynInterval(config.generateSyn ? 10*getMasterNumber(config.ownAddress)+SYN_TIMEOUT : 0),
@@ -110,10 +110,24 @@ class DirectProtocolHandler : public ProtocolHandler {
 
  private:
   /**
-   * Handle the next symbol on the bus.
+   * Handle sending the next symbol to the bus.
-   * @return RESULT_OK on success, or an error code.
+   * @param recvTimeout pointer to a variable in which to put the timeout for the receive.
+   * @param sentSymbol pointer to a variable in which to put the sent symbol.
+   * @param sentTime pointer to a variable in which to put the system time when the symbol was sent.
+   * @return RESULT_OK on success, RESULT_CONTINUE when a symbol was sent, or an error code.
    */
-  result_t handleSymbol();
+  result_t handleSend(unsigned int* recvTimeout, symbol_t* sentSymbol, struct timespec* sentTime);
+
+  /**
+   * Handle receiving the next symbol from the bus.
+   * @param timeout the timeout for the receive.
+   * @param sending whether a symbol was sent before entry.
+   * @param sentSymbol the sent symbol to verify (if sending).
+   * @param sentTime pointer to a variable with the system time when the symbol was sent.
+   * @return RESULT_OK on success, RESULT_CONTINUE when further received symbols are buffered,
+   * or an error code.
+   */
+  result_t handleReceive(unsigned int timeout, bool sending, symbol_t sentSymbol, struct timespec* sentTime);
 
   /**
    * Set a new @a BusState and add a log message if necessary.
@@ -132,6 +146,9 @@ class DirectProtocolHandler : public ProtocolHandler {
    */
   void messageCompleted();
 
+  /** the @a CharDevice instance for accessing the bus. */
+  CharDevice* m_device;
+
   /** the number of AUTO-SYN symbols before sending is allowed after lost arbitration. */
   unsigned int m_lockCount;
 

src/lib/ebus/transport.cpp

@@ -0,0 +1,356 @@
+/*
+ * ebusd - daemon for communication with eBUS heating systems.
+ * Copyright (C) 2023 John Baier <ebusd@ebusd.eu>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifdef HAVE_CONFIG_H
+#  include <config.h>
+#endif
+
+#include "lib/ebus/transport.h"
+#include <fcntl.h>
+#include <sys/ioctl.h>
+#include <sys/file.h>
+#ifdef HAVE_LINUX_SERIAL
+#  include <linux/serial.h>
+#endif
+#ifdef HAVE_FREEBSD_UFTDI
+#  include <dev/usb/uftdiio.h>
+#endif
+#ifdef HAVE_PPOLL
+#  include <poll.h>
+#endif
+#include "lib/ebus/data.h"
+#include "lib/utils/tcpsocket.h"
+
+namespace ebusd {
+
+
+#define MTU 1540
+
+#ifndef POLLRDHUP
+#define POLLRDHUP 0
+#endif
+
+
+#ifdef DEBUG_RAW_TRAFFIC
+  #define DEBUG_RAW_TRAFFIC_HEAD(format, args...) fprintf(stdout, "%ld raw: " format, clockGetMillis(), args)
+  #define DEBUG_RAW_TRAFFIC_ITEM(args...) fprintf(stdout, args)
+  #define DEBUG_RAW_TRAFFIC_FINAL() fprintf(stdout, "\n"); fflush(stdout)
+  #undef DEBUG_RAW_TRAFFIC
+  #define DEBUG_RAW_TRAFFIC(format, args...) fprintf(stdout, "%ld raw: " format "\n", clockGetMillis(), args); fflush(stdout)
+#else
+  #define DEBUG_RAW_TRAFFIC_HEAD(format, args...)
+  #undef DEBUG_RAW_TRAFFIC_ITEM
+  #define DEBUG_RAW_TRAFFIC_FINAL()
+  #define DEBUG_RAW_TRAFFIC(format, args...)
+#endif
+
+
+
+FileTransport::FileTransport(const char* name, unsigned int latency, bool checkDevice)
+  : Transport(name, HOST_LATENCY_MS+latency),
+    m_checkDevice(checkDevice),
+    m_fd(-1),
+    m_bufSize(((MAX_LEN+1+3)/4)*4), m_bufLen(0) {
+  m_buffer = reinterpret_cast<symbol_t*>(malloc(m_bufSize));
+  if (!m_buffer) {
+    m_bufSize = 0;
+  }
+}
+
+FileTransport::~FileTransport() {
+  close();
+  if (m_buffer) {
+    free(m_buffer);
+    m_buffer = nullptr;
+  }
+}
+
+result_t FileTransport::open() {
+  close();
+  result_t result;
+  if (m_bufSize == 0) {
+    result = RESULT_ERR_DEVICE;
+  } else {
+    result = openInternal();
+  }
+  if (m_listener != nullptr) {
+    result = m_listener->notifyTransportStatus(result == RESULT_OK);
+  }
+  if (result != RESULT_OK) {
+    close();
+  }
+  return result;
+}
+
+void FileTransport::close() {
+  if (m_fd == -1) {
+    return;
+  }
+  ::close(m_fd);
+  m_fd = -1;
+  m_bufLen = 0;  // flush read buffer
+  if (m_listener != nullptr) {
+    m_listener->notifyTransportStatus(false);
+  }
+}
+
+bool FileTransport::isValid() {
+  if (m_fd == -1) {
+    return false;
+  }
+  if (m_checkDevice) {
+    checkDevice();
+  }
+  return m_fd != -1;
+}
+
+result_t FileTransport::write(const uint8_t* data, size_t len) {
+  if (!isValid()) {
+    return RESULT_ERR_DEVICE;
+  }
+#ifdef DEBUG_RAW_TRAFFIC_ITEM
+  DEBUG_RAW_TRAFFIC_HEAD("%ld >", len);
+  for (size_t pos=0; pos < len; pos++) {
+    DEBUG_RAW_TRAFFIC_ITEM(" %2.2x", data[pos]);
+  }
+  DEBUG_RAW_TRAFFIC_FINAL();
+#endif
+  return (::write(m_fd, data, len) == len) ? RESULT_OK : RESULT_ERR_DEVICE;
+}
+
+result_t FileTransport::read(unsigned int timeout, const uint8_t** data, size_t* len) {
+  if (!isValid()) {
+    return RESULT_ERR_DEVICE;
+  }
+  if (timeout == 0) {
+    if (m_bufLen > 0) {
+      *data = m_buffer;
+      *len = m_bufLen;
+      return RESULT_OK;
+    }
+    return RESULT_ERR_TIMEOUT;
+  }
+  if (timeout > 0) {
+    timeout += m_latency;
+    int ret;
+    struct timespec tdiff;
+
+    // set select timeout
+    tdiff.tv_sec = timeout/1000;
+    tdiff.tv_nsec = (timeout%1000)*1000000;
+
+#ifdef HAVE_PPOLL
+    nfds_t nfds = 1;
+    struct pollfd fds[nfds];
+
+    memset(fds, 0, sizeof(fds));
+
+    fds[0].fd = m_fd;
+    fds[0].events = POLLIN | POLLERR | POLLHUP | POLLRDHUP;
+    ret = ppoll(fds, nfds, &tdiff, nullptr);
+    if (ret >= 0 && fds[0].revents & (POLLERR | POLLHUP | POLLRDHUP)) {
+      ret = -1;
+    }
+#else
+#ifdef HAVE_PSELECT
+    fd_set readfds, exceptfds;
+
+    FD_ZERO(&readfds);
+    FD_ZERO(&exceptfds);
+    FD_SET(m_fd, &readfds);
+
+    ret = pselect(m_fd + 1, &readfds, nullptr, &exceptfds, &tdiff, nullptr);
+    if (ret >= 1 && FD_ISSET(m_fd, &exceptfds)) {
+      ret = -1;
+    }
+#else
+    ret = 1;  // ignore timeout if neither ppoll nor pselect are available
+#endif
+#endif
+    if (ret == -1) {
+      DEBUG_RAW_TRAFFIC("poll error %d", errno);
+      close();
+      return RESULT_ERR_DEVICE;
+    }
+    if (ret == 0) {
+      return RESULT_ERR_TIMEOUT;
+    }
+  }
+
+  // directly read byte from device
+  if (m_bufLen > 0 && m_bufLen > m_bufSize - m_bufSize / 4) {
+    // more than 3/4 of input buffer consumed is taken as signal that ebusd is too slow
+    m_bufLen = 0;
+    if (m_listener != nullptr) {
+      m_listener->notifyTransportMessage(true, "buffer overflow");
+    }
+  }
+  // fill up the buffer
+  ssize_t size = ::read(m_fd, m_buffer + m_bufLen, m_bufSize - m_bufLen);
+  if (size <= 0) {
+    return RESULT_ERR_TIMEOUT;
+  }
+#ifdef DEBUG_RAW_TRAFFIC_ITEM
+  DEBUG_RAW_TRAFFIC_HEAD("%ld+%ld <", m_bufLen, size);
+  for (int pos=0; pos < size; pos++) {
+    DEBUG_RAW_TRAFFIC_ITEM(" %2.2x", m_buffer[(m_bufLen+pos)%m_bufSize]);
+  }
+  DEBUG_RAW_TRAFFIC_FINAL();
+#endif
+  m_bufLen += size;
+  *data = m_buffer;
+  *len = m_bufLen;
+  return RESULT_OK;
+}
+
+void FileTransport::readConsumed(size_t len) {
+  if (len >= m_bufLen) {
+    m_bufLen = 0;
+  } else if (len > 0) {
+    size_t tail = m_bufLen - len;
+    memmove(m_buffer, m_buffer + len, tail);
+    DEBUG_RAW_TRAFFIC("move %ld @%ld to 0", tail, len);
+    m_bufLen = tail;
+  }
+}
+
+
+result_t SerialTransport::openInternal() {
+  struct termios newSettings;
+
+  // open file descriptor
+  m_fd = ::open(m_name, O_RDWR | O_NOCTTY | O_NDELAY);
+
+  if (m_fd < 0) {
+    return RESULT_ERR_NOTFOUND;
+  }
+  if (isatty(m_fd) == 0) {
+    close();
+    return RESULT_ERR_NOTFOUND;
+  }
+
+  if (flock(m_fd, LOCK_EX|LOCK_NB) != 0) {
+    close();
+    return RESULT_ERR_DEVICE;
+  }
+
+#ifdef HAVE_LINUX_SERIAL
+  struct serial_struct serial;
+  if (ioctl(m_fd, TIOCGSERIAL, &serial) == 0) {
+    serial.flags |= ASYNC_LOW_LATENCY;
+    ioctl(m_fd, TIOCSSERIAL, &serial);
+  }
+#endif
+
+#ifdef HAVE_FREEBSD_UFTDI
+  int param = 0;
+  // flush tx/rx and set low latency on uftdi device
+  if (ioctl(m_fd, UFTDIIOC_GET_LATENCY, &param) == 0) {
+    ioctl(m_fd, UFTDIIOC_RESET_IO, &param);
+    param = 1;
+    ioctl(m_fd, UFTDIIOC_SET_LATENCY, &param);
+  }
+#endif
+
+  // save current settings
+  tcgetattr(m_fd, &m_oldSettings);
+
+  // create new settings
+  memset(&newSettings, 0, sizeof(newSettings));
+
+#ifdef HAVE_CFSETSPEED
+  cfsetspeed(&newSettings, m_speed ? (m_speed > 1 ? B115200 : B9600) : B2400);
+#else
+  cfsetispeed(&newSettings, m_speed ? (m_speed > 1 ? B115200 : B9600) : B2400);
+  cfsetospeed(&newSettings, m_enhancedLevel ? (m_enhancedLevel >= el_speed ? B115200 : B9600) : B2400);
+#endif
+  newSettings.c_cflag |= (CS8 | CLOCAL | CREAD);
+  newSettings.c_lflag &= ~(ICANON | ECHO | ECHOE | ISIG);  // non-canonical mode
+  newSettings.c_iflag |= IGNPAR;  // ignore parity errors
+  newSettings.c_oflag &= ~OPOST;
+
+  // non-canonical mode: read() blocks until at least one byte is available
+  newSettings.c_cc[VMIN]  = 1;
+  newSettings.c_cc[VTIME] = 0;
+
+  // empty device buffer
+  tcflush(m_fd, TCIFLUSH);
+
+  // activate new settings of serial device
+  if (tcsetattr(m_fd, TCSANOW, &newSettings)) {
+    close();
+    return RESULT_ERR_DEVICE;
+  }
+
+  // set serial device into blocking mode
+  fcntl(m_fd, F_SETFL, fcntl(m_fd, F_GETFL) & ~O_NONBLOCK);
+
+  return RESULT_OK;
+}
+
+void SerialTransport::close() {
+  if (m_fd != -1) {
+    // empty device buffer
+    tcflush(m_fd, TCIOFLUSH);
+
+    // restore previous settings of the device
+    tcsetattr(m_fd, TCSANOW, &m_oldSettings);
+  }
+  FileTransport::close();
+}
+
+void SerialTransport::checkDevice() {
+  int cnt;
+  if (ioctl(m_fd, FIONREAD, &cnt) == -1) {
+    close();
+  }
+}
+
+result_t NetworkTransport::openInternal() {
+  m_fd = socketConnect(m_hostOrIp, m_port, m_udp, nullptr, 5, 2);  // wait up to 5 seconds for established connection
+  if (m_fd < 0) {
+    return RESULT_ERR_GENERIC_IO;
+  }
+  if (!m_udp) {
+    usleep(25000);  // wait 25ms for potential initial garbage
+  }
+  int cnt;
+  symbol_t buf[MTU];
+  int ioerr;
+  while ((ioerr=ioctl(m_fd, FIONREAD, &cnt)) >= 0 && cnt > 1) {
+    // skip buffered input
+    ssize_t read = ::read(m_fd, &buf, MTU);
+    if (read <= 0) {
+      break;
+    }
+  }
+  if (ioerr < 0) {
+    close();
+    return RESULT_ERR_GENERIC_IO;
+  }
+  return RESULT_OK;
+}
+
+void NetworkTransport::checkDevice() {
+  int cnt;
+  if (ioctl(m_fd, FIONREAD, &cnt) < 0) {
+    close();
+  }
+}
+
+}  // namespace ebusd

src/lib/ebus/transport.h

@@ -0,0 +1,334 @@
+/*
+ * ebusd - daemon for communication with eBUS heating systems.
+ * Copyright (C) 2023 John Baier <ebusd@ebusd.eu>
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#ifndef LIB_EBUS_TRANSPORT_H_
+#define LIB_EBUS_TRANSPORT_H_
+
+#include <unistd.h>
+#include <termios.h>
+#include <string>
+#include "lib/ebus/result.h"
+#include "lib/ebus/symbol.h"
+
+namespace ebusd {
+
+/** @file lib/ebus/transport.h
+ * Classes for low level transport to/from the eBUS device.
+ *
+ * A @a Transport is either a @a SerialTransport directly connected
+ * to a local tty port or a remote @a NetworkTransport handled via a
+ * socket.
+ */
+
+/** the transfer latency of the network device [ms]. */
+#define NETWORK_LATENCY_MS 30
+
+/** the latency of the host [ms]. */
+#if defined(__CYGWIN__) || defined(_WIN32)
+#define HOST_LATENCY_MS 20
+#else
+#define HOST_LATENCY_MS 10
+#endif
+
+/**
+ * Interface for listening to data received on/sent to a @a Transport.
+ */
+class TransportListener {
+ public:
+  /**
+   * Destructor.
+   */
+  virtual ~TransportListener() {}
+
+  /**
+   * Called to notify a status change from the @a Transport.
+   * @param opened true when the transport was successfully opened, false when it was closed or open failed.
+   * @return the result_t code (other than RESULT_OK if an extra open action was performed unsuccessfully).
+   */
+  virtual result_t notifyTransportStatus(bool opened) = 0;  // abstract
+
+  /**
+   * Called to notify a message from the @a Transport.
+   * @param error true for an error message, false for an info message.
+   * @param message the message string.
+   */
+  virtual void notifyTransportMessage(bool error, const char* message) = 0;  // abstract
+};
+
+
+/**
+ * The base class for low level transport to/from the eBUS device.
+ */
+class Transport {
+ protected:
+  /**
+   * Construct a new instance.
+   * @param name the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
+   */
+  Transport(const char* name, unsigned int latency)
+  : m_name(name), m_latency(latency), m_listener(nullptr) {}
+
+ public:
+  /**
+   * Destructor.
+   */
+  virtual ~Transport() { }
+
+  /**
+   * Get the device name.
+   * @return the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
+   */
+  const char* getName() const { return m_name; }
+
+  /**
+   * Get the transfer latency of this device.
+   * @return the transfer latency in milliseconds.
+   */
+  unsigned int getLatency() const { return m_latency; }
+
+  /**
+   * Get info about the transport as string.
+   * @return a @a string describing the transport.
+   */
+  virtual string getTransportInfo() const = 0;  // abstract
+
+  /**
+   * Set the @a TransportListener.
+   * @param listener the @a TransportListener.
+   */
+  void setListener(TransportListener* listener) { m_listener = listener; }
+
+  /**
+   * Open the transport.
+   * @return the @a result_t code.
+   */
+  virtual result_t open() = 0;  // abstract
+
+  /**
+   * Close the device if opened.
+   */
+  virtual void close() = 0;  // abstract
+
+  /**
+   * Return whether the device is opened and available.
+   * @return whether the device is opened and available.
+   */
+  virtual bool isValid() = 0;  // abstract
+
+  /**
+   * Write arbitrary data to the device.
+   * @param data the data to send.
+   * @param len the length of data.
+   * @return the @a result_t code.
+   */
+  virtual result_t write(const uint8_t* data, size_t len) = 0;  // abstract
+
+  /**
+   * Read data from the device.
+   * @param timeout maximum time to wait for the byte in milliseconds, or 0 for returning only already buffered data.
+   * @param data pointer to a variable in which to put the received data.
+   * @param len pointer to a variable in which to put the number of available bytes.
+   * @return the @a result_t code.
+   */
+  virtual result_t read(unsigned int timeout, const uint8_t** data, size_t* len) = 0;  // abstract
+
+  /**
+   * Needs to be called after @a read() in order to mark all or parts of the available
+   * bytes as consumed.
+   * @param len the number of bytes consumed.
+   */
+  virtual void readConsumed(size_t len) = 0;  // abstract
+
+ protected:
+  /**
+   * Internal method for opening the device. Called from @a open().
+   * @return the @a result_t code.
+   */
+  virtual result_t openInternal() = 0;  // abstract
+
+  /** the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network). */
+  const char* m_name;
+
+  /** the bus transfer latency in milliseconds. */
+  const unsigned int m_latency;
+
+  /** the @a TransportListener, or nullptr. */
+  TransportListener* m_listener;
+};
+
+
+/**
+ * The common base class for transport using a file descriptor.
+ */
+class FileTransport : public Transport {
+ protected:
+  /**
+   * Construct a new instance.
+   * @param name the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
+   * @param latency the bus transfer latency in milliseconds.
+   * @param checkDevice whether to regularly check the device availability.
+   */
+  FileTransport(const char* name, unsigned int latency, bool checkDevice);
+
+ public:
+  /**
+   * Destructor.
+   */
+  virtual ~FileTransport();
+
+  // @copydoc
+  result_t open() override;
+
+  // @copydoc
+  void close() override;
+
+  // @copydoc
+  bool isValid() override;
+
+  // @copydoc
+  result_t write(const uint8_t* data, size_t len) override;
+
+  // @copydoc
+  result_t read(unsigned int timeout, const uint8_t** data, size_t* len) override;
+
+  // @copydoc
+  void readConsumed(size_t len) override;
+
+ protected:
+  /**
+   * Check if the device is still available and close it if not.
+   */
+  virtual void checkDevice() = 0;  // abstract
+
+  /** whether to regularly check the device availability. */
+  const bool m_checkDevice;
+
+  /** the opened file descriptor, or -1. */
+  int m_fd;
+
+ private:
+  /** the receive buffer. */
+  symbol_t* m_buffer;
+
+  /** the receive buffer size (multiple of 4). */
+  size_t m_bufSize;
+
+  /** the receive buffer fill length. */
+  size_t m_bufLen;
+};
+
+
+/**
+ * The @a Transport for a directly connected serial interface (tty).
+ */
+class SerialTransport : public FileTransport {
+ public:
+  /**
+   * Construct a new instance.
+   * @param name the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
+   * @param extraLatency the extra bus transfer latency in milliseconds.
+   * @param checkDevice whether to regularly check the device availability.
+   * @param speed 0 for normal speed, 1 for 4x speed, or 2 for 48x speed.
+   */
+  SerialTransport(const char* name, unsigned int extraLatency, bool checkDevice, uint8_t speed)
+    : FileTransport(name, extraLatency, checkDevice), m_speed(speed) {
+  }
+
+  // @copydoc
+  string getTransportInfo() const override {
+    return m_speed ? (m_speed == 1 ? "serial speed" : "serial high speed") : "serial";
+  }
+
+  // @copydoc
+  result_t openInternal() override;
+
+  // @copydoc
+  void close() override;
+
+
+ protected:
+  // @copydoc
+  void checkDevice() override;
+
+
+ private:
+  /** the previous settings of the device for restoring. */
+  termios m_oldSettings;
+
+  /** 0 for normal speed, 1 for 4x speed, or 2 for 48x speed. */
+  const int m_speed;
+};
+
+
+/**
+ * The @a Transport for a remote network interface.
+ */
+class NetworkTransport : public FileTransport {
+ public:
+  /**
+   * Construct a new instance.
+   * @param name the device name (e.g. "/dev/ttyUSB0" for serial, "127.0.0.1:1234" for network).
+   * @param extraLatency the extra bus transfer latency in milliseconds.
+   * @param address the socket address of the device.
+   * @param hostOrIp the host name or IP address of the device.
+   * @param port the TCP or UDP port of the device.
+   * @param udp true for UDP, false to TCP.
+   */
+  NetworkTransport(const char* name, unsigned int extraLatency, const char* hostOrIp, uint16_t port,
+      bool udp)
+    : FileTransport(name, NETWORK_LATENCY_MS+extraLatency, true),
+    m_hostOrIp(hostOrIp), m_port(port), m_udp(udp) {}
+
+  /**
+   * Destructor.
+   */
+  ~NetworkTransport() override {
+    if (m_hostOrIp) {
+      free((void*)m_hostOrIp);
+      m_hostOrIp = nullptr;
+    }
+  }
+
+  // @copydoc
+  string getTransportInfo() const override {
+    return m_udp ? "UDP" : "TCP";
+  }
+
+  // @copydoc
+  result_t openInternal() override;
+
+
+ protected:
+  // @copydoc
+  void checkDevice() override;
+
+
+ private:
+  /** the host name or IP address of the device. */
+  const char* m_hostOrIp;
+
+  /** the TCP or UDP port of the device. */
+  const uint16_t m_port;
+
+  /** true for UDP, false to TCP. */
+  const bool m_udp;
+};
+
+}  // namespace ebusd
+
+#endif  // LIB_EBUS_TRANSPORT_H_