@Test public void testSendWithCallbackSucceedsAndReturnsSucceeded() { SpringHttpReplyMessage<String> testReplyMessage = new SpringHttpReplyMessage<>(COMMAND_MESSAGE.getIdentifier(), COMMAND_RESULT, serializer); ResponseEntity<SpringHttpReplyMessage<String>> testResponseEntity = new ResponseEntity<>(testReplyMessage, HttpStatus.OK); when(restTemplate.exchange(eq(expectedUri), eq(HttpMethod.POST), any(), argThat(new ParameterizedTypeReferenceMatcher<String>())) ).thenReturn(testResponseEntity); testSubject.send(DESTINATION, COMMAND_MESSAGE, commandCallback); verify(executor).execute(any()); verify(serializer).serialize(COMMAND_MESSAGE.getMetaData(), byte[].class); verify(serializer).serialize(COMMAND_MESSAGE.getPayload(), byte[].class); HttpEntity<SpringHttpDispatchMessage<String>> expectedHttpEntity = new HttpEntity<>(buildDispatchMessage(true)); verify(restTemplate).exchange(eq(expectedUri), eq(HttpMethod.POST), eq(expectedHttpEntity), argThat(new ParameterizedTypeReferenceMatcher<>())); SerializedObject<byte[]> serializedPayload = serializer.serialize(COMMAND_RESULT.getPayload(), byte[].class); SerializedObject<byte[]> serializedMetaData = serializer.serialize(COMMAND_RESULT.getMetaData(), byte[].class); SerializedObject<byte[]> serializedException = serializer.serialize(COMMAND_RESULT.optionalExceptionResult() .orElse(null), byte[].class); ArgumentCaptor<SerializedObject<byte[]>> serializedObjectCaptor = ArgumentCaptor.forClass(SerializedObject.class); verify(serializer, times(3)).deserialize(serializedObjectCaptor.capture()); assertEquals(serializedPayload.getType(), serializedObjectCaptor.getAllValues().get(0).getType()); assertEquals(serializedPayload.getContentType(), serializedObjectCaptor.getAllValues().get(0).getContentType()); assertArrayEquals(serializedPayload.getData(), serializedObjectCaptor.getAllValues().get(0).getData()); assertEquals(serializedException.getType(), serializedObjectCaptor.getAllValues().get(1).getType()); assertEquals(serializedException.getContentType(), serializedObjectCaptor.getAllValues().get(1).getContentType()); assertArrayEquals(serializedException.getData(), serializedObjectCaptor.getAllValues().get(1).getData()); assertEquals(serializedMetaData.getType(), serializedObjectCaptor.getAllValues().get(2).getType()); assertEquals(serializedMetaData.getContentType(), serializedObjectCaptor.getAllValues().get(2).getContentType()); assertArrayEquals(serializedMetaData.getData(), serializedObjectCaptor.getAllValues().get(2).getData()); ArgumentCaptor<CommandMessage<? extends String>> commandMessageArgumentCaptor = ArgumentCaptor.forClass(CommandMessage.class); ArgumentCaptor<CommandResultMessage<? extends String>> commandResultMessageArgumentCaptor = ArgumentCaptor.forClass(CommandResultMessage.class); verify(commandCallback).onResult(commandMessageArgumentCaptor.capture(), commandResultMessageArgumentCaptor.capture()); assertEquals(COMMAND_MESSAGE.getMetaData(), commandMessageArgumentCaptor.getValue().getMetaData()); assertEquals(COMMAND_MESSAGE.getPayload(), commandMessageArgumentCaptor.getValue().getPayload()); assertEquals(COMMAND_RESULT.getMetaData(), commandResultMessageArgumentCaptor.getValue().getMetaData()); assertEquals(COMMAND_RESULT.getPayload(), commandResultMessageArgumentCaptor.getValue().getPayload()); }
@Override public <C> void send(Member destination, CommandMessage<? extends C> commandMessage) { shutdownLatch.ifShuttingDown("JGroupsConnector is shutting down, no new commands will be sent."); if (destination.local()) { localCommandBus.dispatch(commandMessage); } else { executor.execute(() -> sendRemotely(destination, commandMessage, DO_NOT_EXPECT_REPLY)); } }
SpringHttpCommandBusConnector implements CommandBusConnector { @Override public <C> void send(Member destination, CommandMessage<? extends C> commandMessage) { shutdownLatch.ifShuttingDown("JGroupsConnector is shutting down, no new commands will be sent."); if (destination.local()) { localCommandBus.dispatch(commandMessage); } else { executor.execute(() -> sendRemotely(destination, commandMessage, DO_NOT_EXPECT_REPLY)); } } }
SpringHttpCommandBusConnector implements CommandBusConnector { @Override public <C> void send(Member destination, CommandMessage<? extends C> commandMessage) { shutdownLatch.ifShuttingDown("JGroupsConnector is shutting down, no new commands will be sent."); if (destination.local()) { localCommandBus.dispatch(commandMessage); } else { executor.execute(() -> sendRemotely(destination, commandMessage, DO_NOT_EXPECT_REPLY)); } } protected  SpringHttpCommandBusConnector(Builder builder); }
SpringHttpCommandBusConnector implements CommandBusConnector { @Override public <C> void send(Member destination, CommandMessage<? extends C> commandMessage) { shutdownLatch.ifShuttingDown("JGroupsConnector is shutting down, no new commands will be sent."); if (destination.local()) { localCommandBus.dispatch(commandMessage); } else { executor.execute(() -> sendRemotely(destination, commandMessage, DO_NOT_EXPECT_REPLY)); } } protected  SpringHttpCommandBusConnector(Builder builder); @StartHandler(phase = Phase.EXTERNAL_CONNECTIONS) void start(); static Builder builder(); @Override void send(Member destination, CommandMessage<? extends C> commandMessage); @Override void send(Member destination,
                            CommandMessage<C> commandMessage,
                            CommandCallback<? super C, R> callback); @Override CompletableFuture<Void> initiateShutdown(); @Override Registration subscribe(String commandName, MessageHandler<? super CommandMessage<?>> handler); @Override Optional<CommandBus> localSegment(); @PostMapping("/command") CompletableFuture<?> receiveCommand(@RequestBody SpringHttpDispatchMessage<C> dispatchMessage); @Override Registration registerHandlerInterceptor(
            MessageHandlerInterceptor<? super CommandMessage<?>> handlerInterceptor); }
SpringHttpCommandBusConnector implements CommandBusConnector { @Override public <C> void send(Member destination, CommandMessage<? extends C> commandMessage) { shutdownLatch.ifShuttingDown("JGroupsConnector is shutting down, no new commands will be sent."); if (destination.local()) { localCommandBus.dispatch(commandMessage); } else { executor.execute(() -> sendRemotely(destination, commandMessage, DO_NOT_EXPECT_REPLY)); } } protected  SpringHttpCommandBusConnector(Builder builder); @StartHandler(phase = Phase.EXTERNAL_CONNECTIONS) void start(); static Builder builder(); @Override void send(Member destination, CommandMessage<? extends C> commandMessage); @Override void send(Member destination,
                            CommandMessage<C> commandMessage,
                            CommandCallback<? super C, R> callback); @Override CompletableFuture<Void> initiateShutdown(); @Override Registration subscribe(String commandName, MessageHandler<? super CommandMessage<?>> handler); @Override Optional<CommandBus> localSegment(); @PostMapping("/command") CompletableFuture<?> receiveCommand(@RequestBody SpringHttpDispatchMessage<C> dispatchMessage); @Override Registration registerHandlerInterceptor(
            MessageHandlerInterceptor<? super CommandMessage<?>> handlerInterceptor);  }