class EventBusSink {

    SubscriberBuilder<? extends Message<?>, Void> sink() {
        DeliveryOptions options = new DeliveryOptions();
        if (this.codec != null) {
            options.setCodecName(this.codec);
        }
        if (this.timeout != -1) {
            options.setSendTimeout(this.timeout);
        }

        return ReactiveStreams.<Message<?>> builder()
                .flatMapCompletionStage(msg -> {
                    // TODO support getting an EventBusMessage as message.
                    if (!this.publish) {
                        if (expectReply) {
                            return vertx.eventBus().request(address, msg.getPayload(), options).subscribeAsCompletionStage()
                                    .thenApply(m -> msg);
                        } else {
                            vertx.eventBus().sendAndForget(address, msg.getPayload(), options);
                            return CompletableFuture.completedFuture(msg);
                        }
                    } else {
                        vertx.eventBus().publish(address, msg.getPayload(), options);
                        return CompletableFuture.completedFuture(msg);
                    }
                })
                .ignore();
    }

}

class EventBusSinkTest  {

    private WeldContainer container;

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

        String topic = UUID.randomUUID().toString();

        List<Integer> integers = new ArrayList<>();
        AtomicReference<io.vertx.mutiny.core.eventbus.Message<Integer>> last = new AtomicReference<>();
        vertx.eventBus().<Integer> consumer(topic, m -> {
            last.set(m);
            if (m.body() < 8) {
                integers.add(m.body());
                m.replyAndForget("foo");
            }
        });

        Map<String, Object> config = new HashMap<>();
        config.put("address", topic);
        config.put("expect-reply", true);
        EventBusSink sink = new EventBusSink(vertx,
                new VertxEventBusConnectorOutgoingConfiguration(new MapBasedConfig(config)));

        SubscriberBuilder<? extends Message<?>, Void> subscriber = sink.sink();
        Multi.createFrom().range(0, 10)
                .map(Message::of)
                .subscribe((Subscriber<Message<?>>) subscriber.build());

        await().until(() -> integers.size() == 8 && last.get().body() == 8);
        last.get().replyAndForget("bar");
        await().until(() -> last.get().body() == 9);
        assertThat(last.get().body()).isEqualTo(9);
        last.get().replyAndForget("baz");
    }

}