transport_adapter.hpp

Go to the documentation of this file.

// Copyright (c) 2025 [caomengxuan666]
 
#pragma once
 
 
#include <deque>
#include <exception>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <utility>
#include <variant>
 
#include "cxxmcp/client/client.hpp"
#include "cxxmcp/error.hpp"
#include "cxxmcp/protocol/serialization.hpp"
#include "cxxmcp/transport/transport.hpp"
 
namespace mcp::client {
 
namespace detail {
 
inline core::Error adapter_error(std::string_view message) {
  return errors::make(protocol::ErrorCode::InvalidRequest, std::string(message),
                      {}, "transport");
}
 
}  // namespace detail
 
class TransportContractAdapter final : public transport::ClientTransport {
 public:
  explicit TransportContractAdapter(mcp::client::Transport& transport)
      : transport_(&transport) {}
 
  explicit TransportContractAdapter(
      std::unique_ptr<mcp::client::Transport> transport)
      : owned_(std::move(transport)), transport_(owned_.get()) {}
 
  TransportContractAdapter(const TransportContractAdapter&) = delete;
  TransportContractAdapter& operator=(const TransportContractAdapter&) = delete;
 
  TransportContractAdapter(TransportContractAdapter&& other) noexcept
      : owned_(std::move(other.owned_)),
        transport_(owned_ ? owned_.get() : other.transport_),
        received_(std::move(other.received_)) {
    other.transport_ = nullptr;
  }
 
  TransportContractAdapter& operator=(
      TransportContractAdapter&& other) noexcept {
    if (this != &other) {
      owned_ = std::move(other.owned_);
      transport_ = owned_ ? owned_.get() : other.transport_;
      received_ = std::move(other.received_);
      other.transport_ = nullptr;
    }
    return *this;
  }
 
  std::string_view name() const noexcept override {
    return "client-transport-adapter";
  }
 
  core::Result<core::Unit> send(TxMessage message) override {
    if (transport_ == nullptr) {
      return mcp::core::unexpected(
          detail::adapter_error("client transport is null"));
    }
 
    if (const auto* request = std::get_if<protocol::JsonRpcRequest>(&message)) {
      auto response = transport_->send(*request);
      if (!response) {
        return mcp::core::unexpected(response.error());
      }
      received_.push_back(std::move(*response));
      return core::Unit{};
    }
 
    if (const auto* notification =
            std::get_if<protocol::JsonRpcNotification>(&message)) {
      return transport_->send_notification(*notification);
    }
 
    return mcp::core::unexpected(detail::adapter_error(
        "client transport adapter cannot send responses"));
  }
 
  core::Result<std::optional<RxMessage>> receive() override {
    if (received_.empty()) {
      return std::nullopt;
    }
    auto message = std::move(received_.front());
    received_.pop_front();
    return message;
  }
 
  core::Result<core::Unit> close() override {
    if (transport_ != nullptr) {
      transport_->stop();
    }
    return core::Unit{};
  }
 
 private:
  std::unique_ptr<mcp::client::Transport> owned_;
  mcp::client::Transport* transport_ = nullptr;
  std::deque<RxMessage> received_;
};
 
class ContractTransportAdapter final : public mcp::client::Transport {
 public:
  explicit ContractTransportAdapter(transport::ClientTransport& transport)
      : transport_(&transport) {}
 
  explicit ContractTransportAdapter(
      std::unique_ptr<transport::ClientTransport> transport)
      : owned_(std::move(transport)), transport_(owned_.get()) {}
 
  ContractTransportAdapter(const ContractTransportAdapter&) = delete;
  ContractTransportAdapter& operator=(const ContractTransportAdapter&) = delete;
 
  ContractTransportAdapter(ContractTransportAdapter&& other) noexcept
      : owned_(std::move(other.owned_)),
        transport_(owned_ ? owned_.get() : other.transport_),
        request_handler_(std::move(other.request_handler_)),
        notification_handler_(std::move(other.notification_handler_)) {
    other.transport_ = nullptr;
  }
 
  ContractTransportAdapter& operator=(
      ContractTransportAdapter&& other) noexcept {
    if (this != &other) {
      owned_ = std::move(other.owned_);
      transport_ = owned_ ? owned_.get() : other.transport_;
      request_handler_ = std::move(other.request_handler_);
      notification_handler_ = std::move(other.notification_handler_);
      other.transport_ = nullptr;
    }
    return *this;
  }
 
  core::Result<protocol::JsonRpcResponse> send(
      const protocol::JsonRpcRequest& request) override {
    if (transport_ == nullptr) {
      return mcp::core::unexpected(
          detail::adapter_error("client contract transport is null"));
    }
 
    const auto sent = transport_->send(protocol::JsonRpcMessage{request});
    if (!sent) {
      return mcp::core::unexpected(sent.error());
    }
 
    while (true) {
      auto received = transport_->receive();
      if (!received) {
        return mcp::core::unexpected(received.error());
      }
      if (!received->has_value()) {
        return mcp::core::unexpected(detail::adapter_error(
            "client contract transport closed before response"));
      }
 
      if (auto* response =
              std::get_if<protocol::JsonRpcResponse>(&received->value())) {
        if (response->id.has_value() && *response->id == request.id) {
          return *response;
        }
        return mcp::core::unexpected(detail::adapter_error(
            "client contract transport received unexpected response id"));
      }
 
      if (auto* notification =
              std::get_if<protocol::JsonRpcNotification>(&received->value())) {
        const auto handled = handle_notification(*notification);
        if (!handled) {
          return mcp::core::unexpected(handled.error());
        }
        continue;
      }
 
      if (auto* inbound_request =
              std::get_if<protocol::JsonRpcRequest>(&received->value())) {
        const auto handled = handle_request(*inbound_request);
        if (!handled) {
          return mcp::core::unexpected(handled.error());
        }
        continue;
      }
    }
  }
 
  core::Result<core::Unit> send_notification(
      const protocol::JsonRpcNotification& notification) override {
    if (transport_ == nullptr) {
      return mcp::core::unexpected(
          detail::adapter_error("client contract transport is null"));
    }
    return transport_->send(protocol::JsonRpcMessage{notification});
  }
 
  core::Result<core::Unit> start(
      TransportRequestHandler request_handler,
      TransportNotificationHandler notification_handler = {}) override {
    request_handler_ = std::move(request_handler);
    notification_handler_ = std::move(notification_handler);
    return core::Unit{};
  }
 
  void stop() noexcept override {
    if (transport_ != nullptr) {
      (void)transport_->close();
    }
  }
 
 private:
  core::Result<core::Unit> handle_notification(
      const protocol::JsonRpcNotification& notification) {
    if (!notification_handler_) {
      return core::Unit{};
    }
    try {
      return notification_handler_(notification);
    } catch (const std::exception& ex) {
      return mcp::core::unexpected(errors::handler_failed(ex.what()));
    } catch (...) {
      return mcp::core::unexpected(errors::handler_unknown_exception());
    }
  }
 
  core::Result<core::Unit> handle_request(
      const protocol::JsonRpcRequest& request) {
    protocol::JsonRpcResponse response;
    if (request_handler_) {
      try {
        auto handled = request_handler_(request);
        if (!handled) {
          response = protocol::make_error_response(
              request.id, error_object_from_core_error(handled.error()));
        } else {
          response = std::move(*handled);
        }
      } catch (const std::exception& ex) {
        response = protocol::make_error_response(
            request.id,
            error_object_from_core_error(errors::handler_failed(ex.what())));
      } catch (...) {
        response = protocol::make_error_response(
            request.id,
            error_object_from_core_error(errors::handler_unknown_exception()));
      }
    } else {
      response = protocol::make_error_response(
          request.id,
          protocol::make_error(protocol::ErrorCode::MethodNotFound,
                               "client request handler is not set"));
    }
 
    auto sent = transport_->send(protocol::JsonRpcMessage{std::move(response)});
    if (!sent) {
      return mcp::core::unexpected(sent.error());
    }
    return core::Unit{};
  }
 
  static protocol::ErrorObject error_object_from_core_error(
      const core::Error& error) {
    return errors::to_json_rpc_error(error);
  }
 
  std::unique_ptr<transport::ClientTransport> owned_;
  transport::ClientTransport* transport_ = nullptr;
  TransportRequestHandler request_handler_;
  TransportNotificationHandler notification_handler_;
};
 
}  // namespace mcp::client