MessageReceiver extends ch.squaredesk.nova.comm.retrieving.MessageReceiver<Destination, String, IncomingMessageMetaData> { @Override public Flowable<IncomingMessage<String, IncomingMessageMetaData>> messages(Destination destination) { Objects.requireNonNull(destination, "destination must not ne bull"); Objects.requireNonNull(destination, "unmarshaller must not ne bull"); String destinationId = jmsObjectRepository.idFor(destination); return mapDestinationIdToMessageStream.computeIfAbsent(destinationId, key -> { Flowable<IncomingMessage<String, IncomingMessageMetaData>> f = Flowable.generate( () -> { logger.info("Opening connection to destination {}", destinationId); metricsCollector.subscriptionCreated(destinationId); return jmsObjectRepository.createMessageConsumer(destination); }, (consumer, emitter) -> { IncomingMessage<String, IncomingMessageMetaData> incomingMessage = null; while (incomingMessage == null) { Message m = null; try { metricsCollector.messageReceived(destinationId); m = consumer.receive(); } catch (Exception e) { } if (m == null) { logger.info("Unable to receive message from consumer for destination {}. Closing the connection...", destinationId); emitter.onComplete(); return; } if (!(m instanceof TextMessage)) { logger.error("Unsupported type of incoming message {}", m); metricsCollector.unparsableMessageReceived(destinationId); continue; } String transportMessage; try { transportMessage = ((TextMessage) m).getText(); } catch (Exception e) { logger.error("Unable to read incoming message " + m, e); metricsCollector.unparsableMessageReceived(destinationId); continue; } IncomingMessageMetaData meta = messageDetailsCreator.createIncomingMessageMetaData(m); incomingMessage = new IncomingMessage<>(transportMessage, meta); metricsCollector.messageReceived(destinationId); } emitter.onNext(incomingMessage); }, consumer -> { metricsCollector.subscriptionDestroyed(destinationId); jmsObjectRepository.destroyConsumer(consumer); mapDestinationIdToMessageStream.remove(destinationId); logger.info("Closed connection to destination {}", destinationId); } ); return f.subscribeOn(Schedulers.io()) .share(); }); } MessageReceiver(String identifier,
                    JmsObjectRepository jmsObjectRepository,
                    Metrics metrics); @Override Flowable<IncomingMessage<String, IncomingMessageMetaData>> messages(Destination destination);  }
@Test void messagesCanBeRetrievedFromQueue() throws Exception { CountDownLatch cdl = new CountDownLatch(5); List<String> receivedMessages = new ArrayList<>(); Destination destination = jmsHelper.createQueue("1"); sut.messages(destination) .map(im -> im.message) .subscribe(msg -> { receivedMessages.add(msg); cdl.countDown(); }); for (String msg: Arrays.asList("One", "Two", "3", "For", "Fve")) { jmsHelper.sendMessage(destination, msg); } cdl.await(5, TimeUnit.SECONDS); assertThat(cdl.getCount(), is(0L)); assertThat(receivedMessages, contains("One", "Two", "3", "For", "Fve")); }
@Test void messagesCanBeRetrievedByMultipleSubscribers() throws Exception { int numSubscribers = 5; CountDownLatch cdl = new CountDownLatch(5 * numSubscribers); List<String>[] receivedMessages = new ArrayList[numSubscribers]; Destination destination = jmsHelper.createQueue("2"); for (int i = 0; i < numSubscribers; i++) { int idx = i; receivedMessages[i] = new ArrayList<>(); sut.messages(destination) .map(im -> im.message) .subscribe(msg -> { receivedMessages[idx].add(msg); cdl.countDown(); }); } for (String msg: Arrays.asList("One", "Two", "3", "For", "Fve")) { jmsHelper.sendMessage(destination, msg); } cdl.await(5, TimeUnit.SECONDS); assertThat(cdl.getCount(), is(0L)); for (int i = 0; i < numSubscribers; i++) { assertThat(receivedMessages[i], contains("One", "Two", "3", "For", "Fve")); } }
@Test void subscribeWithNullDestinationThrows() throws Exception { assertThrows(NullPointerException.class, () -> sut.messages(null)); }