KafkaAdapter extends CommAdapter<String> { public Flowable<String> messages (String destination) { return messageReceiver.messages(destination).map(incomingMessage -> incomingMessage.message); } KafkaAdapter(MessageSender messageSender,
                 MessageReceiver messageReceiver,
                 MessageTranscriber<String> messageTranscriber,
                 Metrics metrics); Single<OutgoingMessageMetaData> sendMessage(String destination, String message); Single<OutgoingMessageMetaData> sendMessage(String destination, T message); Single<OutgoingMessageMetaData> sendMessage(String destination, T message, Function<T, String> transcriber); Flowable<String> messages(String destination); Flowable<T> messages(String destination, Class<T> messageType); Flowable<T> messages(String destination, Function<String,T> messageTranscriber); void shutdown(); static Builder builder();  }
@Disabled @Test void resubscriptionWorks(KafkaHelper kafkaHelper) throws Exception { String topic = "topic4SubsTest"; ensureTopicsExists(topic); kafkaHelper.produceStrings(topic, "One", "Two"); AtomicInteger counter = new AtomicInteger(0); CountDownLatch cdl = new CountDownLatch(2); List<String> messages = new ArrayList<>(); Disposable subscription1 = sut.messages(topic).subscribe( x -> { messages.add(x); counter.incrementAndGet(); cdl.countDown(); } ); cdl.await(60, SECONDS); assertThat(cdl.getCount(),is(0L)); assertThat(counter.get(), is(2)); assertThat(messages, containsInAnyOrder("One", "Two")); subscription1.dispose(); CountDownLatch cdl2 = new CountDownLatch(1); List<String> messages2 = new ArrayList<>(); Disposable subscription2 = sut.messages(topic).subscribe(x -> { messages2.add(x); cdl2.countDown(); }); kafkaHelper.produceStrings(topic, "Three"); cdl2.await(10, SECONDS); assertThat(counter.get(), is(2)); assertThat(cdl2.getCount(), is(0L)); assertThat(messages, containsInAnyOrder("One", "Two")); assertThat(messages2, anyOf(contains("Three"), containsInAnyOrder("One", "Two", "Three"))); subscription2.dispose(); }
@Test void multipleTopicsProperlySupported(KafkaHelper kafkaHelper) throws Exception { String topicEven = "even"; String topicOdd = "odd"; ensureTopicsExists(topicEven, topicOdd); CountDownLatch cdlEven = new CountDownLatch(4); CountDownLatch cdlOdd = new CountDownLatch(6); Consumer<String> messageConsumerOdd = msg -> cdlOdd.countDown(); Consumer<String> messageConsumerEven = msg -> cdlEven.countDown(); sut.messages(topicEven).subscribe(messageConsumerEven); sut.messages(topicEven).subscribe(messageConsumerEven); sut.messages(topicOdd).subscribe(messageConsumerOdd); sut.messages(topicOdd).subscribe(messageConsumerOdd); kafkaHelper.produceStrings(topicOdd, "1", "3", "5"); kafkaHelper.produceStrings(topicEven, "2", "4"); cdlEven.await(20, SECONDS); cdlOdd.await(20, SECONDS); assertThat(cdlEven.getCount(), is (0L)); assertThat(cdlOdd.getCount(), is (0L)); }
@Test void errorInMessageHandlingKillsSubscription(KafkaHelper kafkaHelper) throws Exception { String topic = "topic4SubsErrorTest"; ensureTopicsExists(topic); KafkaProducer<String, String> stringProducer = kafkaHelper.createStringProducer(); stringProducer.send(new ProducerRecord<>(topic, "1")).get(); logger.error("Sent 1"); stringProducer.send(new ProducerRecord<>(topic, "Two")).get(); logger.error("Sent Two"); stringProducer.send(new ProducerRecord<>(topic, "3")).get(); logger.error("Sent 3"); AtomicInteger counter = new AtomicInteger(); List<Integer> messagesFromSut = new ArrayList<>(); sut.messages(topic).subscribe( x -> { logger.error("Received " + x); try { messagesFromSut.add(Integer.parseInt(x)); } finally { counter.incrementAndGet(); } } ); List<String> messagesFromBroker = kafkaHelper.consumeStrings(topic, 3).get(20, SECONDS); assertThat(messagesFromBroker, contains("1", "Two", "3")); SECONDS.sleep(10); assertThat(counter.get(),is(2)); assertThat(messagesFromSut, containsInAnyOrder(1)); }
@Test void errorInMessageHandlingForOneSubscriptionDoesNotAffectOtherSubscriptions(KafkaHelper kafkaHelper) throws Exception { String topic = "topic4SubsErrorMultipleTest"; ensureTopicsExists(topic); AtomicInteger counterBrokenSubscriprion = new AtomicInteger(); CountDownLatch cdlGood = new CountDownLatch(3); List<Integer> messagesGood = new ArrayList<>(); List<Integer> messagesBroken = new ArrayList<>(); sut.messages(topic).subscribe( x -> { try { messagesBroken.add(Integer.parseInt(x)); } finally { counterBrokenSubscriprion.incrementAndGet(); } } ); sut.messages(topic).subscribe( x -> { try { messagesGood.add(Integer.parseInt(x)); } catch (Exception e) { } finally { cdlGood.countDown(); } } ); KafkaProducer<String, String> stringProducer = kafkaHelper.createStringProducer(); stringProducer.send(new ProducerRecord<>(topic, "1")).get(); stringProducer.send(new ProducerRecord<>(topic, "Two")).get(); stringProducer.send(new ProducerRecord<>(topic, "3")).get(); cdlGood.await(10, SECONDS); assertThat(cdlGood.getCount(),is(0L)); assertThat(counterBrokenSubscriprion.get(),is(2)); assertThat(messagesGood, containsInAnyOrder(1, 3)); }