Rename modules, update .gitignore

This commit is contained in:
Mark Pollack 2024-12-10 18:13:06 -05:00
commit 910be3cee7
6 changed files with 71 additions and 156 deletions

View file

@ -1,80 +0,0 @@
/*
* 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.attic;
import java.time.Duration;
import com.fasterxml.jackson.databind.ObjectMapper;
import spring.ai.experimental.mcp.client.McpClient;
import spring.ai.experimental.mcp.client.McpSyncClient;
import spring.ai.experimental.mcp.client.stdio.ServerParameters;
import spring.ai.experimental.mcp.client.stdio.StdioServerTransport;
import spring.ai.experimental.mcp.spec.McpSchema.ListResourcesResult;
import spring.ai.experimental.mcp.spec.McpSchema.ListToolsResult;
import spring.ai.experimental.mcp.spec.McpSchema.Resource;
/**
* @author Christian Tzolov
* @since 1.0.0
*/
public class ClientSessionTests2 {
public static void main(String[] args) {
var stdioParams = ServerParameters.builder("uv")
.args("--directory", "dir", "run", "mcp-server-sqlite",
"--db-path", "~/test.db")
.build();
try (McpSyncClient clientSession =
McpClient.sync(new StdioServerTransport(stdioParams),
Duration.ofSeconds(10), new ObjectMapper())) {
clientSession.initialize();
ListToolsResult tools = clientSession.listTools(null);
System.out.println("Tools: " + tools);
clientSession.ping();
// CallToolRequest callToolRequest = new CallToolRequest("echo",
// Map.of("message", "Hello MCP Spring AI!"));
// CallToolResult callToolResult =
// clientSession.callTool(callToolRequest);
// System.out.println("Call Tool Result: " + callToolResult);
// //
// clientSession.sendRootsListChanged();
// Resources
ListResourcesResult resources = clientSession.listResources(null);
System.out.println("Resources Size: " + resources.resources().size());
System.out.println("Resources: " + resources);
for (Resource resource : resources.resources()) {
System.out.println(clientSession.readResource(resource));
}
// var resourceTemplate = clientSession.listResourceTemplates(null);
// System.out.println("Resource Templates: " + resourceTemplate);
// stdioClient.awaitForExit();
}
catch (Exception e) {
e.printStackTrace();
}
}
}

View file

@ -1,75 +0,0 @@
/*
* 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.attic;
import java.time.Duration;
import com.fasterxml.jackson.databind.ObjectMapper;
import spring.ai.experimental.mcp.client.McpClient;
import spring.ai.experimental.mcp.client.McpSyncClient;
import spring.ai.experimental.mcp.client.stdio.ServerParameters;
import spring.ai.experimental.mcp.client.stdio.StdioServerTransport;
import spring.ai.experimental.mcp.spec.McpSchema.ListResourcesResult;
import spring.ai.experimental.mcp.spec.McpSchema.ListToolsResult;
import spring.ai.experimental.mcp.spec.McpSchema.Resource;
/**
* @author Christian Tzolov
* @since 1.0.0
*/
public class ClientSessionTests3 {
public static void main(String[] args) {
var stdioParams = ServerParameters.builder("uv")
.args("--directory", "dir", "run", "mcp-server-sqlite",
"--db-path", "~/test.db")
.build();
McpSyncClient clientSession = null;
try {
clientSession = McpClient.sync(new StdioServerTransport(stdioParams),
Duration.ofSeconds(10), new ObjectMapper());
clientSession.initialize();
ListToolsResult tools = clientSession.listTools(null);
System.out.println("Tools: " + tools);
clientSession.ping();
// Resources
ListResourcesResult resources = clientSession.listResources(null);
System.out.println("Resources Size: " + resources.resources().size());
System.out.println("Resources: " + resources);
for (Resource resource : resources.resources()) {
System.out.println(clientSession.readResource(resource));
}
}
catch (Exception e) {
e.printStackTrace();
}
finally {
if (clientSession != null) {
clientSession.close();
}
}
}
}

View file

@ -1,180 +0,0 @@
/*
* 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;
import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import spring.ai.experimental.mcp.client.stdio.ServerParameters;
import spring.ai.experimental.mcp.client.stdio.StdioServerTransport;
import spring.ai.experimental.mcp.spec.McpSchema.CallToolRequest;
import spring.ai.experimental.mcp.spec.McpSchema.GetPromptRequest;
import spring.ai.experimental.mcp.spec.McpSchema.Prompt;
import spring.ai.experimental.mcp.spec.McpSchema.Resource;
import spring.ai.experimental.mcp.spec.McpSchema.Tool;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Unit tests for MCP Client Session functionality.
*
* @author Christian Tzolov
* @since 1.0.0
*/
class McpAsyncClientTests {
private McpAsyncClient mcpAsyncClient;
private ServerParameters stdioParams;
private static final Duration TIMEOUT = Duration.ofSeconds(10);
private static final String TEST_MESSAGE = "Hello MCP Spring AI!";
@BeforeEach
void setUp() {
stdioParams = ServerParameters.builder("npx")
.args("-y", "@modelcontextprotocol/server-everything", "dir")
.build();
assertThatCode(() -> {
mcpAsyncClient = McpClient.async(new StdioServerTransport(stdioParams), TIMEOUT, new ObjectMapper());
mcpAsyncClient.initialize();
}).doesNotThrowAnyException();
}
@AfterEach
void tearDown() {
// if (mcpSyncClient != null) {
// assertThatCode(() -> mcpSyncClient.close()).doesNotThrowAnyException();
// }
}
@Test
void testConstructorWithInvalidArguments() {
assertThatThrownBy(() -> McpClient.sync(null, TIMEOUT, new ObjectMapper()))
.isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() -> McpClient.sync(new StdioServerTransport(stdioParams), null, new ObjectMapper()))
.isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() -> McpClient.sync(new StdioServerTransport(stdioParams), TIMEOUT, null))
.isInstanceOf(IllegalArgumentException.class);
}
@Test
@Timeout(15) // Giving extra time beyond the client timeout
void testListTools() {
mcpAsyncClient.listTools(null).subscribe(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull();
});
}
@Test
@Timeout(15)
void testPing() {
assertThatCode(() -> mcpAsyncClient.ping().block()).doesNotThrowAnyException();
}
@Test
@Timeout(15)
void testCallTool() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE));
mcpAsyncClient.callTool(callToolRequest).subscribe(callToolResult -> {
assertThat(callToolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).isNotNull();
assertThat(result.isError()).isNull();
});
});
}
@Test
@Timeout(15)
void testCallToolWithInvalidTool() {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", TEST_MESSAGE));
assertThatThrownBy(() -> mcpAsyncClient.callTool(invalidRequest).block()).isInstanceOf(Exception.class);
}
@Test
@Timeout(15)
void testRootsListChanged() {
assertThatCode(() -> mcpAsyncClient.sendRootsListChanged().block()).doesNotThrowAnyException();
}
@Test
@Timeout(15)
void testListResources() {
mcpAsyncClient.listResources(null).subscribe(resources -> {
assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull();
}
});
});
}
@Test
void testMcpAsyncClientState() {
assertThat(mcpAsyncClient).isNotNull();
}
@Test
@Timeout(15)
void testListPrompts() {
mcpAsyncClient.listPrompts(null).subscribe(prompts -> {
assertThat(prompts).isNotNull()
.satisfies(result -> {
assertThat(result.prompts()).isNotNull();
if (!result.prompts().isEmpty()) {
Prompt firstPrompt = result.prompts().get(0);
assertThat(firstPrompt.name()).isNotNull();
assertThat(firstPrompt.description()).isNotNull();
}
});
});
}
@Test
@Timeout(15)
void testGetPrompt() {
mcpAsyncClient.getPrompt(new GetPromptRequest("simple_prompt", Map.of())).subscribe(prompt -> {
assertThat(prompt).isNotNull()
.satisfies(result -> {
assertThat(result.description()).isNotNull();
assertThat(result.messages()).isNotEmpty();
});
});
}
}

View file

@ -1,174 +0,0 @@
/*
* 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;
import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import spring.ai.experimental.mcp.client.stdio.ServerParameters;
import spring.ai.experimental.mcp.client.stdio.StdioServerTransport;
import spring.ai.experimental.mcp.spec.McpSchema.CallToolRequest;
import spring.ai.experimental.mcp.spec.McpSchema.CallToolResult;
import spring.ai.experimental.mcp.spec.McpSchema.ListResourcesResult;
import spring.ai.experimental.mcp.spec.McpSchema.ListToolsResult;
import spring.ai.experimental.mcp.spec.McpSchema.Resource;
import spring.ai.experimental.mcp.spec.McpSchema.TextContent;
import spring.ai.experimental.mcp.spec.McpSchema.Tool;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatCode;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Unit tests for MCP Client Session functionality.
*
* @author Christian Tzolov
* @since 1.0.0
*/
class McpSyncClientTests {
private McpSyncClient mcpSyncClient;
private ServerParameters stdioParams;
private static final Duration TIMEOUT = Duration.ofSeconds(10);
private static final String TEST_MESSAGE = "Hello MCP Spring AI!";
@BeforeEach
void setUp() {
stdioParams = ServerParameters.builder("npx")
.args("-y", "@modelcontextprotocol/server-everything", "dir")
.build();
assertThatCode(() -> {
mcpSyncClient = McpClient.sync(new StdioServerTransport(stdioParams), TIMEOUT, new ObjectMapper());
mcpSyncClient.initialize();
}).doesNotThrowAnyException();
}
@AfterEach
void tearDown() {
if (mcpSyncClient != null) {
assertThatCode(() -> mcpSyncClient.close()).doesNotThrowAnyException();
}
}
@Test
void testConstructorWithInvalidArguments() {
assertThatThrownBy(() -> McpClient.sync(null, TIMEOUT, new ObjectMapper()))
.isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() -> McpClient.sync(new StdioServerTransport(stdioParams), null, new ObjectMapper()))
.isInstanceOf(IllegalArgumentException.class);
assertThatThrownBy(() -> McpClient.sync(new StdioServerTransport(stdioParams), TIMEOUT, null))
.isInstanceOf(IllegalArgumentException.class);
}
@Test
@Timeout(15) // Giving extra time beyond the client timeout
void testListTools() {
ListToolsResult tools = mcpSyncClient.listTools(null);
assertThat(tools).isNotNull().satisfies(result -> {
assertThat(result.tools()).isNotNull().isNotEmpty();
Tool firstTool = result.tools().get(0);
assertThat(firstTool.name()).isNotNull();
assertThat(firstTool.description()).isNotNull();
});
}
@Test
@Timeout(15) // Giving extra time beyond the client timeout
void testCallTools() {
CallToolResult toolResult = mcpSyncClient.callTool(new CallToolRequest("add", Map.of("a", 3, "b", 4)));
assertThat(toolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).hasSize(1);
TextContent content = (TextContent) result.content().get(0);
assertThat(content).isNotNull();
assertThat(content.text()).isNotNull();
assertThat(content.text()).contains("7");
System.out.println(content);
});
}
@Test
@Timeout(15)
void testPing() {
assertThatCode(() -> mcpSyncClient.ping()).doesNotThrowAnyException();
}
@Test
@Timeout(15)
void testCallTool() {
CallToolRequest callToolRequest = new CallToolRequest("echo", Map.of("message", TEST_MESSAGE));
CallToolResult callToolResult = mcpSyncClient.callTool(callToolRequest);
assertThat(callToolResult).isNotNull().satisfies(result -> {
assertThat(result.content()).isNotNull();
assertThat(result.isError()).isNull();
});
}
@Test
@Timeout(15)
void testCallToolWithInvalidTool() {
CallToolRequest invalidRequest = new CallToolRequest("nonexistent_tool", Map.of("message", TEST_MESSAGE));
assertThatThrownBy(() -> mcpSyncClient.callTool(invalidRequest)).isInstanceOf(Exception.class);
}
@Test
@Timeout(15)
void testRootsListChanged() {
assertThatCode(() -> mcpSyncClient.sendRootsListChanged()).doesNotThrowAnyException();
}
@Test
@Timeout(15)
void testListResources() {
ListResourcesResult resources = mcpSyncClient.listResources(null);
assertThat(resources).isNotNull().satisfies(result -> {
assertThat(result.resources()).isNotNull();
if (!result.resources().isEmpty()) {
Resource firstResource = result.resources().get(0);
assertThat(firstResource.uri()).isNotNull();
assertThat(firstResource.name()).isNotNull();
}
});
}
@Test
void testClientSessionState() {
assertThat(mcpSyncClient).isNotNull();
}
}

View file

@ -1,132 +0,0 @@
/*
* 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.stdio;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* @author Christian Tzolov
*
*/
public class StdioServerTransportTests {
@Mock
private ServerParameters mockParams;
@Mock
private Process mockProcess;
@Mock
private BufferedReader mockErrorReader;
@Mock
private BufferedReader mockReader;
@Mock
private BufferedWriter mockWriter;
@Mock
ProcessBuilder mockProcessBuilder;
private StdioServerTransport stdioServerTransport;
@BeforeEach
public void setUp() {
MockitoAnnotations.openMocks(this);
when(mockParams.getCommand()).thenReturn("dummyCommand");
when(mockParams.getArgs()).thenReturn(List.of("arg1", "arg2"));
when(mockParams.getEnv()).thenReturn(new HashMap<>(Map.of("key", "value")));
stdioServerTransport = new StdioServerTransport(mockParams, new ObjectMapper()) {
@Override
protected ProcessBuilder getProcessBuilder() {
return mockProcessBuilder;
}
};
}
@Test
public void testStart_Success() throws IOException {
// Arrange
when(mockProcessBuilder.start()).thenReturn(mockProcess);
when(mockProcess.getInputStream()).thenReturn(mock(InputStream.class));
when(mockProcess.getOutputStream()).thenReturn(mock(OutputStream.class));
when(mockProcess.errorReader()).thenReturn(mockErrorReader);
when(mockProcess.inputReader()).thenReturn(mockReader);
when(mockProcess.outputWriter()).thenReturn(mockWriter);
// Act
stdioServerTransport.start();
// Assert
// TODO: assertThat(stdioServerTransport.isRunning).isTrue();
verify(mockProcessBuilder).start();
}
@Test
public void testStart_ProcessStartFailure() throws IOException {
// Arrange
when(mockProcessBuilder.start()).thenThrow(new IOException("Failed to start process"));
// Act & Assert
RuntimeException exception = assertThrows(RuntimeException.class, () -> stdioServerTransport.start());
assertThat(exception.getMessage()).contains("Failed to start process with command");
}
@Test
public void testStart_NullInputStream() throws IOException {
// Arrange
when(mockProcessBuilder.start()).thenReturn(mockProcess);
when(mockProcess.getInputStream()).thenReturn(null);
when(mockProcess.getOutputStream()).thenReturn(mock(OutputStream.class));
// Act & Assert
RuntimeException exception = assertThrows(RuntimeException.class, () -> stdioServerTransport.start());
assertThat(exception.getMessage()).contains("Process input or output stream is null");
verify(mockProcess).destroy();
}
@Test
public void testStart_NullOutputStream() throws IOException {
// Arrange
when(mockProcessBuilder.start()).thenReturn(mockProcess);
when(mockProcess.getInputStream()).thenReturn(mock(InputStream.class));
when(mockProcess.getOutputStream()).thenReturn(null);
// Act & Assert
RuntimeException exception = assertThrows(RuntimeException.class, () -> stdioServerTransport.start());
assertThat(exception.getMessage()).contains("Process input or output stream is null");
verify(mockProcess).destroy();
}
}

View file

@ -1,177 +0,0 @@
/*
* 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.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.McpSchema.JSONRPCMessage;
import spring.ai.experimental.mcp.spec.McpSchema.JSONRPCRequest;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
class DefaultMcpTransportTests {
private AbstractMcpTransport transport;
private ObjectMapper objectMapper;
@BeforeEach
void setUp() {
objectMapper = new ObjectMapper();
transport = new AbstractMcpTransport(objectMapper) {
@Override
public Mono<Void> closeGracefully() {
return null;
}
};
}
@Test
void constructorShouldValidateObjectMapper() {
assertThatThrownBy(() -> new AbstractMcpTransport(null) {
@Override
public Mono<Void> closeGracefully() {
return null;
}
})
.isInstanceOf(IllegalArgumentException.class)
.hasMessage("ObjectMapper must not be null");
}
@Test
void defaultConstructorShouldCreateValidInstance() {
AbstractMcpTransport defaultTransport = new AbstractMcpTransport() {
@Override
public Mono<Void> closeGracefully() {
return null;
}
};
assertThat(defaultTransport.getObjectMapper()).isNotNull();
}
@Test
void customObjectMapperConstructorShouldCreateValidInstance() {
assertThat(transport.getObjectMapper()).isSameAs(objectMapper);
}
@Test
void sendMessageShouldEmitToOutboundSink() {
JSONRPCRequest message = new JSONRPCRequest(
"2.0",
"test",
1,
null);
AtomicReference<JSONRPCMessage> receivedMessage = new AtomicReference<>();
transport.getOutboundSink().asFlux().subscribe(receivedMessage::set);
Mono<Void> result = transport.sendMessage(message);
StepVerifier.create(result)
.verifyComplete();
assertThat(receivedMessage.get())
.isNotNull()
.satisfies(msg -> {
assertThat(((JSONRPCRequest) msg).jsonrpc()).isEqualTo("2.0");
assertThat(((JSONRPCRequest) msg).method()).isEqualTo("test");
});
}
@Test
void customInboundMessageHandlerShouldReceiveMessages() {
AtomicReference<JSONRPCMessage> receivedMessage = new AtomicReference<>();
transport.setInboudMessageHandler(receivedMessage::set);
transport.start();
JSONRPCRequest message = new JSONRPCRequest(
"2.0",
"test",
1,
null);
transport.getInboundSink().tryEmitNext(message);
assertThat(receivedMessage.get())
.isNotNull()
.satisfies(msg -> {
assertThat(((JSONRPCRequest) msg).jsonrpc()).isEqualTo("2.0");
assertThat(((JSONRPCRequest) msg).method()).isEqualTo("test");
});
}
@Test
void customErrorHandlerShouldReceiveErrors() {
AtomicReference<String> receivedError = new AtomicReference<>();
transport.setInboundErrorHandler(receivedError::set);
transport.start();
String errorMessage = "Test error";
transport.getErrorSink().tryEmitNext(errorMessage);
assertThat(receivedError.get())
.isNotNull()
.isEqualTo(errorMessage);
}
@Test
void startShouldInitializeMessageAndErrorHandling() {
AtomicReference<JSONRPCMessage> receivedMessage = new AtomicReference<>();
AtomicReference<String> receivedError = new AtomicReference<>();
transport.setInboudMessageHandler(receivedMessage::set);
transport.setInboundErrorHandler(receivedError::set);
transport.start();
JSONRPCRequest message = new JSONRPCRequest(
"2.0",
"test",
1,
null);
String errorMessage = "Test error";
transport.getInboundSink().tryEmitNext(message);
transport.getErrorSink().tryEmitNext(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();
Mono<Void> result = transport.sendMessage(message);
StepVerifier.create(result)
.verifyErrorSatisfies(error -> assertThat(error)
.isInstanceOf(RuntimeException.class)
.hasMessage("Failed to enqueue message"));
}
}