Client transport error handling

include/canbus/isotp_fast.h

@@ -113,6 +113,14 @@ struct isotp_fast_opts
     uint8_t stmin;
     uint8_t flags;
     enum isotp_fast_addressing_mode addressing_mode;
+    /**
+     * Optional override for the RX filter mask installed by
+     * @ref isotp_fast_bind. 0 means the default mask for the addressing
+     * mode is used. Use this e.g. to additionally match the sender address
+     * so that concurrent contexts bound to different peers on one interface
+     * do not receive each other's traffic
+     */
+    uint32_t rx_mask;
 };
 
 /**

include/thingset++/can/zephyr/ThingSetZephyrCanInterface.hpp

@@ -49,6 +49,7 @@ class ThingSetZephyrCanInterface : public _ThingSetZephyrCanInterface
 
     AddressClaimWorkItem _addressClaimWork;
     k_event _events;
+    k_mutex _bindLock;
     int _claimFilterId;
     int _discoverFilterId;
 

include/thingset++/can/zephyr/ThingSetZephyrCanRequestResponseContext.hpp

@@ -52,6 +52,7 @@ class ThingSetZephyrCanRequestResponseContext {
     static void onRequestResponseReceived(net_buf *buffer, int remainingLength, isotp_fast_addr address, void *arg);
     void onRequestResponseReceived(net_buf *buffer, int remainingLength, isotp_fast_addr address);
     static const isotp_fast_opts flowControlOptions;
+    static const isotp_fast_opts peerFlowControlOptions;
 };
 
 } // namespace ThingSet::Can::Zephyr

src/ThingSetClient.cpp

@@ -9,6 +9,9 @@
 
 namespace ThingSet {
 
+static constexpr uint8_t cborNull = 0xF6;
+static constexpr size_t responseHeaderSize = 2; /* status code + CBOR null */
+
 ThingSetClient::ThingSetClient(ThingSetClientTransport &transport, uint8_t *rxBuffer, size_t rxBufferSize,
                                          uint8_t *txBuffer, size_t txBufferSize)
     : _transport(transport), _rxBuffer(rxBuffer), _rxBufferSize(rxBufferSize), _txBuffer(txBuffer),
@@ -22,13 +25,17 @@ bool ThingSetClient::connect()
 
 ThingSetResult ThingSetClient::read(uint8_t **responseBuffer, size_t &responseSize)
 {
-    responseSize = _transport.read(_rxBuffer, _rxBufferSize);
-    if (responseSize == 0) {
-        return ThingSetResult(ThingSetStatusCode::internalServerError);
+    responseSize = 0;
+
+    int received = _transport.read(_rxBuffer, _rxBufferSize);
+    if (received <= 0) {
+        // No response (0) or a transport error such as a receive timeout
+        // (negative errno). The rx buffer may still hold a previous response
+        return ThingSetResult(ThingSetStatusCode::gatewayTimeout);
     }
 
 #ifdef DEBUG_LOGGING
-    for (size_t i = 0; i < responseSize; i++)
+    for (int i = 0; i < received; i++)
     {
         if (i > 0 && i % 16 == 0) {
             printf("\n");
@@ -43,14 +50,14 @@ ThingSetResult ThingSetClient::read(uint8_t **responseBuffer, size_t &responseSi
         return result;
     }
 
-    // first value is always a CBOR null
-    if (_rxBuffer[1] != 0xF6) {
+    // a successful response carries at least the status code plus a CBOR null
+    if ((size_t)received < responseHeaderSize || _rxBuffer[1] != cborNull) {
         return ThingSetResult(ThingSetStatusCode::internalServerError);
     }
 
     // return size having accounted for response code and null
-    responseSize -= 2;
-    *responseBuffer = &_rxBuffer[2];
+    responseSize = (size_t)received - responseHeaderSize;
+    *responseBuffer = &_rxBuffer[responseHeaderSize];
 
     return result;
 }

src/can/zephyr/ThingSetZephyrCanInterface.cpp

@@ -50,6 +50,7 @@ ThingSetZephyrCanInterface::ThingSetZephyrCanInterface(const device *const canDe
 {
     k_work_init(&_addressClaimWork.work, addressClaimWorkHandler);
     _addressClaimWork.instance = this;
+    k_mutex_init(&_bindLock);
 }
 
 ThingSetZephyrCanInterface::~ThingSetZephyrCanInterface()
@@ -141,8 +142,27 @@ int ThingSetZephyrCanInterface::addFilter(CanID &canId, void (*callback)(const d
     return can_add_rx_filter(_canDevice, callback, this, &filter);
 }
 
+namespace {
+struct MutexGuard
+{
+    k_mutex &_mutex;
+
+    explicit MutexGuard(k_mutex &mutex) : _mutex(mutex)
+    {
+        k_mutex_lock(&_mutex, K_FOREVER);
+    }
+
+    ~MutexGuard()
+    {
+        k_mutex_unlock(&_mutex);
+    }
+};
+} // namespace
+
 bool ThingSetZephyrCanInterface::bind(uint8_t nodeAddress)
 {
+    MutexGuard guard(_bindLock);
+
     if (_nodeAddress == CanID::broadcastAddress) {
         _nodeAddress = nodeAddress;
         LOG_INFO("Starting address claim for CAN interface %s", _canDevice->name);

src/can/zephyr/ThingSetZephyrCanRequestResponseContext.cpp

@@ -35,6 +35,16 @@ const isotp_fast_opts ThingSetZephyrCanRequestResponseContext::flowControlOption
     .addressing_mode = ISOTP_FAST_ADDRESSING_MODE_FIXED,
 };
 
+const isotp_fast_opts ThingSetZephyrCanRequestResponseContext::peerFlowControlOptions = {
+    .bs = 8,
+    .stmin = CONFIG_THINGSET_PLUS_PLUS_CAN_FRAME_SEPARATION_TIME,
+#ifdef CONFIG_CAN_FD_MODE
+    .flags = ISOTP_MSG_FDF,
+#endif
+    .addressing_mode = ISOTP_FAST_ADDRESSING_MODE_FIXED,
+    .rx_mask = ISOTP_FIXED_ADDR_RX_MASK | ISOTP_FIXED_ADDR_SA_MASK,
+};
+
 static void onRequestResponseError(int8_t error, isotp_fast_addr addr, void *arg);
 static void onRequestResponseSent(int result, isotp_fast_addr addr, void *arg);
 
@@ -68,13 +78,22 @@ bool ThingSetZephyrCanRequestResponseContext::bind(uint8_t otherNodeAddress, std
                       .setMessageType(MessageType::requestResponse)
                       .setMessagePriority(MessagePriority::channel)
                       .setTarget(_canInterface.getNodeAddress());
+    const isotp_fast_opts *options = &ThingSetZephyrCanRequestResponseContext::flowControlOptions;
     if (otherNodeAddress != CanID::broadcastAddress) {
         canId.setSource(otherNodeAddress);
+        /* bound to one peer: only accept traffic from that peer */
+        options = &ThingSetZephyrCanRequestResponseContext::peerFlowControlOptions;
     }
     _inboundRequestCallback = callback;
-    return isotp_fast_bind(&_requestResponseContext, _canInterface.getDevice(), IsoTpFastAddress(canId),
-                           &ThingSetZephyrCanRequestResponseContext::flowControlOptions, onRequestResponseReceived,
-                           this, onRequestResponseError, onRequestResponseSent) == 0;
+    int result = isotp_fast_bind(&_requestResponseContext, _canInterface.getDevice(), IsoTpFastAddress(canId),
+                                 options, onRequestResponseReceived, this, onRequestResponseError,
+                                 onRequestResponseSent);
+    if (result != 0) {
+        LOG_ERROR("Failed to bind request/response context for node 0x%x (err %d)", otherNodeAddress, result);
+        _requestResponseContext.filter_id = THINGSET_PLUS_PLUS_ZEPHYR_CAN_FILTER_ID_NONE;
+        return false;
+    }
+    return true;
 }
 
 bool ThingSetZephyrCanRequestResponseContext::send(const uint8_t otherNodeAddress, uint8_t *buffer, size_t len)