RpcClient extends ch.squaredesk.nova.comm.rpc.RpcClient<String, OutgoingMessageMetaData, IncomingMessageMetaData> { @Override public <RequestType, ReplyType> Single<RpcReply<ReplyType>> sendRequest( RequestType request, OutgoingMessageMetaData requestMetaData, Function<RequestType, String> requestTranscriber, Function<String, ReplyType> replyTranscriber, Duration timeout) { requireNonNull(timeout, "timeout must not be null"); requireNonNull(requestMetaData, "metaData must not be null"); requireNonNull(requestMetaData.details, "metaData.details must not be null"); requireNonNull(requestMetaData.details.correlationId, "correlationId must not be null"); requireNonNull(requestMetaData.details.replyDestination, "replyDestination must not be null"); String metricsInfo = requestMetaData.destination + "." + request; Single<RpcReply<ReplyType>> replySingle = messageReceiver.messages(requestMetaData.details.replyDestination, replyTranscriber) .filter(incomingMessage -> incomingMessage.metaData.details != null && requestMetaData.details.correlationId.equals(incomingMessage.metaData.details.correlationId)) .take(1) .doOnNext(reply -> metricsCollector.rpcCompleted(metricsInfo, reply)) .map(incomingMessage -> new RpcReply<>(incomingMessage.message, incomingMessage.metaData)) .singleOrError(); try { messageSender .send(request, requestMetaData, requestTranscriber) .blockingGet(); } catch (Exception e) { return Single.error(e); } return replySingle .timeout(timeout.toMillis(), TimeUnit.MILLISECONDS) .doOnError(t -> { if (t instanceof TimeoutException) { metricsCollector.rpcTimedOut(metricsInfo); } }); } RpcClient(String identifier,
                     MessageSender messageSender,
                     MessageReceiver messageReceiver,
                     Metrics metrics); @Override Single<RpcReply<ReplyType>> sendRequest(
            RequestType request,
            OutgoingMessageMetaData requestMetaData,
            Function<RequestType, String> requestTranscriber,
            Function<String, ReplyType> replyTranscriber,
            Duration timeout);  }
@Test void noReplyWithinTimeoutReturnsTimeoutException() throws Exception { Destination queue = jmsHelper.createQueue("request1"); Destination replyTo = jmsHelper.createQueue("replyTo1"); SendInfo sendingDetails = new SendInfo("c1", replyTo, null, Message.DEFAULT_DELIVERY_MODE, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE); OutgoingMessageMetaData metaData = new OutgoingMessageMetaData(queue, sendingDetails); TestObserver<RpcReply<String>> replyObserver = sut.sendRequest("aRequest", metaData, s -> s, s -> s, Duration.ofMillis(250)).test().await(); replyObserver.assertError(TimeoutException.class); }
@Test void replyProperlyDispatchedToCaller() throws Exception { Destination queue = jmsHelper.createQueue("request2"); Destination replyTo = jmsHelper.createQueue("replyTo2"); SendInfo sendingDetails = new SendInfo("c2", replyTo, null, Message.DEFAULT_DELIVERY_MODE, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE); OutgoingMessageMetaData metaData = new OutgoingMessageMetaData(queue, sendingDetails); TestObserver<RpcReply<String>> replyObserver = sut.sendRequest("aRequest", metaData, s -> s, s -> s, Duration.ofSeconds(20)).test(); jmsHelper.sendReply(replyTo, "aReply", "c2"); replyObserver.await(21, SECONDS); replyObserver.assertComplete(); replyObserver.assertValueCount(1); assertThat(replyObserver.values().get(0).result, is("aReply")); }
@Test void subscriptionAndRpcRepliesSupportedOnSameQueue() throws Exception { Destination queue = jmsHelper.createQueue("request3"); Destination sharedQueue = jmsHelper.createQueue("someDest"); SendInfo sendingDetails = new SendInfo("c3", sharedQueue, null, Message.DEFAULT_DELIVERY_MODE, Message.DEFAULT_PRIORITY, Message.DEFAULT_TIME_TO_LIVE); OutgoingMessageMetaData metaData = new OutgoingMessageMetaData(queue, sendingDetails); TestObserver<RpcReply<String>> replyObserver = sut.sendRequest("aRequest", metaData, s -> s, s -> s, Duration.ofSeconds(20)).test(); jmsHelper.sendReply(sharedQueue, "aReply1", null); jmsHelper.sendReply(sharedQueue, "aReply2", "c3"); jmsHelper.sendReply(sharedQueue, "aReply3", null); replyObserver.await(21, SECONDS); replyObserver.assertComplete(); replyObserver.assertValueCount(1); assertThat(replyObserver.values().get(0).result, is("aReply2")); }