From 7abf1184123675aa70f4beb033eafc50f43a5dc2 Mon Sep 17 00:00:00 2001 From: Christian Tzolov Date: Tue, 10 Dec 2024 13:25:16 +0100 Subject: [PATCH] refactor(transport): improve MCP transport layer design - Rename DefaultMcpTransport to AbstractMcpTransport and make it abstract - Add graceful shutdown support via closeGracefully() method - Enhance McpTransport interface with default close() implementation - Improve documentation across transport layer classes --- mcp/pom.xml | 1 - .../ai/experimental/mcp/client/McpClient.java | 124 ++++++---- .../client/stdio/StdioServerTransport.java | 80 +++---- ...ansport.java => AbstractMcpTransport.java} | 14 +- .../ai/experimental/mcp/spec/McpSession.java | 1 - .../experimental/mcp/spec/McpTransport.java | 92 +++++-- .../mcp/spec/DefaultMcpTransportTests.java | 226 ++++++++++-------- 7 files changed, 314 insertions(+), 224 deletions(-) rename mcp/src/main/java/spring/ai/experimental/mcp/spec/{DefaultMcpTransport.java => AbstractMcpTransport.java} (93%) diff --git a/mcp/pom.xml b/mcp/pom.xml index 9cb32e7..341090e 100644 --- a/mcp/pom.xml +++ b/mcp/pom.xml @@ -24,7 +24,6 @@ 21 21 - diff --git a/mcp/src/main/java/spring/ai/experimental/mcp/client/McpClient.java b/mcp/src/main/java/spring/ai/experimental/mcp/client/McpClient.java index 28ce6ce..3bf20a4 100644 --- a/mcp/src/main/java/spring/ai/experimental/mcp/client/McpClient.java +++ b/mcp/src/main/java/spring/ai/experimental/mcp/client/McpClient.java @@ -1,18 +1,18 @@ /* -* Copyright 2024 - 2024 the original author or authors. -* -* Licensed under the Apache License, Version 2.0 (the "License"); -* you may not use this file except in compliance with the License. -* You may obtain a copy of the License at -* -* https://www.apache.org/licenses/LICENSE-2.0 -* -* Unless required by applicable law or agreed to in writing, software -* distributed under the License is distributed on an "AS IS" BASIS, -* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -* See the License for the specific language governing permissions and -* limitations under the License. -*/ + * Copyright 2024 - 2024 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package spring.ai.experimental.mcp.client; import java.time.Duration; @@ -21,41 +21,85 @@ import com.fasterxml.jackson.databind.ObjectMapper; import spring.ai.experimental.mcp.spec.McpTransport; /** - * The MCP client is the main entry point for interacting with the Model Context Protocol - * (MCP) server. + * Factory class providing static methods for creating Model Context Protocol (MCP) clients. + * This class serves as the main entry point for establishing connections with MCP servers, + * offering both synchronous and asynchronous client implementations. + * + *

The class provides factory methods to create either: + *

    + *
  • {@link McpAsyncClient} for non-blocking operations + *
  • {@link McpSyncClient} for blocking operations + *
+ * + *

Each client type can be instantiated with default settings or with custom configuration + * including request timeout and JSON object mapping. + * + *

Future implementations will introduce a builder pattern for more flexible client configuration. * * @author Christian Tzolov * @author Dariusz Jędrzejczyk */ public class McpClient { - private McpClient() { - } + /** + * Private constructor to prevent instantiation as this is a utility class + * containing only static factory methods. + */ + private McpClient() { + } - // TODO: introduce a builder like: - // McpClient.using(transport) - // .withRequestTimeout(Duration.ofSeconds(5)) - // .withObjectMapper(objectMapper); <- or even a more sophisticated - // JSONtoPOJOCodec type - // .sync(); + // TODO: introduce a builder like: + // McpClient.using(transport) + // .withRequestTimeout(Duration.ofSeconds(5)) + // .withObjectMapper(objectMapper); <- or even a more sophisticated + // JSONtoPOJOCodec type + // .sync(); - public static McpAsyncClient async(McpTransport transport) { - return new McpAsyncClient(transport); - } + /** + * Creates an asynchronous MCP client with default configuration. + * + * @param transport The transport layer implementation for MCP communication + * @return A new instance of {@link McpAsyncClient} + */ + public static McpAsyncClient async(McpTransport transport) { + return new McpAsyncClient(transport); + } - public static McpAsyncClient async(McpTransport transport, Duration requestTimeout, - ObjectMapper objectMapper) { - return new McpAsyncClient(transport, requestTimeout, objectMapper); - } - - public static McpSyncClient sync(McpTransport transport) { - return new McpSyncClient(async(transport)); - } - - public static McpSyncClient sync(McpTransport transport, Duration requestTimeout, - ObjectMapper objectMapper) { - return new McpSyncClient(async(transport, requestTimeout, objectMapper)); - } + /** + * Creates an asynchronous MCP client with custom configuration. + * + * @param transport The transport layer implementation for MCP communication + * @param requestTimeout The duration to wait before timing out requests + * @param objectMapper The Jackson object mapper for JSON serialization/deserialization + * @return A new instance of {@link McpAsyncClient} + */ + public static McpAsyncClient async(McpTransport transport, Duration requestTimeout, + ObjectMapper objectMapper) { + return new McpAsyncClient(transport, requestTimeout, objectMapper); + } + /** + * Creates a synchronous MCP client with default configuration. + * This method wraps an asynchronous client to provide synchronous operations. + * + * @param transport The transport layer implementation for MCP communication + * @return A new instance of {@link McpSyncClient} + */ + public static McpSyncClient sync(McpTransport transport) { + return new McpSyncClient(async(transport)); + } + /** + * Creates a synchronous MCP client with custom configuration. + * This method wraps an asynchronous client to provide synchronous operations. + * + * @param transport The transport layer implementation for MCP communication + * @param requestTimeout The duration to wait before timing out requests + * @param objectMapper The Jackson object mapper for JSON serialization/deserialization + * @return A new instance of {@link McpSyncClient} + */ + public static McpSyncClient sync(McpTransport transport, Duration requestTimeout, + ObjectMapper objectMapper) { + return new McpSyncClient(async(transport, requestTimeout, objectMapper)); + } } diff --git a/mcp/src/main/java/spring/ai/experimental/mcp/client/stdio/StdioServerTransport.java b/mcp/src/main/java/spring/ai/experimental/mcp/client/stdio/StdioServerTransport.java index ee05901..0ed8607 100644 --- a/mcp/src/main/java/spring/ai/experimental/mcp/client/stdio/StdioServerTransport.java +++ b/mcp/src/main/java/spring/ai/experimental/mcp/client/stdio/StdioServerTransport.java @@ -10,10 +10,11 @@ import java.util.concurrent.Executors; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; +import reactor.core.publisher.Mono; import reactor.core.scheduler.Scheduler; import reactor.core.scheduler.Schedulers; import spring.ai.experimental.mcp.client.util.Assert; -import spring.ai.experimental.mcp.spec.DefaultMcpTransport; +import spring.ai.experimental.mcp.spec.AbstractMcpTransport; import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCMessage; import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCNotification; import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCRequest; @@ -25,7 +26,7 @@ import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCResponse; * @author Christian Tzolov * @author Dariusz Jędrzejczyk */ -public class StdioServerTransport extends DefaultMcpTransport { +public class StdioServerTransport extends AbstractMcpTransport { private final static TypeReference> MAP_TYPE_REF = new TypeReference>() { }; @@ -194,47 +195,46 @@ public class StdioServerTransport extends DefaultMcpTransport { .subscribe(); } - // TODO: provide a non-blocking variant with graceful option - public void stop() { - this.outboundScheduler.dispose(); - this.inboundScheduler.dispose(); - this.errorScheduler.dispose(); - - // Destroy process - if (this.process != null) { - this.process.destroyForcibly(); - } - } - @Override - public void close() { + public Mono closeGracefully() { - this.stop(); + this.isRunning = false; - super.close(); // Do we need this? - - // Close resources - if (this.processReader != null) { - try { - this.processReader.close(); - } catch (IOException e) { - e.printStackTrace(); - } - } - if (this.processWriter != null) { - try { - this.processWriter.close(); - } catch (IOException e) { - e.printStackTrace(); - } - } - if (this.processErrorReader != null) { - try { - this.processErrorReader.close(); - } catch (IOException e) { - e.printStackTrace(); - } - } + return Mono.whenDelayError( + Mono.fromRunnable(() -> { + try { + this.processErrorReader.close(); + } catch (IOException e) { + throw new RuntimeException(e); + } + }).subscribeOn(errorScheduler), + Mono.fromRunnable(() -> { + try { + this.processReader.close(); + } catch (IOException e) { + throw new RuntimeException(e); + } + }).subscribeOn(inboundScheduler), + Mono.fromRunnable(() -> { + try { + this.processWriter.close(); + } catch (IOException e) { + throw new RuntimeException(e); + } + }).subscribeOn(outboundScheduler)) + .then(Mono.whenDelayError(inboundScheduler.disposeGracefully(), + outboundScheduler.disposeGracefully(), + errorScheduler.disposeGracefully())) + .then(Mono.fromRunnable(() -> { + if (this.process != null) { + var process = this.process.destroyForcibly(); + if (process.exitValue() != 0) { + throw new RuntimeException("Failed to destroy process"); + } + } + })) + .then() + .subscribeOn(Schedulers.boundedElastic()); } } \ No newline at end of file diff --git a/mcp/src/main/java/spring/ai/experimental/mcp/spec/DefaultMcpTransport.java b/mcp/src/main/java/spring/ai/experimental/mcp/spec/AbstractMcpTransport.java similarity index 93% rename from mcp/src/main/java/spring/ai/experimental/mcp/spec/DefaultMcpTransport.java rename to mcp/src/main/java/spring/ai/experimental/mcp/spec/AbstractMcpTransport.java index ac93903..db99e26 100644 --- a/mcp/src/main/java/spring/ai/experimental/mcp/spec/DefaultMcpTransport.java +++ b/mcp/src/main/java/spring/ai/experimental/mcp/spec/AbstractMcpTransport.java @@ -27,7 +27,7 @@ import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCMessage; * @author Christian Tzolov * @author Dariusz Jędrzejczyk */ -public class DefaultMcpTransport implements McpTransport { +public abstract class AbstractMcpTransport implements McpTransport { protected final ObjectMapper objectMapper; @@ -39,11 +39,11 @@ public class DefaultMcpTransport implements McpTransport { .println("Received message: " + message); private Consumer errorHandler = error -> System.err.println("Received error: " + error); - public DefaultMcpTransport() { + public AbstractMcpTransport() { this(new ObjectMapper()); } - public DefaultMcpTransport(ObjectMapper objectMapper) { + public AbstractMcpTransport(ObjectMapper objectMapper) { Assert.notNull(objectMapper, "ObjectMapper must not be null"); this.objectMapper = objectMapper; @@ -73,7 +73,7 @@ public class DefaultMcpTransport implements McpTransport { private void handleIncomingErrors() { this.errorSink.asFlux().subscribe(e -> { this.errorHandler.accept(e); - System.err.println(e); + // System.err.println(e); TODO: log the error }); } @@ -103,12 +103,6 @@ public class DefaultMcpTransport implements McpTransport { this.handleIncomingMessages(); } - // Close the transport - @Override - public void close() { - // this.onClose(); - } - @Override public Mono sendMessage(JSONRPCMessage message) { if (this.outboundSink.tryEmitNext(message).isSuccess()) { diff --git a/mcp/src/main/java/spring/ai/experimental/mcp/spec/McpSession.java b/mcp/src/main/java/spring/ai/experimental/mcp/spec/McpSession.java index b0faeab..894a0ee 100644 --- a/mcp/src/main/java/spring/ai/experimental/mcp/spec/McpSession.java +++ b/mcp/src/main/java/spring/ai/experimental/mcp/spec/McpSession.java @@ -111,7 +111,6 @@ public class McpSession { McpSchema.JSONRPCNotification jsonrpcNotification = new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION, method, params); try { - // TODO: make it non-blocking this.transport.sendMessage(jsonrpcNotification); } catch (Exception e) { return Mono.error(new McpError(e)); diff --git a/mcp/src/main/java/spring/ai/experimental/mcp/spec/McpTransport.java b/mcp/src/main/java/spring/ai/experimental/mcp/spec/McpTransport.java index ea06c3f..8751fff 100644 --- a/mcp/src/main/java/spring/ai/experimental/mcp/spec/McpTransport.java +++ b/mcp/src/main/java/spring/ai/experimental/mcp/spec/McpTransport.java @@ -12,67 +12,104 @@ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. - */ + */ package spring.ai.experimental.mcp.spec; +import java.time.Duration; import java.util.function.Consumer; import reactor.core.publisher.Mono; import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCMessage; /** - * Defines the asynchronous transport layer for the Model Context Protocol (MCP). + * Defines the asynchronous transport layer for the Model Context Protocol + * (MCP). * - *

The McpTransport interface provides the foundation for implementing custom transport - * mechanisms in the Model Context Protocol. It handles the bidirectional communication between - * the client and server components, supporting asynchronous message exchange using JSON-RPC format.

+ *

+ * The McpTransport interface provides the foundation for implementing custom + * transport + * mechanisms in the Model Context Protocol. It handles the bidirectional + * communication between + * the client and server components, supporting asynchronous message exchange + * using JSON-RPC format. + *

* - *

Implementations of this interface are responsible for:

+ *

+ * Implementations of this interface are responsible for: + *

*
    - *
  • Managing the lifecycle of the transport connection
    • - *
  • Handling incoming messages and errors from the server
    • - *
  • Sending outbound messages to the server
    • + *
  • Managing the lifecycle of the transport connection
    • + *
  • Handling incoming messages and errors from the server
    • + *
  • Sending outbound messages to the server
    • *
* - *

The transport layer is designed to be protocol-agnostic, allowing for various implementations - * such as WebSocket, HTTP, or custom protocols.

+ *

+ * The transport layer is designed to be protocol-agnostic, allowing for various + * implementations + * such as WebSocket, HTTP, or custom protocols. + *

* * @author Christian Tzolov * @author Dariusz Jędrzejczyk */ -public interface McpTransport extends AutoCloseable { - +public interface McpTransport { + /** * Initializes and starts the transport connection. * - *

This method should be called before any message exchange can occur. It sets up - * the necessary resources and establishes the connection to the server.

+ *

+ * This method should be called before any message exchange can occur. It sets + * up + * the necessary resources and establishes the connection to the server. + *

*/ void start(); /** * Closes the transport connection and releases any associated resources. * - *

This method ensures proper cleanup of resources when the transport is no longer needed. - * It should handle the graceful shutdown of any active connections.

+ *

+ * This method ensures proper cleanup of resources when the transport is no + * longer needed. + * It should handle the graceful shutdown of any active connections. + *

*/ - void close(); + default void close() { + this.closeGracefully().subscribe(); + } + + /** + * Closes the transport connection and releases any associated resources + * asynchronously. + * + * @return a {@link Mono} that completes when the connection has been + * closed. + */ + Mono closeGracefully(); /** * Sets the handler for processing incoming messages from the server. * - *

The provided handler will be called whenever a new message is received from the server. - * Messages are expected to follow the JSON-RPC format as defined in the MCP specification.

+ *

+ * The provided handler will be called whenever a new message is received from + * the server. + * Messages are expected to follow the JSON-RPC format as defined in the MCP + * specification. + *

* - * @param inboundMessageHandler a consumer that processes incoming {@link JSONRPCMessage}s + * @param inboundMessageHandler a consumer that processes incoming + * {@link JSONRPCMessage}s */ void setInboudMessageHandler(Consumer inboundMessageHandler); /** * Sets the handler for processing transport-level errors. * - *

The provided handler will be called when errors occur during transport operations, - * such as connection failures or protocol violations.

+ *

+ * The provided handler will be called when errors occur during transport + * operations, + * such as connection failures or protocol violations. + *

* * @param inboundErrorHandler a consumer that processes error messages */ @@ -81,11 +118,14 @@ public interface McpTransport extends AutoCloseable { /** * Sends a message to the server asynchronously. * - *

This method handles the transmission of messages to the server in an asynchronous manner. - * Messages are sent in JSON-RPC format as specified by the MCP protocol.

+ *

+ * This method handles the transmission of messages to the server in an + * asynchronous manner. + * Messages are sent in JSON-RPC format as specified by the MCP protocol. + *

* * @param message the {@link JSONRPCMessage} to be sent to the server * @return a {@link Mono} that completes when the message has been sent */ - Mono sendMessage(JSONRPCMessage message); + Mono sendMessage(JSONRPCMessage message); } diff --git a/mcp/src/test/java/spring/ai/experimental/mcp/spec/DefaultMcpTransportTests.java b/mcp/src/test/java/spring/ai/experimental/mcp/spec/DefaultMcpTransportTests.java index 182edd7..0a2abac 100644 --- a/mcp/src/test/java/spring/ai/experimental/mcp/spec/DefaultMcpTransportTests.java +++ b/mcp/src/test/java/spring/ai/experimental/mcp/spec/DefaultMcpTransportTests.java @@ -15,148 +15,162 @@ */ package spring.ai.experimental.mcp.spec; +import java.util.concurrent.atomic.AtomicReference; + import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; -import spring.ai.experimental.mcp.spec.DefaultMcpTransport; import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCMessage; import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCRequest; -import java.util.concurrent.atomic.AtomicReference; - import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; class DefaultMcpTransportTests { - private DefaultMcpTransport transport; - private ObjectMapper objectMapper; + private AbstractMcpTransport transport; + private ObjectMapper objectMapper; - @BeforeEach - void setUp() { - objectMapper = new ObjectMapper(); - transport = new DefaultMcpTransport(objectMapper); - } + @BeforeEach + void setUp() { + objectMapper = new ObjectMapper(); + transport = new AbstractMcpTransport(objectMapper) { + @Override + public Mono closeGracefully() { + return null; + } + }; + } - @Test - void constructorShouldValidateObjectMapper() { - assertThatThrownBy(() -> new DefaultMcpTransport(null)) - .isInstanceOf(IllegalArgumentException.class) - .hasMessage("ObjectMapper must not be null"); - } + @Test + void constructorShouldValidateObjectMapper() { + assertThatThrownBy(() -> new AbstractMcpTransport(null) { + @Override + public Mono closeGracefully() { + return null; + } + }) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("ObjectMapper must not be null"); + } - @Test - void defaultConstructorShouldCreateValidInstance() { - DefaultMcpTransport defaultTransport = new DefaultMcpTransport(); - assertThat(defaultTransport.getObjectMapper()).isNotNull(); - } + @Test + void defaultConstructorShouldCreateValidInstance() { + AbstractMcpTransport defaultTransport = new AbstractMcpTransport() { + @Override + public Mono closeGracefully() { + return null; + } + }; + assertThat(defaultTransport.getObjectMapper()).isNotNull(); + } - @Test - void customObjectMapperConstructorShouldCreateValidInstance() { - assertThat(transport.getObjectMapper()).isSameAs(objectMapper); - } + @Test + void customObjectMapperConstructorShouldCreateValidInstance() { + assertThat(transport.getObjectMapper()).isSameAs(objectMapper); + } - @Test - void sendMessageShouldEmitToOutboundSink() { - JSONRPCRequest message = new JSONRPCRequest( - "2.0", - "test", - 1, - null); + @Test + void sendMessageShouldEmitToOutboundSink() { + JSONRPCRequest message = new JSONRPCRequest( + "2.0", + "test", + 1, + null); - AtomicReference receivedMessage = new AtomicReference<>(); - transport.getOutboundSink().asFlux().subscribe(receivedMessage::set); + AtomicReference receivedMessage = new AtomicReference<>(); + transport.getOutboundSink().asFlux().subscribe(receivedMessage::set); - Mono result = transport.sendMessage(message); + Mono result = transport.sendMessage(message); - StepVerifier.create(result) - .verifyComplete(); + StepVerifier.create(result) + .verifyComplete(); - assertThat(receivedMessage.get()) - .isNotNull() - .satisfies(msg -> { - assertThat(((JSONRPCRequest) msg).jsonrpc()).isEqualTo("2.0"); - assertThat(((JSONRPCRequest) msg).method()).isEqualTo("test"); - }); - } + assertThat(receivedMessage.get()) + .isNotNull() + .satisfies(msg -> { + assertThat(((JSONRPCRequest) msg).jsonrpc()).isEqualTo("2.0"); + assertThat(((JSONRPCRequest) msg).method()).isEqualTo("test"); + }); + } - @Test - void customInboundMessageHandlerShouldReceiveMessages() { - AtomicReference receivedMessage = new AtomicReference<>(); - transport.setInboudMessageHandler(receivedMessage::set); - transport.start(); + @Test + void customInboundMessageHandlerShouldReceiveMessages() { + AtomicReference receivedMessage = new AtomicReference<>(); + transport.setInboudMessageHandler(receivedMessage::set); + transport.start(); - JSONRPCRequest message = new JSONRPCRequest( - "2.0", - "test", - 1, - null); + JSONRPCRequest message = new JSONRPCRequest( + "2.0", + "test", + 1, + null); - transport.getInboundSink().tryEmitNext(message); + transport.getInboundSink().tryEmitNext(message); - assertThat(receivedMessage.get()) - .isNotNull() - .satisfies(msg -> { - assertThat(((JSONRPCRequest) msg).jsonrpc()).isEqualTo("2.0"); - assertThat(((JSONRPCRequest) msg).method()).isEqualTo("test"); - }); - } + assertThat(receivedMessage.get()) + .isNotNull() + .satisfies(msg -> { + assertThat(((JSONRPCRequest) msg).jsonrpc()).isEqualTo("2.0"); + assertThat(((JSONRPCRequest) msg).method()).isEqualTo("test"); + }); + } - @Test - void customErrorHandlerShouldReceiveErrors() { - AtomicReference receivedError = new AtomicReference<>(); - transport.setInboundErrorHandler(receivedError::set); - transport.start(); + @Test + void customErrorHandlerShouldReceiveErrors() { + AtomicReference receivedError = new AtomicReference<>(); + transport.setInboundErrorHandler(receivedError::set); + transport.start(); - String errorMessage = "Test error"; - transport.getErrorSink().tryEmitNext(errorMessage); + String errorMessage = "Test error"; + transport.getErrorSink().tryEmitNext(errorMessage); - assertThat(receivedError.get()) - .isNotNull() - .isEqualTo(errorMessage); - } + assertThat(receivedError.get()) + .isNotNull() + .isEqualTo(errorMessage); + } - @Test - void startShouldInitializeMessageAndErrorHandling() { - AtomicReference receivedMessage = new AtomicReference<>(); - AtomicReference receivedError = new AtomicReference<>(); + @Test + void startShouldInitializeMessageAndErrorHandling() { + AtomicReference receivedMessage = new AtomicReference<>(); + AtomicReference receivedError = new AtomicReference<>(); - transport.setInboudMessageHandler(receivedMessage::set); - transport.setInboundErrorHandler(receivedError::set); - transport.start(); + transport.setInboudMessageHandler(receivedMessage::set); + transport.setInboundErrorHandler(receivedError::set); + transport.start(); - JSONRPCRequest message = new JSONRPCRequest( - "2.0", - "test", - 1, - null); + JSONRPCRequest message = new JSONRPCRequest( + "2.0", + "test", + 1, + null); - String errorMessage = "Test error"; + String errorMessage = "Test error"; - transport.getInboundSink().tryEmitNext(message); - transport.getErrorSink().tryEmitNext(errorMessage); + transport.getInboundSink().tryEmitNext(message); + transport.getErrorSink().tryEmitNext(errorMessage); - assertThat(receivedMessage.get()).isNotNull(); - assertThat(receivedError.get()).isEqualTo(errorMessage); - } + assertThat(receivedMessage.get()).isNotNull(); + assertThat(receivedError.get()).isEqualTo(errorMessage); + } - @Test - void sendMessageShouldReturnErrorWhenSinkFails() { - JSONRPCRequest message = new JSONRPCRequest( - "2.0", - "test", - 1, - null); - // Close the sink to simulate failure - transport.getOutboundSink().tryEmitComplete(); + @Test + void sendMessageShouldReturnErrorWhenSinkFails() { + JSONRPCRequest message = new JSONRPCRequest( + "2.0", + "test", + 1, + null); + // Close the sink to simulate failure + transport.getOutboundSink().tryEmitComplete(); - Mono result = transport.sendMessage(message); + Mono result = transport.sendMessage(message); - StepVerifier.create(result) - .verifyErrorSatisfies(error -> assertThat(error) - .isInstanceOf(RuntimeException.class) - .hasMessage("Failed to enqueue message")); - } + StepVerifier.create(result) + .verifyErrorSatisfies(error -> assertThat(error) + .isInstanceOf(RuntimeException.class) + .hasMessage("Failed to enqueue message")); + } }