KafkaAdapter extends CommAdapter<String> { public Single<OutgoingMessageMetaData> sendMessage(String destination, String message) { SendInfo sendInfo = new SendInfo(); OutgoingMessageMetaData meta = new OutgoingMessageMetaData(destination, sendInfo); return messageSender.send(message, meta); } 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();  }
@Test void messageMarshallingErrorOnSendForwardedToSubscriber(KafkaHelper kafkaHelper) throws Exception { Single<OutgoingMessageMetaData> completable = sut.sendMessage("dest", "myMessage", s -> { throw new RuntimeException("for test"); }); TestObserver<OutgoingMessageMetaData> observer = completable.test(); observer.await(); observer.assertError(RuntimeException.class); observer.assertErrorMessage("for test"); }
@Test void sendMessage(KafkaHelper kafkaHelper) throws Exception { String topic = "topicForSendTest"; ensureTopicsExists(topic); sut.sendMessage(topic, "One").blockingGet(); sut.sendMessage(topic, "Two").blockingGet(); sut.sendMessage(topic, "Three").blockingGet(); List<String> messages = kafkaHelper.consumeStrings(topic, 3).get(10, SECONDS); assertThat(messages.size(), is(3)); assertThat(messages, contains("One", "Two", "Three")); }
@Test void sendingCanBeDoneFromMultipleThreads(KafkaHelper kafkaHelper) throws Exception { String topic = "topicForMultiThreadSendTest"; ensureTopicsExists(topic); class Sender extends Thread { private final String id; public Sender(String id) { super("MultiThreadedSendTestSender-" + id); this.id = id; } @Override public void run() { try { sut.sendMessage(topic, "One-" + id).blockingGet(); sut.sendMessage(topic, "Two-" + id).blockingGet(); sut.sendMessage(topic, "Three-" + id).blockingGet(); } catch (Exception e) { e.printStackTrace(); } } } Sender sender1 = new Sender("1"); Sender sender2 = new Sender("2"); sender1.start(); sender2.start(); sender1.join(); sender2.join(); List<String> messages = kafkaHelper.consumeStrings(topic, 6).get(20, SECONDS); assertThat(messages.size(), is(6)); assertThat(messages, containsInAnyOrder("One-1", "Two-1", "Three-1", "One-2", "Two-2", "Three-2")); }