Add on_error callback to CanDevice

ci/gitlab/normal.yml

@@ -71,17 +71,29 @@ Pytest on Windows:
     - win2025-container
   needs:
     - Compile on Windows
+  before_script:
+    - Start-Process toxiproxy-server.exe
   script:
     - python -m pytest
 
-Pytest on Linux:
+Pytest on Linux (Only socketcan):
   stage: unstagged
   tags:
     - linux-socketcan
   needs:
     - Compile on Linux
   script:
-    - python -m pytest
+    - python -m pytest test/python/test_socketcan.py
+
+Pytest on Linux (All except socketcan):
+  stage: unstagged
+  image: $ALMA_LATEST_IMAGE
+  needs:
+    - Compile on Linux
+  before_script:
+    - toxiproxy-server >/dev/null 2>&1 &
+  script:
+    - python -m pytest --ignore=test/python/test_socketcan.py
 
 
 Run Sanity Checks:

python/canmodule/__init__.pyi

@@ -44,6 +44,7 @@ class CanDeviceArguments:
         self,
         config: CanDeviceConfiguration,
         receiver: typing.Callable[[CanFrame], None],
+        on_error: typing.Callable[[CanReturnCode], None],
     ) -> None: ...
     @property
     def config(self) -> CanDeviceConfiguration: ...

src/include/CanDevice.h

@@ -135,7 +135,40 @@ struct CanDevice {
     LOG(Log::DBG, CanLogIt::h()) << "Received CAN frame: " << frame;
     if (m_args.receiver != nullptr) {
       LOG(Log::DBG, CanLogIt::h()) << "Calling receiver function";
-      m_args.receiver(frame);
+      try {
+        m_args.receiver(frame);
+      } catch (const std::exception& e) {
+        LOG(Log::ERR, CanLogIt::h())
+            << "Exception in receiver function: " << e.what();
+      } catch (...) {
+        LOG(Log::ERR, CanLogIt::h())
+            << "Unknown exception in receiver function";
+      }
+    }
+  }
+
+  /**
+   * @brief Handles errors.
+   *
+   * This function is called whenever an error occurs on the CAN device.
+   * It passes the received return code to the on_error function specified in
+   * the CanDeviceArguments object.
+   *
+   * @param code The return code received
+   */
+  inline void notify_error(CanReturnCode code) const noexcept {
+    LOG(Log::WRN, CanLogIt::h()) << "CAN device error: " << code;
+    if (m_args.on_error != nullptr) {
+      LOG(Log::DBG, CanLogIt::h()) << "Calling on_error function";
+      try {
+        m_args.on_error(code);
+      } catch (const std::exception& e) {
+        LOG(Log::ERR, CanLogIt::h())
+            << "Exception in on_error function: " << e.what();
+      } catch (...) {
+        LOG(Log::ERR, CanLogIt::h())
+            << "Unknown exception in on_error function";
+      }
     }
   }
 

src/include/CanDeviceArguments.h

@@ -35,6 +35,17 @@ struct CanDeviceArguments {
    * to a CanFrame object and returns void.
    */
   const std::function<void(const CanFrame&)> receiver;
+
+  /**
+   * @brief Callback function to handle an error.
+   *
+   * This function is called whenever an error happens, receiving the return
+   * code.
+   *
+   * @param on_error A function, lambda or functor that takes a CanReturnCode
+   * and returns void.
+   */
+  const std::function<void(CanReturnCode)> on_error = nullptr;
 };
 
 #endif  // SRC_INCLUDE_CANDEVICEARGUMENTS_H_

src/include/CanVendorAnagate.h

@@ -1,9 +1,11 @@
 #ifndef SRC_INCLUDE_CANVENDORANAGATE_H_
 #define SRC_INCLUDE_CANVENDORANAGATE_H_
 
+#include <atomic>
 #include <cstdint>
 #include <map>
 #include <mutex>  //NOLINT
+#include <thread>
 
 #include "AnaGateDllCan.h"
 #include "CanDevice.h"
@@ -36,9 +38,18 @@ struct CanVendorAnagate : CanDevice {
   static std::map<int, CanVendorAnagate*> m_handles;
 
   AnaInt32 m_handle{0};
+
+  // A separate thread periodically checks the connection health using the
+  // ALIVE mechanism
+  // (https://www.anagate.de/download/Manual-AnaGateAPI2-en.pdf)
+  CanReturnCode start_alive() noexcept;
+  void alive_monitor() noexcept;
+  std::thread m_alive_thread;
+  std::atomic<bool> m_alive_run{false};
 };
 
 namespace AnagateConstants {
+constexpr int connected = 3;
 constexpr int extendedId = 1 << 0;
 constexpr int remoteRequest = 1 << 1;
 constexpr char emptyMessage[8] = {0};

src/main/CanVendorAnagate.cpp

@@ -126,7 +126,11 @@ CanReturnCode CanVendorAnagate::vendor_open() noexcept {
   switch (r) {
     case AnagateConstants::errorNone:
       CanVendorAnagate::m_handles[m_handle] = this;
-      return CanReturnCode::success;
+
+      if (args().on_error != nullptr)
+        return start_alive();
+      else
+        return CanReturnCode::success;
     case AnagateConstants::errorOpenMaxConn:
       return CanReturnCode::too_many_connections;
     case AnagateConstants::errorTcpipSocket:
@@ -368,6 +372,11 @@ CanReturnCode CanVendorAnagate::vendor_close() noexcept {
     return CanReturnCode::success;
   }  // Already closed
 
+  // Stop the alive thread
+  m_alive_run = false;
+  if (m_alive_thread.joinable()) {
+    m_alive_thread.join();
+  }
   const int r = CANCloseDevice(m_handle);
   print_anagate_error(r);
   std::lock_guard<std::mutex> guard(CanVendorAnagate::m_handles_lock);
@@ -403,3 +412,57 @@ void CanVendorAnagate::print_anagate_error(AnaUInt32 r) noexcept {
     LOG(Log::ERR, CanLogIt::h()) << "ANAGATE ERROR: " << error_string;
   }
 }
+
+/**
+ * @brief Starts the ALIVE mechanism
+ *
+ * For more info check:
+ * (https://www.anagate.de/download/Manual-AnaGateAPI2-en.pdf)
+ */
+CanReturnCode CanVendorAnagate::start_alive() noexcept {
+  AnaInt32 r{0};
+
+  r = CANStartAlive(
+      m_handle,
+      args().config.timeout.value_or(AnagateConstants::defaultTimeout) / 1000);
+
+  if (r != 0) {
+    LOG(Log::ERR, CanLogIt::h()) << "Failed to start ALIVE mechanism";
+    vendor_close();
+    return CanReturnCode::internal_api_error;
+  }
+
+  m_alive_run = true;
+  m_alive_thread = std::thread(&CanVendorAnagate::alive_monitor, this);
+
+  return CanReturnCode::success;
+}
+
+/**
+ * @brief Monitors the AnaGate connection state on a background thread.
+ *
+ * Runs in a loop while m_alive_run is true, polling the connection
+ * state once per second via CANDeviceConnectState. Logs and notifies
+ * on disconnect via notify_error(CanReturnCode::disconnected).
+ *
+ * For more info check:
+ * (https://www.anagate.de/download/Manual-AnaGateAPI2-en.pdf)
+ */
+void CanVendorAnagate::alive_monitor() noexcept {
+  bool was_connected = true;
+  AnaInt32 state{0};
+  while (m_alive_run) {
+    std::this_thread::sleep_for(std::chrono::seconds(1));
+    if (!m_alive_run) break;
+
+    state = CANDeviceConnectState(m_handle);
+    bool connected = (state == AnagateConstants::connected);
+
+    // Only notify when it changes (not every poll when disconnected)
+    if (was_connected && !connected) {
+      LOG(Log::ERR, CanLogIt::h()) << "AnaGate connection lost";
+      notify_error(CanReturnCode::disconnected);
+    }
+    was_connected = connected;
+  }
+}

src/main/CanVendorSocketCan.cpp

@@ -108,7 +108,7 @@ CanReturnCode CanVendorSocketCan::vendor_open() noexcept {
     return CanReturnCode::internal_api_error;
   }
 
-  if (args().receiver != nullptr) {
+  if (args().receiver != nullptr || args().on_error != nullptr) {
     LOG(Log::DBG, CanLogIt::h()) << "Starting CAN subscriber thread";
     // Create epoll instance
     m_epoll_fd = epoll_create1(0);
@@ -122,7 +122,8 @@ CanReturnCode CanVendorSocketCan::vendor_open() noexcept {
 
     // Add the socket to the epoll instance
     struct epoll_event ev;
-    ev.events = EPOLLIN;
+    ev.events = EPOLLIN | EPOLLERR | EPOLLHUP;
+
     ev.data.fd = m_socket_fd;
     if (epoll_ctl(m_epoll_fd, EPOLL_CTL_ADD, m_socket_fd, &ev) < 0) {
       ::close(m_epoll_fd);
@@ -384,16 +385,24 @@ int CanVendorSocketCan::subscriber() noexcept {
       } else {
         LOG(Log::ERR, CanLogIt::h())
             << "Error occurred during epoll_wait: " << strerror(errno);
+        notify_error(CanReturnCode::internal_api_error);
         return -1;  // Error occurred
       }
     }
 
     if (nfds > 0 && events[0].data.fd == m_socket_fd) {
+      if (events[0].events & (EPOLLHUP | EPOLLERR)) {
+        LOG(Log::ERR, CanLogIt::h()) << "SocketCAN interface closed/errored";
+        notify_error(CanReturnCode::disconnected);
+        return -1;
+      }
+
       struct can_frame canFrame;
       int nbytes = ::read(m_socket_fd, &canFrame, sizeof(struct can_frame));
-      if (nbytes < 0) {
+      if (nbytes <= 0) {
         LOG(Log::ERR, CanLogIt::h())
             << "Unexpected error reading from socket, exiting";
+        notify_error(CanReturnCode::disconnected);
         return -1;
       }
 

src/main/CanVendorSocketCanSystec.cpp

@@ -43,7 +43,8 @@ CanVendorSocketCanSystec::CanVendorSocketCanSystec(
                        // set to a non-zero value.
 
   m_can_vendor_socketcan = CanDevice::create(
-      "socketcan", CanDeviceArguments{config, filter_busoff_callback});
+      "socketcan",
+      CanDeviceArguments{config, filter_busoff_callback, args.on_error});
 }
 
 /**

src/python/CanModule.cpp

@@ -69,8 +69,10 @@ PYBIND11_MODULE(canmodule, m) {
 
   py::class_<CanDeviceArguments>(m, "CanDeviceArguments")
       .def(py::init<const CanDeviceConfiguration&,
-                    const std::function<void(const CanFrame&)>&>(),
-           py::arg("config"), py::arg("receiver") = nullptr)
+                    const std::function<void(const CanFrame&)>&,
+                    const std::function<void(CanReturnCode)>&>(),
+           py::arg("config"), py::arg("receiver") = nullptr,
+           py::arg("on_error") = nullptr)
       .def_readonly("config", &CanDeviceArguments::config);
 
   py::class_<CanDeviceConfiguration>(m, "CanDeviceConfiguration")

test/cpp/CanDevice_test.cpp

@@ -3,9 +3,32 @@
 #include <gtest/gtest.h>
 
 #include <iostream>
+#include <stdexcept>
 #include <thread>  // NOLINT
 #include <vector>
 
+struct TestableCanDevice : CanDevice {
+  TestableCanDevice(std::string_view vendor_name,
+                    const CanDeviceArguments& args) noexcept
+      : CanDevice(vendor_name, args) {}
+
+  using CanDevice::notify_error;
+  using CanDevice::received;
+
+  CanReturnCode vendor_open() noexcept override {
+    return CanReturnCode::success;
+  }
+  CanReturnCode vendor_close() noexcept override {
+    return CanReturnCode::success;
+  }
+  CanReturnCode vendor_send(const CanFrame&) noexcept override {
+    return CanReturnCode::success;
+  }
+  CanDiagnostics vendor_diagnostics() noexcept override {
+    return CanDiagnostics{};
+  }
+};
+
 // Test fixture for CanFrame
 class CanDeviceTest : public ::testing::Test {
  protected:
@@ -52,3 +75,45 @@ TEST_F(CanDeviceTest, LoopbackDeviceMessageTransmission) {
               inFrames[i].is_remote_request());
   }
 }
+
+// Test for error notification
+TEST_F(CanDeviceTest, OnErrorCallbackIsInvoked) {
+  CanReturnCode captured_code = CanReturnCode::success;
+  bool called = false;
+  auto on_error_cb = [&](CanReturnCode code) {
+    called = true;
+    captured_code = code;
+  };
+
+  TestableCanDevice device{
+      "test", CanDeviceArguments{CanDeviceConfiguration{"dummy"},
+                                 [](const CanFrame&) {}, on_error_cb}};
+  device.notify_error(CanReturnCode::disconnected);
+
+  ASSERT_TRUE(called);
+  ASSERT_EQ(captured_code, CanReturnCode::disconnected);
+}
+
+TEST_F(CanDeviceTest, ThrowingCallbacksAreHandled) {
+  bool receiver_called = false;
+  auto receiver_cb = [&](const CanFrame&) {
+    receiver_called = true;
+    throw std::runtime_error("error");
+  };
+
+  bool on_error_called = false;
+  auto on_error_cb = [&](CanReturnCode) {
+    on_error_called = true;
+    throw std::runtime_error("error");
+  };
+
+  TestableCanDevice device{
+      "test", CanDeviceArguments{CanDeviceConfiguration{"dummy"}, receiver_cb,
+                                 on_error_cb}};
+
+  ASSERT_NO_THROW(device.received(CanFrame{0}));
+  ASSERT_TRUE(receiver_called);
+
+  ASSERT_NO_THROW(device.notify_error(CanReturnCode::disconnected));
+  ASSERT_TRUE(on_error_called);
+}