@SuppressWarnings("unchecked")
@Test
public void testSinkUsingInteger() {

    String topic = UUID.randomUUID().toString();
    AtomicInteger expected = new AtomicInteger(0);
    usage.consumeIntegers(topic, 10, 10, TimeUnit.SECONDS,
            v -> expected.getAndIncrement());
    Map<String, Object> config = new HashMap<>();
    config.put("address", topic);
    EventBusSink sink = new EventBusSink(vertx,
            new VertxEventBusConnectorOutgoingConfiguration(new MapBasedConfig(config)));
    SubscriberBuilder<? extends Message<?>, Void> subscriber = sink.sink();
    Multi.createFrom().range(0, 10)
            .map(v -> (Message<?>) Message.of(v))
            .subscribe((Subscriber<Message<?>>) subscriber.build());
    await().untilAtomic(expected, is(10));
    assertThat(expected).hasValue(10);
}