@Test
@SuppressWarnings("unchecked")
public void testExpectReply() {

    String topic = UUID.randomUUID().toString();

    List<Integer> integers = new ArrayList<>();
    AtomicReference<io.vertx.mutiny.core.eventbus.Message<Integer>> last = new AtomicReference<>();
    vertx.eventBus().<Integer> consumer(topic, m -> {
        last.set(m);
        if (m.body() < 8) {
            integers.add(m.body());
            m.replyAndForget("foo");
        }
    });

    Map<String, Object> config = new HashMap<>();
    config.put("address", topic);
    config.put("expect-reply", true);
    EventBusSink sink = new EventBusSink(vertx,
            new VertxEventBusConnectorOutgoingConfiguration(new MapBasedConfig(config)));

    SubscriberBuilder<? extends Message<?>, Void> subscriber = sink.sink();
    Multi.createFrom().range(0, 10)
            .map(Message::of)
            .subscribe((Subscriber<Message<?>>) subscriber.build());

    await().until(() -> integers.size() == 8 && last.get().body() == 8);
    last.get().replyAndForget("bar");
    await().until(() -> last.get().body() == 9);
    assertThat(last.get().body()).isEqualTo(9);
    last.get().replyAndForget("baz");
}