MqttSink { public SubscriberBuilder<? extends Message<?>, Void> getSink() { return sink; } MqttSink(Vertx vertx, MqttConnectorOutgoingConfiguration config); SubscriberBuilder<? extends Message<?>, Void> getSink(); boolean isReady();  }
@SuppressWarnings("unchecked") @Test public void testSinkUsingInteger() throws InterruptedException { String topic = UUID.randomUUID().toString(); CountDownLatch latch = new CountDownLatch(1); AtomicInteger expected = new AtomicInteger(0); usage.consumeIntegers(topic, 10, 10, TimeUnit.SECONDS, latch::countDown, v -> expected.getAndIncrement()); Map<String, Object> config = new HashMap<>(); config.put("topic", topic); config.put("host", address); config.put("port", port); MqttSink sink = new MqttSink(vertx, new MqttConnectorOutgoingConfiguration(new MapBasedConfig(config))); Subscriber<? extends Message<?>> subscriber = sink.getSink().build(); Flowable.range(0, 10) .map(Message::of) .subscribe((Subscriber<? super Message<Integer>>) subscriber); assertThat(latch.await(1, TimeUnit.MINUTES)).isTrue(); await().untilAtomic(expected, is(10)); assertThat(expected).hasValue(10); }
@SuppressWarnings("unchecked") @Test public void testSinkUsingChannelName() throws InterruptedException { String topic = UUID.randomUUID().toString(); CountDownLatch latch = new CountDownLatch(1); AtomicInteger expected = new AtomicInteger(0); usage.consumeIntegers(topic, 10, 10, TimeUnit.SECONDS, latch::countDown, v -> expected.getAndIncrement()); Map<String, Object> config = new HashMap<>(); config.put("channel-name", topic); config.put("host", address); config.put("port", port); MqttSink sink = new MqttSink(vertx, new MqttConnectorOutgoingConfiguration(new MapBasedConfig(config))); Subscriber<? extends Message<?>> subscriber = sink.getSink().build(); Flowable.range(0, 10) .map(Message::of) .subscribe((Subscriber<? super Message<Integer>>) subscriber); assertThat(latch.await(1, TimeUnit.MINUTES)).isTrue(); await().untilAtomic(expected, is(10)); assertThat(expected).hasValue(10); }
@SuppressWarnings("unchecked") @Test public void testSinkUsingString() throws InterruptedException { String topic = UUID.randomUUID().toString(); CountDownLatch latch = new CountDownLatch(1); AtomicInteger expected = new AtomicInteger(0); usage.consumeStrings(topic, 10, 10, TimeUnit.SECONDS, latch::countDown, v -> expected.getAndIncrement()); Map<String, Object> config = new HashMap<>(); config.put("topic", topic); config.put("host", address); config.put("port", port); MqttSink sink = new MqttSink(vertx, new MqttConnectorOutgoingConfiguration(new MapBasedConfig(config))); Subscriber<? extends Message<?>> subscriber = sink.getSink().build(); Flowable.range(0, 10) .map(i -> Integer.toString(i)) .map(Message::of) .subscribe((Subscriber<? super Message<String>>) subscriber); assertThat(latch.await(1, TimeUnit.MINUTES)).isTrue(); await().untilAtomic(expected, is(10)); assertThat(expected).hasValue(10); }