@Test public void testStreamMessage_validateRoutingPlanHeader() { final MessageRoutingPlan routingPlan = MessageRoutingPlan.builder() .withRoute("microservice1", "microservice1.queue.output", "blob=store", "runId", true) .withRoute("microservice2", "microservice2.queue.output", "blob=store", "runId", true) .withRoute("microservice3", "microservice3.queue.output", "blob=store", "runId", true) .withRoute("microservice4", "microservice4.queue.output", "blob=store", "runId", true) .withRoute("microservice5", "microservice5.queue.output", "blob=store", "runId", true) .build(); Mockito.when(resourceProvider.getServiceRegistryApi()).thenReturn(serviceRegistryApi); Mockito.when(resourceProvider.getQueueConfigurationClass()).thenReturn(QueueConfiguration.class); Mockito .when(serviceRegistryApi.getQueueConfiguration(eq(QueueConfiguration.class), anyString(), eq(context))) .thenReturn(mock(QueueConfiguration.class)); MessageWriter messageWriter = mock(MessageWriter.class); String messageGroup = "message-group"; Map<String, String> messageHeaders = mock(Map.class); Map<String, String> messageContext = mock(Map.class); messageSender.streamMessage(messageWriter, messageHeaders, messageContext, routingPlan, messageGroup); ArgumentCaptor<Map> actualMessageHeadersCaptor = ArgumentCaptor.forClass(Map.class); Mockito .verify(runtimeMessageSender) .streamMessage(eq(messageWriter), actualMessageHeadersCaptor.capture(), eq(messageGroup)); Map<String, String> actualMessageHeaders = actualMessageHeadersCaptor.getValue(); assertEquals( "Unexpected route properties in message headers", "blobstoreKeyHeaderName,runId,logInDebug,true,destinationQueueURI,microservice1.queue.output,blobstoreNameSpace,blob=store", actualMessageHeaders.get("__contextHeader__MS_API_OUTPUT_QUEUE_NAME_microservice1")); assertEquals( "Unexpected route properties in message headers", "blobstoreKeyHeaderName,runId,logInDebug,true,destinationQueueURI,microservice2.queue.output,blobstoreNameSpace,blob=store", actualMessageHeaders.get("__contextHeader__MS_API_OUTPUT_QUEUE_NAME_microservice2")); assertEquals( "Unexpected route properties in message headers", "blobstoreKeyHeaderName,runId,logInDebug,true,destinationQueueURI,microservice3.queue.output,blobstoreNameSpace,blob=store", actualMessageHeaders.get("__contextHeader__MS_API_OUTPUT_QUEUE_NAME_microservice3")); assertEquals( "Unexpected route properties in message headers", "blobstoreKeyHeaderName,runId,logInDebug,true,destinationQueueURI,microservice4.queue.output,blobstoreNameSpace,blob=store", actualMessageHeaders.get("__contextHeader__MS_API_OUTPUT_QUEUE_NAME_microservice4")); assertEquals( "Unexpected route properties in message headers", "blobstoreKeyHeaderName,runId,logInDebug,true,destinationQueueURI,microservice5.queue.output,blobstoreNameSpace,blob=store", actualMessageHeaders.get("__contextHeader__MS_API_OUTPUT_QUEUE_NAME_microservice5")); assertEquals( "Unexpected routing plan in message headers", "microservice1,microservice2,microservice3,microservice4,microservice5", actualMessageHeaders.get("__contextHeader__MS_API_ROUTING_PLAN")); }
@Override public void streamMessage(MessageWriter messageWriter, Map<String, String> messageHeaders) { streamMessage(messageWriter, messageHeaders, null, null, null); }
MessageSenderImpl implements MessageSender { @Override public void streamMessage(MessageWriter messageWriter, Map<String, String> messageHeaders) { streamMessage(messageWriter, messageHeaders, null, null, null); } }
MessageSenderImpl implements MessageSender { @Override public void streamMessage(MessageWriter messageWriter, Map<String, String> messageHeaders) { streamMessage(messageWriter, messageHeaders, null, null, null); } MessageSenderImpl(
            RuntimeMessageSender runtimeMessageSender,
            SerializationManager serializationManager,
            ResourceProvider resourceProvider,
            Context context); }
MessageSenderImpl implements MessageSender { @Override public void streamMessage(MessageWriter messageWriter, Map<String, String> messageHeaders) { streamMessage(messageWriter, messageHeaders, null, null, null); } MessageSenderImpl(
            RuntimeMessageSender runtimeMessageSender,
            SerializationManager serializationManager,
            ResourceProvider resourceProvider,
            Context context); @Override void streamMessage(MessageWriter messageWriter, Map<String, String> messageHeaders); @Override void streamMessage(MessageWriter messageWriter, Map<String, String> messageHeaders, String messageGroup); @Override void streamMessage(
            MessageWriter messageWriter,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext); @Override void streamMessage(
            MessageWriter messageWriter,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            String messageGroup); @Override void streamMessage(
            MessageWriter messageWriter,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            MessageRoutingPlan messageRoutingPlan); @Override void streamMessage(
            MessageWriter messageWriter,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            MessageRoutingPlan messageRoutingPlan,
            String messageGroup); @Override void sendMessage(Class<T> format, T objectForSerialization, Map<String, String> messageHeaders); @Override void sendMessage(
            Class<T> format,
            T objectForSerialization,
            Map<String, String> messageHeaders,
            String messageGroup); @Override void sendMessage(
            Class<T> format,
            T objectForSerialization,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext); @Override void sendMessage(
            Class<T> format,
            T objectForSerialization,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            String messageGroup); @Override void sendMessage(
            Class<T> format,
            final T objectForSerialization,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            MessageRoutingPlan messageRoutingPlan); @Override void sendMessage(
            Class<T> format,
            final T objectForSerialization,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            MessageRoutingPlan messageRoutingPlan,
            String messageGroup); }
MessageSenderImpl implements MessageSender { @Override public void streamMessage(MessageWriter messageWriter, Map<String, String> messageHeaders) { streamMessage(messageWriter, messageHeaders, null, null, null); } MessageSenderImpl(
            RuntimeMessageSender runtimeMessageSender,
            SerializationManager serializationManager,
            ResourceProvider resourceProvider,
            Context context); @Override void streamMessage(MessageWriter messageWriter, Map<String, String> messageHeaders); @Override void streamMessage(MessageWriter messageWriter, Map<String, String> messageHeaders, String messageGroup); @Override void streamMessage(
            MessageWriter messageWriter,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext); @Override void streamMessage(
            MessageWriter messageWriter,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            String messageGroup); @Override void streamMessage(
            MessageWriter messageWriter,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            MessageRoutingPlan messageRoutingPlan); @Override void streamMessage(
            MessageWriter messageWriter,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            MessageRoutingPlan messageRoutingPlan,
            String messageGroup); @Override void sendMessage(Class<T> format, T objectForSerialization, Map<String, String> messageHeaders); @Override void sendMessage(
            Class<T> format,
            T objectForSerialization,
            Map<String, String> messageHeaders,
            String messageGroup); @Override void sendMessage(
            Class<T> format,
            T objectForSerialization,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext); @Override void sendMessage(
            Class<T> format,
            T objectForSerialization,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            String messageGroup); @Override void sendMessage(
            Class<T> format,
            final T objectForSerialization,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            MessageRoutingPlan messageRoutingPlan); @Override void sendMessage(
            Class<T> format,
            final T objectForSerialization,
            Map<String, String> messageHeaders,
            Map<String, String> messageContext,
            MessageRoutingPlan messageRoutingPlan,
            String messageGroup);  }