RpcServer extends ch.squaredesk.nova.comm.rpc.RpcServer<Destination, String> { @Override public <T> Flowable<RpcInvocation<T>> requests(Destination destination, Class<T> requestType) { return messageReceiver.messages(destination, messageTranscriber.getIncomingMessageTranscriber(requestType)) .filter(this::isRpcRequest) .map(incomingRequest -> { Timer.Context timerContext = metricsCollector.requestReceived(incomingRequest.metaData.destination); return new RpcInvocation<>( incomingRequest, reply -> { SendInfo sendingInfo = new SendInfo( incomingRequest.metaData.details.correlationId, null, null, null, null, null); OutgoingMessageMetaData meta = new OutgoingMessageMetaData(incomingRequest.metaData.details.replyDestination, sendingInfo); messageSender.send(reply._1, meta).subscribe(); metricsCollector.requestCompleted(timerContext, reply); }, error -> metricsCollector.requestCompletedExceptionally(timerContext, incomingRequest.metaData.destination, error)); }); } RpcServer(String identifier,
              MessageReceiver messageReceiver,
              MessageSender messageSender,
              Metrics metrics);   RpcServer(String identifier,
              MessageReceiver messageReceiver,
              MessageSender messageSender,
              MessageTranscriber<String> messageTranscriber,
              Metrics metrics); @Override Flowable<RpcInvocation<T>> requests(Destination destination, Class<T> requestType);  }
@Test void completingRpcInvocationProperlyTriggersReplySending() throws Exception { Destination queue = jmsHelper.createQueue("completeRpc"); TestSubscriber<RpcInvocation<String>> testSubscriber = sut.requests(queue, String.class).test(); Message requestMessage = jmsHelper.sendRequest(queue, "Two"); int maxLoops = 10; for (int i = 0; i < maxLoops && testSubscriber.valueCount()==0; i++) { TimeUnit.MILLISECONDS.sleep(1000); } testSubscriber.values().iterator().next().complete("reply", s->s); assertThat(mySender.message, is("reply")); assertNotNull(mySender.sendingInfo); assertThat(mySender.sendingInfo.destination, sameInstance(requestMessage.getJMSReplyTo())); assertThat(mySender.sendingInfo.details.correlationId, sameInstance(requestMessage.getJMSCorrelationID())); assertNull(mySender.sendingInfo.details.replyDestination); assertNotNull(mySender.sendingInfo.details.correlationId); }
@Test void subribeToIncomingRequests() throws Exception { Destination queue = jmsHelper.createQueue("subscribeToRequests"); TestSubscriber<RpcInvocation<String>> testSubscriber = sut.requests(queue, String.class).test(); jmsHelper.sendMessage(queue,"One"); jmsHelper.sendRequest(queue,"Two"); jmsHelper.sendMessage(queue,"Three"); jmsHelper.sendRequest(queue,"Four"); jmsHelper.sendMessage(queue,"Five"); int maxLoops = 10; for (int i = 0; i < maxLoops && testSubscriber.valueCount() == 0; i++) { TimeUnit.MILLISECONDS.sleep(1000); } testSubscriber.assertValueCount(2); }