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);   }
@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); }
@SuppressWarnings("unchecked") @Test public void testSinkUsingString() { String topic = UUID.randomUUID().toString(); AtomicInteger expected = new AtomicInteger(0); usage.consumeStrings(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(i -> Integer.toString(i)) .map(Message::of) .subscribe((Subscriber<Message<?>>) subscriber.build()); await().untilAtomic(expected, is(10)); assertThat(expected).hasValue(10); }
@SuppressWarnings("unchecked") @Test public void testPublish() { String topic = UUID.randomUUID().toString(); AtomicInteger expected1 = new AtomicInteger(0); usage.consumeStrings(topic, 10, 10, TimeUnit.SECONDS, v -> expected1.getAndIncrement()); AtomicInteger expected2 = new AtomicInteger(0); usage.consumeStrings(topic, 10, 10, TimeUnit.SECONDS, v -> expected2.getAndIncrement()); Map<String, Object> config = new HashMap<>(); config.put("address", topic); config.put("publish", true); EventBusSink sink = new EventBusSink(vertx, new VertxEventBusConnectorOutgoingConfiguration(new MapBasedConfig(config))); SubscriberBuilder<? extends Message<?>, Void> subscriber = sink.sink(); Multi.createFrom().range(0, 10) .map(i -> Integer.toString(i)) .map(Message::of) .subscribe((Subscriber<Message<?>>) subscriber.build()); await().untilAtomic(expected1, is(10)); assertThat(expected1).hasValue(10); assertThat(expected2).hasValue(10); }
@Test @SuppressWarnings("unchecked") public void testSendAndMultipleConsumers() { String topic = UUID.randomUUID().toString(); AtomicInteger expected1 = new AtomicInteger(0); usage.consumeStrings(topic, 5, 10, TimeUnit.SECONDS, v -> expected1.getAndIncrement()); AtomicInteger expected2 = new AtomicInteger(0); usage.consumeStrings(topic, 5, 10, TimeUnit.SECONDS, v -> expected2.getAndIncrement()); Map<String, Object> config = new HashMap<>(); config.put("address", topic); config.put("publish", false); EventBusSink sink = new EventBusSink(vertx, new VertxEventBusConnectorOutgoingConfiguration(new MapBasedConfig(config))); SubscriberBuilder<? extends Message<?>, Void> subscriber = sink.sink(); Multi.createFrom().range(0, 10) .map(i -> Integer.toString(i)) .map(Message::of) .subscribe((Subscriber<Message<?>>) subscriber.build()); await().untilAtomic(expected1, is(5)); await().untilAtomic(expected2, is(5)); assertThat(expected1).hasValueBetween(4, 6); assertThat(expected2).hasValueBetween(4, 6); }
@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"); }
@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); }