@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(); }
PublisherBuilder<? extends Message<?>> source() { MessageConsumer<Message<?>> consumer = vertx.eventBus().consumer(address); Multi<io.vertx.mutiny.core.eventbus.Message<Message<?>>> multi = consumer.toMulti(); if (broadcast) { multi = multi.broadcast().toAllSubscribers(); } return ReactiveStreams.fromPublisher(multi) .map(this::adapt); }
EventBusSource { PublisherBuilder<? extends Message<?>> source() { MessageConsumer<Message<?>> consumer = vertx.eventBus().consumer(address); Multi<io.vertx.mutiny.core.eventbus.Message<Message<?>>> multi = consumer.toMulti(); if (broadcast) { multi = multi.broadcast().toAllSubscribers(); } return ReactiveStreams.fromPublisher(multi) .map(this::adapt); } }
EventBusSource { PublisherBuilder<? extends Message<?>> source() { MessageConsumer<Message<?>> consumer = vertx.eventBus().consumer(address); Multi<io.vertx.mutiny.core.eventbus.Message<Message<?>>> multi = consumer.toMulti(); if (broadcast) { multi = multi.broadcast().toAllSubscribers(); } return ReactiveStreams.fromPublisher(multi) .map(this::adapt); } EventBusSource(Vertx vertx, VertxEventBusConnectorIncomingConfiguration config); }
EventBusSource { PublisherBuilder<? extends Message<?>> source() { MessageConsumer<Message<?>> consumer = vertx.eventBus().consumer(address); Multi<io.vertx.mutiny.core.eventbus.Message<Message<?>>> multi = consumer.toMulti(); if (broadcast) { multi = multi.broadcast().toAllSubscribers(); } return ReactiveStreams.fromPublisher(multi) .map(this::adapt); } EventBusSource(Vertx vertx, VertxEventBusConnectorIncomingConfiguration config);  }
EventBusSource { PublisherBuilder<? extends Message<?>> source() { MessageConsumer<Message<?>> consumer = vertx.eventBus().consumer(address); Multi<io.vertx.mutiny.core.eventbus.Message<Message<?>>> multi = consumer.toMulti(); if (broadcast) { multi = multi.broadcast().toAllSubscribers(); } return ReactiveStreams.fromPublisher(multi) .map(this::adapt); } EventBusSource(Vertx vertx, VertxEventBusConnectorIncomingConfiguration config);   }