JmsAdapter extends CommAdapter<String> { public <T,U> Single<RpcReply<U>> sendRequest( Destination destination, Destination replyDestination, T request, Map<String, Object> customHeaders, Function<String, U> replyTranscriber, Integer deliveryMode, Integer priority, Long timeToLive, Duration timeout) { requireNonNull(replyDestination, "ReplyDestination must not be null"); Duration timeoutToUse = Optional.ofNullable(timeout).orElse(defaultRpcTimeout); int deliveryModeToUse = deliveryMode == null ? defaultMessageDeliveryMode : deliveryMode; int priorityToUse = priority == null ? defaultMessagePriority : priority; long timeToLiveToUse = timeToLive == null ? defaultMessageTimeToLive : timeToLive; timeToLiveToUse = Math.min(timeToLiveToUse, timeoutToUse.toMillis()); String correlationId = correlationIdGenerator.get(); SendInfo jmsSpecificInfo = new SendInfo( correlationId, replyDestination, customHeaders, deliveryModeToUse, priorityToUse, timeToLiveToUse); OutgoingMessageMetaData sendingInfo = new OutgoingMessageMetaData(destination, jmsSpecificInfo); return rpcClient.sendRequest( request, sendingInfo, messageTranscriber.getOutgoingMessageTranscriber(request), replyTranscriber, timeoutToUse) .doOnError(t -> examineSendExceptionForDeadDestinationAndInformListener(t, destination)); } JmsAdapter(Builder builder); Single<OutgoingMessageMetaData> sendMessage(Destination destination, T message); Single<OutgoingMessageMetaData> sendMessage(Destination destination, T message, Map<String, Object> customHeaders); Single<OutgoingMessageMetaData> sendMessage(Destination destination, T message, Map<String, Object> customHeaders, Integer deliveryMode, Integer priority, Long timeToLive); Single<OutgoingMessageMetaData> sendMessage(Destination destination, T message, Function<T, String> transcriber); Single<OutgoingMessageMetaData> sendMessage(Destination destination, T message, Function<T, String> transcriber, Map<String, Object> customHeaders); Single<OutgoingMessageMetaData> sendMessage(Destination destination, T message, Function<T, String> transcriber, Map<String, Object> customHeaders, Integer deliveryMode, Integer priority, Long timeToLive); Single<OutgoingMessageMetaData> sendMessage(Destination destination, String message); Single<OutgoingMessageMetaData> sendMessage(Destination destination, String message, Map<String, Object> customHeaders); Flowable<String> messages(Destination destination); Flowable<T> messages(Destination destination, Class<T> messageType); Flowable<T> messages(Destination destination, Function<String, T> messageTranscriber); Flowable<RpcInvocation<T>> requests(Destination destination, Class<T> requestType); Single<RpcReply<U>> sendRequest(
            Destination destination,
            Destination replyDestination,
            T request,
            Map<String, Object> customHeaders,
            Function<String, U> replyTranscriber,
            Integer deliveryMode, Integer priority, Long timeToLive,
            Duration timeout); Single<RpcReply<U>> sendRequest(
            Destination destination,
            Destination replyDestination,
            T message,
            Map<String, Object> customHeaders,
            Function<String, U> replyTranscriber,
            Duration timeout); Single<RpcReply<U>> sendRequest(
            Destination destination,
            T message,
            Map<String, Object> customHeaders,
            Function<String, U> replyTranscriber,
            Duration timeout); Single<RpcReply<U>> sendRequest(
            Destination destination,
            T message,
            Map<String, Object> customHeaders,
            Function<String, U> replyTranscriber); Single<RpcReply<U>> sendRequest(
            Destination destination,
            T message,
            Function<String, U> replyTranscriber,
            Duration timeout); Single<RpcReply<U>> sendRequest(
            Destination destination,
            T message,
            Function<String, U> replyTranscriber); Single<RpcReply<U>> sendRequest(
            Destination destination,
            Destination replyDestination,
            T request,
            Map<String, Object> customHeaders,
            Class<U> replyType,
            Integer deliveryMode, Integer priority, Long timeToLive,
            Duration timeout); Single<RpcReply<U>> sendRequest(
            Destination destination,
            Destination replyDestination,
            T message,
            Map<String, Object> customHeaders,
            Class<U> replyType,
            Duration timeout); Single<RpcReply<U>> sendRequest(
            Destination destination,
            T message,
            Map<String, Object> customHeaders,
            Class<U> replyType,
            Duration timeout); Single<RpcReply<U>> sendRequest(
            Destination destination,
            T message,
            Map<String, Object> customHeaders,
            Class<U> replyType); Single<RpcReply<U>> sendRequest(
            Destination destination,
            T message,
            Class<U> replyType,
            Duration timeout); Single<RpcReply<U>> sendRequest(
            Destination destination,
            T message,
            Class<U> replyType); void addDestinationListener(Consumer<Destination> destinationListener); void removeDestinationListener(Consumer<Destination> destinationListener); void shutdown(); void start(); static Builder builder();  }
@Test void nullDestinationThrows() { assertThrows(NullPointerException.class, ()-> sut.sendRequest(null, "some request", String.class)); }
@Test void jmsReplyToAddedToRequestSendingInfo() { sut.sendRequest(new MyDestination(), "message", String.class); assertNotNull(myRpcClient.outgoingMessageMetaData.details.replyDestination); }
@Test void jmsReplyToCanBeSpecified() { MyDestination destination = new MyDestination(); MyDestination replyToDestination = new MyDestination(); sut.sendRequest(destination, replyToDestination, "message", null, String.class, Duration.ofSeconds(1)); assertThat(myRpcClient.outgoingMessageMetaData.details.replyDestination, is(replyToDestination)); }
@Test void destinationMustBeProvidedOnRpc() { assertThrows(NullPointerException.class, () -> sut.sendRequest(null, "message", null, String.class, Duration.ofMillis(100))); }
@Test void defaultTimeoutUsedForRpcIfNoTimeoutProvided() { sut.sendRequest(new MyDestination(), "message", String.class); assertThat(myRpcClient.timeout, is(defaultRpcTimeout)); }
@Test void correlationIdGeneratedForRpcs() throws Exception { sut.sendRequest(new MyDestination(), "message", String.class); assertNotNull(myRpcClient.outgoingMessageMetaData.details.correlationId); }
@Test void destinationAndMessagePassedUnchangedForRpc() { MyDestination destination = new MyDestination(); sut.sendRequest(destination, "some message", String.class); assertThat(myRpcClient.request, is("some message")); assertThat(myRpcClient.outgoingMessageMetaData.destination, is(destination)); }