Add Session::ClosedHandler (#99)

A callback function to signal that the endpoint has closed its connection.

Add this as an optional argument to Session::bind() and Session::startProcessingMessages().

Bug: #98

include/dap/session.h

@@ -137,6 +137,10 @@ class Session {
   // errors.
   using ErrorHandler = std::function<void(const char*)>;
 
+  // ClosedHandler is the type of callback function used to signal that a
+  // connected endpoint has closed.
+  using ClosedHandler = std::function<void()>;
+
   // create() constructs and returns a new Session.
   static std::unique_ptr<Session> create();
 
@@ -205,9 +209,13 @@ class Session {
 
   // bind() connects this Session to an endpoint using connect(), and then
   // starts processing incoming messages with startProcessingMessages().
-  inline void bind(const std::shared_ptr<Reader>&,
-                   const std::shared_ptr<Writer>&);
-  inline void bind(const std::shared_ptr<ReaderWriter>&);
+  // onClose is the optional callback which will be called when the session
+  // endpoint has been closed.
+  inline void bind(const std::shared_ptr<Reader>& reader,
+                   const std::shared_ptr<Writer>& writer,
+                   const ClosedHandler& onClose);
+  inline void bind(const std::shared_ptr<ReaderWriter>& readerWriter,
+                   const ClosedHandler& onClose);
 
   //////////////////////////////////////////////////////////////////////////////
   // Note:
@@ -227,9 +235,11 @@ class Session {
 
   // startProcessingMessages() starts a new thread to receive and dispatch
   // incoming messages.
+  // onClose is the optional callback which will be called when the session
+  // endpoint has been closed.
   // Note: This method is used for explicit control over message handling.
   //       Most users will use bind() instead of calling this method directly.
-  virtual void startProcessingMessages() = 0;
+  virtual void startProcessingMessages(const ClosedHandler& onClose = {}) = 0;
 
   // getPayload() blocks until the next incoming message is received, returning
   // the payload or an empty function if the connection was lost. The returned
@@ -423,13 +433,15 @@ void Session::connect(const std::shared_ptr<ReaderWriter>& rw) {
 }
 
 void Session::bind(const std::shared_ptr<dap::Reader>& r,
-                   const std::shared_ptr<dap::Writer>& w) {
+                   const std::shared_ptr<dap::Writer>& w,
+                   const ClosedHandler& onClose = {}) {
   connect(r, w);
-  startProcessingMessages();
+  startProcessingMessages(onClose);
 }
 
-void Session::bind(const std::shared_ptr<ReaderWriter>& rw) {
-  bind(rw, rw);
+void Session::bind(const std::shared_ptr<ReaderWriter>& rw,
+                   const ClosedHandler& onClose = {}) {
+  bind(rw, rw, onClose);
 }
 
 }  // namespace dap

src/session.cpp

@@ -65,7 +65,7 @@ class Impl : public dap::Session {
   void connect(const std::shared_ptr<dap::Reader>& r,
                const std::shared_ptr<dap::Writer>& w) override {
     if (isBound.exchange(true)) {
-      handlers.error("Session is already bound!");
+      handlers.error("Session::connect called twice");
       return;
     }
 
@@ -73,13 +73,21 @@ class Impl : public dap::Session {
     writer = dap::ContentWriter(w);
   }
 
-  void startProcessingMessages() override {
-    recvThread = std::thread([this] {
+  void startProcessingMessages(
+      const ClosedHandler& onClose /* = {} */) override {
+    if (isProcessingMessages.exchange(true)) {
+      handlers.error("Session::startProcessingMessages() called twice");
+      return;
+    }
+    recvThread = std::thread([this, onClose] {
       while (reader.isOpen()) {
         if (auto payload = getPayload()) {
           inbox.put(std::move(payload));
         }
       }
+      if (onClose) {
+        onClose();
+      }
     });
 
     dispatchThread = std::thread([this] {
@@ -471,6 +479,7 @@ class Impl : public dap::Session {
   }
 
   std::atomic<bool> isBound = {false};
+  std::atomic<bool> isProcessingMessages = {false};
   dap::ContentReader reader;
   dap::ContentWriter writer;
 

src/session_test.cpp

@@ -579,3 +579,47 @@ TEST_F(SessionTest, Concurrency) {
   client.reset();
   server.reset();
 }
+
+TEST_F(SessionTest, OnClientClosed) {
+  std::mutex mutex;
+  std::condition_variable cv;
+  bool clientClosed = false;
+
+  auto client2server = dap::pipe();
+  auto server2client = dap::pipe();
+
+  client->bind(server2client, client2server);
+  server->bind(client2server, server2client, [&] {
+    std::unique_lock<std::mutex> lock(mutex);
+    clientClosed = true;
+    cv.notify_all();
+  });
+
+  client.reset();
+
+  // Wait for the client closed handler to be called.
+  std::unique_lock<std::mutex> lock(mutex);
+  cv.wait(lock, [&] { return static_cast<bool>(clientClosed); });
+}
+
+TEST_F(SessionTest, OnServerClosed) {
+  std::mutex mutex;
+  std::condition_variable cv;
+  bool serverClosed = false;
+
+  auto client2server = dap::pipe();
+  auto server2client = dap::pipe();
+
+  client->bind(server2client, client2server, [&] {
+    std::unique_lock<std::mutex> lock(mutex);
+    serverClosed = true;
+    cv.notify_all();
+  });
+  server->bind(client2server, server2client);
+
+  server.reset();
+
+  // Wait for the client closed handler to be called.
+  std::unique_lock<std::mutex> lock(mutex);
+  cv.wait(lock, [&] { return static_cast<bool>(serverClosed); });
+}