@Test @SuppressWarnings("unchecked") public void testCodec() { String topic = UUID.randomUUID().toString(); List<Person> persons = new ArrayList<>(); vertx.eventBus().<Person> consumer(topic, m -> persons.add(m.body())); vertx.eventBus().getDelegate().registerCodec(new PersonCodec()); Map<String, Object> config = new HashMap<>(); config.put("address", topic); config.put("codec", "PersonCodec"); EventBusSink sink = new EventBusSink(vertx, new VertxEventBusConnectorOutgoingConfiguration(new MapBasedConfig(config))); SubscriberBuilder<? extends Message<?>, Void> subscriber = sink.sink(); Multi.createFrom().range(0, 10) .map(i -> new Person().setName("name").setAge(i)) .map(Message::of) .subscribe((Subscriber<Message<?>>) subscriber.build()); await().until(() -> persons.size() == 10); assertThat(persons.size()).isEqualTo(10); }
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);   }