@Test
public void testAcknowledgement() {

    String topic = UUID.randomUUID().toString();
    Map<String, Object> config = new HashMap<>();
    config.put("address", topic);
    config.put("use-reply-as-ack", true);
    EventBusSource source = new EventBusSource(vertx,
            new VertxEventBusConnectorIncomingConfiguration(new MapBasedConfig(config)));
    Multi<? extends EventBusMessage<?>> multi = Multi.createFrom().publisher(source.source().buildRs())
            .onItem().transform(x -> (EventBusMessage<?>) x)
            .onItem().transformToUniAndConcatenate(
                    m -> Uni.createFrom().completionStage(m.ack().toCompletableFuture().thenApply(x -> m)));
    List<EventBusMessage<?>> messages = new ArrayList<>();
    multi.subscribe().with(messages::add);
    AtomicBoolean acked = new AtomicBoolean();
    vertx.eventBus().request(topic, 1)
            .onItem().invoke(rep -> acked.set(true))
            .await().indefinitely();
    await().untilTrue(acked);
    assertThat(messages).isNotEmpty();
}