added shared memrory transport to match other transports design patterns

Author: rianna-parla

CMakeLists.txt

@@ -44,6 +44,7 @@ add_library(mcp_cpp STATIC
     src/mcp/JSONParser.cpp
     src/mcp/StdioTransport.cpp
     src/mcp/InMemoryTransport.cpp
+    src/mcp/SharedMemoryTransport.cpp
     src/mcp/HTTPTransport.cpp
     src/mcp/HTTPServer.cpp
     src/mcp/Client.cpp
@@ -81,7 +82,8 @@ target_include_directories(mcp_cpp PUBLIC ${BOOST_INCLUDE_DIR})
 target_compile_definitions(mcp_cpp PUBLIC
   BOOST_ERROR_CODE_HEADER_ONLY
   BOOST_SYSTEM_NO_DEPRECATED
-  BOOST_ASIO_NO_DEPRECATED)
+  BOOST_ASIO_NO_DEPRECATED
+  BOOST_DATE_TIME_NO_LIB)
 
 target_link_libraries(mcp_cpp PUBLIC OpenSSL::SSL OpenSSL::Crypto)
 

examples/mcp_server/main.cpp

@@ -47,8 +47,10 @@ int main() {
 
     // Prepare waitable completion and handler (fires on EOF or transport error)
     std::promise<void> stopped;
-    server->SetErrorHandler([&stopped](const std::string& err) {
+    server->SetErrorHandler([&stopped, &server](const std::string& err) {
         LOG_INFO("Server stopping: {}", err);
+        // Proactively stop the server/transport so the demo process exits promptly
+        try { (void)server->Stop().get(); } catch (...) {}
         try { stopped.set_value(); } catch (...) {}
     });
 

include/mcp/SharedMemoryTransport.hpp

@@ -0,0 +1,97 @@
+//==========================================================================================================
+// SPDX-License-Identifier: MIT
+// Copyright (c) 2025 Vinny Parla
+// File: SharedMemoryTransport.hpp
+// Purpose: Cross-process shared-memory JSON-RPC transport built on Boost.Interprocess
+//==========================================================================================================
+
+#pragma once
+
+#include <future>
+#include <memory>
+#include <string>
+#include "mcp/Transport.h"
+
+namespace mcp {
+
+class SharedMemoryTransport : public ITransport {
+public:
+    //==========================================================================================================
+    // Options
+    // Purpose: Configuration for channel identity and queue sizing.
+    // Fields:
+    //   channelName: Base channel name. Two queues are derived: "<channelName>_c2s" and "<channelName>_s2c".
+    //   create: When true, create queues (server-side). When false, open existing queues (client-side).
+    //   maxMessageSize: Maximum size per message in bytes when creating queues (default: 1 MiB).
+    //   maxMessageCount: Maximum number of messages buffered per queue when creating (default: 256).
+    //==========================================================================================================
+    struct Options {
+        std::string channelName;
+        bool create{false};
+        std::size_t maxMessageSize{1024ull * 1024ull};
+        unsigned int maxMessageCount{256u};
+    };
+
+    explicit SharedMemoryTransport(const Options& opts);
+    ~SharedMemoryTransport() override;
+
+    ////////////////////////////////////////// ITransport //////////////////////////////////////////
+    //==========================================================================================================
+    // Starts the transport I/O loop. Returns when worker is ready to accept requests.
+    //==========================================================================================================
+    std::future<void> Start() override;
+
+    //==========================================================================================================
+    // Closes the transport and stops the I/O loop.
+    //==========================================================================================================
+    std::future<void> Close() override;
+
+    //==========================================================================================================
+    // Indicates whether the transport session is currently connected.
+    //==========================================================================================================
+    bool IsConnected() const override;
+
+    //==========================================================================================================
+    // Returns a transport session identifier for diagnostics.
+    //==========================================================================================================
+    std::string GetSessionId() const override;
+
+    //==========================================================================================================
+    // Sends a JSON-RPC request and returns a future for the response.
+    //==========================================================================================================
+    std::future<std::unique_ptr<JSONRPCResponse>> SendRequest(
+        std::unique_ptr<JSONRPCRequest> request) override;
+
+    //==========================================================================================================
+    // Sends a JSON-RPC notification (no response expected).
+    //==========================================================================================================
+    std::future<void> SendNotification(
+        std::unique_ptr<JSONRPCNotification> notification) override;
+
+    //==========================================================================================================
+    // Registers handlers for incoming notifications and errors. RequestHandler is used when this
+    // transport instance is wired on the server side.
+    //==========================================================================================================
+    void SetNotificationHandler(NotificationHandler handler) override;
+    void SetRequestHandler(RequestHandler handler) override;
+    void SetErrorHandler(ErrorHandler handler) override;
+
+private:
+    class Impl;
+    std::unique_ptr<Impl> pImpl;
+};
+
+//==========================================================================================================
+// SharedMemoryTransportFactory
+// Purpose: Factory for creating shared-memory transports from a configuration string.
+//          Supported formats:
+//            - "shm://<channelName>?create=true&maxSize=<bytes>&maxCount=<n>"
+//            - "<channelName>" (defaults to create=false)
+//          Unknown parameters are ignored.
+//==========================================================================================================
+class SharedMemoryTransportFactory : public ITransportFactory {
+public:
+    std::unique_ptr<ITransport> CreateTransport(const std::string& config) override;
+};
+
+} // namespace mcp

src/mcp/Server.cpp

@@ -14,6 +14,7 @@
 #include <limits>
 #include <stop_token>
 #include <thread>
+#include <condition_variable>
 #include <unordered_map>
 #include <unordered_set>
 
@@ -81,6 +82,8 @@ class Server::Impl {
     std::atomic<unsigned int> keepaliveFailureThreshold{3u};
     std::atomic<bool> keepaliveSending{false};
     std::atomic<bool> keepaliveSendFailed{false};
+    std::mutex keepaliveMutex;
+    std::condition_variable cvKeepalive;
 
     // Client logging preference (minimum severity)
     std::atomic<Logger::Level> clientLogMin{Logger::Level::DEBUG};
@@ -890,10 +893,8 @@ mcp::async::Task<void> Server::Impl::coStart(std::unique_ptr<ITransport> transpo
 
     this->transport->SetErrorHandler([this](const std::string& err){
         LOG_ERROR("Transport error: {}", err);
-        // If an error occurs during a keepalive send, mark last send as failed
-        if (this->keepaliveSending.load()) {
-            this->keepaliveSendFailed.store(true);
-        }
+        // Treat any transport error as a keepalive send failure to avoid races
+        this->keepaliveSendFailed.store(true);
         if (this->errorCallback) {
             try { 
                 this->errorCallback(err);
@@ -982,34 +983,33 @@ std::unique_ptr<JSONRPCResponse> Server::Impl::dispatchRequest(const JSONRPCRequ
 }
 
 void Server::Impl::sendInitializedAndListChangedAsync() {
-    std::thread([this]() {
-        try {
-            auto note = std::make_unique<JSONRPCNotification>();
-            note->method = Methods::Initialized;
-            note->params = JSONValue{JSONValue::Object{}};
-            (void)this->transport->SendNotification(std::move(note));
-            {
-                auto n = std::make_unique<JSONRPCNotification>();
-                n->method = Methods::ToolListChanged;
-                n->params = JSONValue{JSONValue::Object{}};
-                (void)this->transport->SendNotification(std::move(n));
-            }
-            {
-                auto n = std::make_unique<JSONRPCNotification>();
-                n->method = Methods::ResourceListChanged;
-                n->params = JSONValue{JSONValue::Object{}};
-                (void)this->transport->SendNotification(std::move(n));
-            }
-            {
-                auto n = std::make_unique<JSONRPCNotification>();
-                n->method = Methods::PromptListChanged;
-                n->params = JSONValue{JSONValue::Object{}};
-                (void)this->transport->SendNotification(std::move(n));
-            }
-        } catch (const std::exception& e) {
-            LOG_ERROR("Initialized/list_changed async send exception: {}", e.what());
+    // Send synchronously to avoid lifetime races in tests/demos
+    try {
+        auto note = std::make_unique<JSONRPCNotification>();
+        note->method = Methods::Initialized;
+        note->params = JSONValue{JSONValue::Object{}};
+        (void)this->transport->SendNotification(std::move(note));
+        {
+            auto n = std::make_unique<JSONRPCNotification>();
+            n->method = Methods::ToolListChanged;
+            n->params = JSONValue{JSONValue::Object{}};
+            (void)this->transport->SendNotification(std::move(n));
+        }
+        {
+            auto n = std::make_unique<JSONRPCNotification>();
+            n->method = Methods::ResourceListChanged;
+            n->params = JSONValue{JSONValue::Object{}};
+            (void)this->transport->SendNotification(std::move(n));
+        }
+        {
+            auto n = std::make_unique<JSONRPCNotification>();
+            n->method = Methods::PromptListChanged;
+            n->params = JSONValue{JSONValue::Object{}};
+            (void)this->transport->SendNotification(std::move(n));
         }
-    }).detach();
+    } catch (const std::exception& e) {
+        LOG_ERROR("Initialized/list_changed send exception: {}", e.what());
+    }
 }
 
 // Common helpers for paging and strict validation (used by list handlers)
@@ -1168,6 +1168,7 @@ mcp::async::Task<void> Server::Impl::coStop() {
     LOG_INFO("Stopping MCP server");
     if (this->keepaliveThread.joinable()) {
         this->keepaliveStop.store(true);
+        this->cvKeepalive.notify_all();
         try {
             this->keepaliveThread.join();
         } catch (...) {
@@ -1512,6 +1513,7 @@ Server::~Server() {
     // Ensure keepalive thread is stopped
     if (pImpl && pImpl->keepaliveThread.joinable()) {
         pImpl->keepaliveStop.store(true);
+        pImpl->cvKeepalive.notify_all();
         try {
             pImpl->keepaliveThread.join();
         } catch (...) {
@@ -1766,10 +1768,11 @@ void Server::SetKeepaliveIntervalMs(const std::optional<int>& intervalMs) {
         pImpl->capabilities.experimental.erase("keepalive");
         if (pImpl->keepaliveThread.joinable()) {
             pImpl->keepaliveStop.store(true);
+            // Wake the keepalive thread if it's waiting
+            pImpl->cvKeepalive.notify_all();
             try {
                 pImpl->keepaliveThread.join();
-            } catch (...) {
-            }
+            } catch (...) {}
             pImpl->keepaliveStop.store(false);
         }
         return;
@@ -1783,20 +1786,29 @@ void Server::SetKeepaliveIntervalMs(const std::optional<int>& intervalMs) {
     pImpl->capabilities.experimental["keepalive"] = JSONValue{kv};
 
     pImpl->keepaliveIntervalMs.store(ms);
+    // Wake the keepalive thread to re-evaluate the interval immediately
+    pImpl->cvKeepalive.notify_all();
 
     // Start background loop if not running
     if (!pImpl->keepaliveThread.joinable()) {
         pImpl->keepaliveStop.store(false);
         pImpl->keepaliveThread = std::thread([this]() {
             while (!pImpl->keepaliveStop.load()) {
                 int delay = pImpl->keepaliveIntervalMs.load();
-                if (delay <= 0) {
-                    break;
-                }
-                std::this_thread::sleep_for(std::chrono::milliseconds(delay));
-                if (pImpl->keepaliveStop.load()) {
-                    break;
+                if (delay <= 0) { break; }
+                {
+                    std::unique_lock<std::mutex> lk(pImpl->keepaliveMutex);
+                    // Wake early if stop requested or interval changes
+                    bool pred = pImpl->cvKeepalive.wait_for(
+                        lk,
+                        std::chrono::milliseconds(delay),
+                        [this, delay]() { return pImpl->keepaliveStop.load() || pImpl->keepaliveIntervalMs.load() != delay; }
+                    );
+                    (void)pred; // if pred true and interval changed, loop re-evaluates delay; if stop, loop breaks below
                 }
+                if (pImpl->keepaliveStop.load()) { break; }
+                // Re-check interval and connection after potential interval change
+                if (pImpl->keepaliveIntervalMs.load() <= 0) { break; }
                 if (!pImpl->transport || !pImpl->transport->IsConnected()) {
                     continue;
                 }
@@ -1807,8 +1819,13 @@ void Server::SetKeepaliveIntervalMs(const std::optional<int>& intervalMs) {
                     n->method = Methods::Keepalive;
                     n->params = JSONValue{JSONValue::Object{}};
                     (void)pImpl->transport->SendNotification(std::move(n));
-                } catch (...) {
+                } catch (const std::exception& e) {
                     pImpl->keepaliveSendFailed.store(true);
+                    if (pImpl->errorCallback) {
+                        try {
+                            pImpl->errorCallback("Keepalive failure: " + std::string(e.what()));
+                        } catch (...) {}
+                    }
                 }
                 pImpl->keepaliveSending.store(false);
                 if (pImpl->keepaliveSendFailed.load()) {
@@ -1819,21 +1836,15 @@ void Server::SetKeepaliveIntervalMs(const std::optional<int>& intervalMs) {
                     }
                     pImpl->keepaliveConsecutiveFailures.store(next);
                     if (next >= pImpl->keepaliveFailureThreshold.load()) {
-                        LOG_ERROR("Keepalive failure threshold reached ({}); closing transport", next);
-                        try {
-                            (void)pImpl->transport->Close();
-                        } catch (...) {
-                        }
+                        LOG_ERROR("Keepalive failure threshold reached ({})", next);
+                        try { (void)pImpl->transport->Close(); } catch (...) {}
                         if (pImpl->errorCallback) {
-                            try {
-                                pImpl->errorCallback("Keepalive failure threshold reached; closing transport");
-                            }
-                            catch (...) {
-                            }
+                            try { pImpl->errorCallback("Keepalive failure threshold reached; closing transport"); } catch (...) {}
                         }
                         break;
                     }
                 } else {
+                    // Success path: reset consecutive failures
                     pImpl->keepaliveConsecutiveFailures.store(0u);
                 }
             }