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);   }
@Test public void testSource() { String topic = UUID.randomUUID().toString(); Map<String, Object> config = new HashMap<>(); config.put("address", topic); EventBusSource source = new EventBusSource(vertx, new VertxEventBusConnectorIncomingConfiguration(new MapBasedConfig(config))); List<EventBusMessage<Integer>> messages = new ArrayList<>(); Multi.createFrom().publisher(source.source().buildRs()) .onItem().castTo(EventBusMessage.class) .subscribe().with(messages::add); AtomicInteger counter = new AtomicInteger(); new Thread(() -> usage.produceIntegers(topic, 10, true, null, counter::getAndIncrement)).start(); await().atMost(2, TimeUnit.MINUTES).until(() -> messages.size() >= 10); assertThat(messages.stream() .map(EventBusMessage::getPayload) .collect(Collectors.toList())) .containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9); }
@SuppressWarnings("unchecked") @Test public void testBroadcast() { String topic = UUID.randomUUID().toString(); Map<String, Object> config = new HashMap<>(); config.put("address", topic); config.put("broadcast", true); EventBusSource source = new EventBusSource(vertx, new VertxEventBusConnectorIncomingConfiguration(new MapBasedConfig(config))); List<EventBusMessage<Integer>> messages1 = new ArrayList<>(); List<EventBusMessage<Integer>> messages2 = new ArrayList<>(); Multi<EventBusMessage<Integer>> multi = Multi.createFrom().publisher(source.source().buildRs()) .onItem().transform(x -> (EventBusMessage<Integer>) x); multi.subscribe().with(messages1::add); multi.subscribe().with(messages2::add); AtomicInteger counter = new AtomicInteger(); new Thread(() -> usage.produceIntegers(topic, 10, true, null, counter::getAndIncrement)).start(); await().atMost(2, TimeUnit.MINUTES).until(() -> messages1.size() >= 10); await().atMost(2, TimeUnit.MINUTES).until(() -> messages2.size() >= 10); assertThat(messages1.stream().map(EventBusMessage::getPayload) .collect(Collectors.toList())) .containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9); assertThat(messages2.stream().map(EventBusMessage::getPayload) .collect(Collectors.toList())) .containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9); }
@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(); }
@Test public void testMessageHeader() { String topic = UUID.randomUUID().toString(); Map<String, Object> config = new HashMap<>(); config.put("address", topic); 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); List<EventBusMessage<?>> messages1 = new ArrayList<>(); multi.subscribe().with(messages1::add); vertx.eventBus().sendAndForget(topic, 1, new DeliveryOptions() .addHeader("X-key", "value")); await().until(() -> messages1.size() == 1); EventBusMessage<?> message = messages1.get(0); assertThat(message.getHeader("X-key")).contains("value"); assertThat(message.getHeaders("X-key")).containsExactly("value"); assertThat(message.getAddress()).isEqualTo(topic); assertThat(message.unwrap()).isNotNull(); assertThat(message.getReplyAddress()).isEmpty(); }