@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); }
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);   }