Protect event loop from user callback exceptions

src/core/DescriptorEventReceiver.cpp

@@ -171,6 +171,14 @@ namespace core {
         dispatchEvent();
     }
 
+    void DescriptorEventReceiver::onEventException(std::exception_ptr exceptionPtr) {
+        EventReceiver::onEventException(exceptionPtr);
+
+        if (isEnabled()) {
+            disable();
+        }
+    }
+
     void DescriptorEventReceiver::onSignal(int signum) {
         signalEvent(signum);
     }

src/core/DescriptorEventReceiver.h

@@ -52,6 +52,7 @@ namespace core {
 
 #include "utils/Timeval.h"
 
+#include <exception>
 #include <string>
 
 #endif /* DOXYGEN_SHOULD_SKIP_THIS */
@@ -109,6 +110,8 @@ namespace core {
 
         void checkTimeout(const utils::Timeval& currentTime);
 
+        void onEventException(std::exception_ptr exceptionPtr) override;
+
     private:
         void onEvent(const utils::Timeval& currentTime) final;
         void onSignal(int signum);

src/core/Event.cpp

@@ -47,6 +47,8 @@
 
 #ifndef DOXYGEN_SHOULD_SKIP_THIS
 
+#include <exception>
+
 #endif /* DOXYGEN_SHOULD_SKIP_THIS */
 
 namespace core {
@@ -81,7 +83,11 @@ namespace core {
 
     void Event::dispatch(const utils::Timeval& currentTime) {
         published = false;
-        eventReceiver->onEvent(currentTime);
+        try {
+            eventReceiver->onEvent(currentTime);
+        } catch (...) {
+            eventReceiver->onEventException(std::current_exception());
+        }
     }
 
     EventReceiver* Event::getEventReceiver() const {

src/core/EventReceiver.cpp

@@ -43,6 +43,10 @@
 
 #ifndef DOXYGEN_SHOULD_SKIP_THIS
 
+#include "log/Logger.h"
+
+#include <exception>
+
 #endif /* DOXYGEN_SHOULD_SKIP_THIS */
 
 namespace core {
@@ -76,6 +80,18 @@ namespace core {
         delete this;
     }
 
+    void EventReceiver::onEventException(std::exception_ptr exceptionPtr) {
+        try {
+            if (exceptionPtr != nullptr) {
+                std::rethrow_exception(exceptionPtr);
+            }
+        } catch (const std::exception& ex) {
+            LOG(ERROR) << getName() << ": Unhandled exception in event callback: " << ex.what();
+        } catch (...) {
+            LOG(ERROR) << getName() << ": Unhandled non-standard exception in event callback";
+        }
+    }
+
     void EventReceiver::span() {
         event.span();
     }

src/core/EventReceiver.h

@@ -46,6 +46,7 @@
 
 #ifndef DOXYGEN_SHOULD_SKIP_THIS
 
+#include <exception>
 #include <functional>
 
 namespace utils {
@@ -80,6 +81,7 @@ namespace core {
         void relax();
 
         virtual void onEvent(const utils::Timeval& currentTime) = 0;
+        virtual void onEventException(std::exception_ptr exceptionPtr);
 
         const std::string& getName() const;
 

src/core/socket/stream/SocketReader.cpp

@@ -79,6 +79,11 @@ namespace core::socket::stream {
         }
     }
 
+    void SocketReader::onEventException(std::exception_ptr exceptionPtr) {
+        core::eventreceiver::ReadEventReceiver::onEventException(exceptionPtr);
+        onStatus(EIO);
+    }
+
     void SocketReader::signalEvent([[maybe_unused]] int sigNum) { // Do nothing in case a signal was received
     }
 

src/core/socket/stream/SocketReader.h

@@ -48,6 +48,7 @@
 
 #include "utils/Timeval.h"
 
+#include <exception>
 #include <cstddef>
 #include <functional>
 #include <string>
@@ -76,6 +77,7 @@ namespace core::socket::stream {
         virtual void onReceivedFromPeer(std::size_t available) = 0;
 
         void readEvent() final;
+        void onEventException(std::exception_ptr exceptionPtr) override;
 
         std::size_t doRead();
         void signalEvent(int sigNum) final;

src/core/socket/stream/SocketWriter.cpp

@@ -84,6 +84,11 @@ namespace core::socket::stream {
         }
     }
 
+    void SocketWriter::onEventException(std::exception_ptr exceptionPtr) {
+        core::eventreceiver::WriteEventReceiver::onEventException(exceptionPtr);
+        onStatus(EIO);
+    }
+
     ssize_t SocketWriter::write(const char* chunk, std::size_t chunkLen) {
         return core::system::send(this->getRegisteredFd(), chunk, chunkLen, MSG_NOSIGNAL);
     }

src/core/socket/stream/SocketWriter.h

@@ -52,6 +52,7 @@ namespace core::pipe {
 
 #include "utils/Timeval.h"
 
+#include <exception>
 #include <cstddef>
 #include <functional>
 #include <string>
@@ -79,6 +80,7 @@ namespace core::socket::stream {
     private:
         void writeEvent() final;
         void signalEvent(int sigNum) final;
+        void onEventException(std::exception_ptr exceptionPtr) override;
 
         void doWrite();