@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"); }
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 -> { 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(); }
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 -> { 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(); } }
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 -> { 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(); } EventBusSink(Vertx vertx, VertxEventBusConnectorOutgoingConfiguration config); }
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 -> { 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(); } EventBusSink(Vertx vertx, VertxEventBusConnectorOutgoingConfiguration config);  }
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 -> { 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(); } EventBusSink(Vertx vertx, VertxEventBusConnectorOutgoingConfiguration config);   }