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