JmsAdapter extends CommAdapter<String> { public Flowable<String> messages (Destination destination) { return this.messageReceiver.messages(destination) .filter(incomingMessage -> !incomingMessage.metaData.details.isRpcReply()) .map(incomingMessage -> incomingMessage.message); } 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 messagesStreamOnlyDeliversNonRpcMessages() { MyDestination origin = new MyDestination(); RetrieveInfo rpcInfo = new RetrieveInfo("c1", null, null); RetrieveInfo messageInfo = new RetrieveInfo(null, null, null); IncomingMessage[] messages = new IncomingMessage[] { new IncomingMessage<>("one", new IncomingMessageMetaData(origin, null, rpcInfo)), new IncomingMessage<>("two", new IncomingMessageMetaData(origin, null, messageInfo)), new IncomingMessage<>("three", new IncomingMessageMetaData(origin, null, messageInfo)), new IncomingMessage<>("four", new IncomingMessageMetaData(origin, null, messageInfo)), new IncomingMessage<>("five", new IncomingMessageMetaData(origin, null, rpcInfo)) }; TestSubscriber<String> subscriber = sut.messages(origin).test(); for (IncomingMessage msg: messages) { myMessageReceiver.publishSubject.onNext(msg); } assertThat(subscriber.valueCount(), is(3)); assertThat(subscriber.values().get(0), is(messages[1].message)); assertThat(subscriber.values().get(1), is(messages[2].message)); assertThat(subscriber.values().get(2), is(messages[3].message)); }