MqttSource { PublisherBuilder<MqttMessage<?>> getSource() { return source; } MqttSource(Vertx vertx, MqttConnectorIncomingConfiguration config);   }
@Test public void testSource() { String topic = UUID.randomUUID().toString(); Map<String, Object> config = new HashMap<>(); config.put("topic", topic); config.put("host", address); config.put("port", port); config.put("channel-name", topic); MqttSource source = new MqttSource(vertx, new MqttConnectorIncomingConfiguration(new MapBasedConfig(config))); List<MqttMessage<?>> messages = new ArrayList<>(); PublisherBuilder<MqttMessage<?>> stream = source.getSource(); stream.forEach(messages::add).run(); await().until(source::isSubscribed); AtomicInteger counter = new AtomicInteger(); new Thread(() -> usage.produceIntegers(topic, 10, null, counter::getAndIncrement)).start(); await().atMost(2, TimeUnit.MINUTES).until(() -> messages.size() >= 10); assertThat(messages.stream() .map(Message::getPayload) .map(x -> (byte[]) x) .map(bytes -> Integer.valueOf(new String(bytes))) .collect(Collectors.toList())) .containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9); }
@Test public void testSourceUsingChannelName() { String topic = UUID.randomUUID().toString(); Map<String, Object> config = new HashMap<>(); config.put("channel-name", topic); config.put("host", address); config.put("port", port); MqttSource source = new MqttSource(vertx, new MqttConnectorIncomingConfiguration(new MapBasedConfig(config))); List<MqttMessage<?>> messages = new ArrayList<>(); PublisherBuilder<MqttMessage<?>> stream = source.getSource(); stream.forEach(messages::add).run(); await().until(source::isSubscribed); AtomicInteger counter = new AtomicInteger(); new Thread(() -> usage.produceIntegers(topic, 10, null, counter::getAndIncrement)).start(); await().atMost(2, TimeUnit.MINUTES).until(() -> messages.size() >= 10); assertThat(messages.stream() .map(Message::getPayload) .map(x -> (byte[]) x) .map(bytes -> Integer.valueOf(new String(bytes))) .collect(Collectors.toList())) .containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9); }
@Test public void testBroadcast() { String topic = UUID.randomUUID().toString(); Map<String, Object> config = new HashMap<>(); config.put("topic", topic); config.put("host", address); config.put("port", port); config.put("channel-name", topic); config.put("broadcast", true); MqttSource source = new MqttSource(vertx, new MqttConnectorIncomingConfiguration(new MapBasedConfig(config))); List<MqttMessage<?>> messages1 = new ArrayList<>(); List<MqttMessage<?>> messages2 = new ArrayList<>(); PublisherBuilder<MqttMessage<?>> stream = source.getSource(); stream.forEach(messages1::add).run(); stream.forEach(messages2::add).run(); await().until(source::isSubscribed); AtomicInteger counter = new AtomicInteger(); new Thread(() -> usage.produceIntegers(topic, 10, null, counter::getAndIncrement)).start(); await().atMost(2, TimeUnit.MINUTES).until(() -> messages1.size() >= 10); await().atMost(2, TimeUnit.MINUTES).until(() -> messages2.size() >= 10); assertThat(messages1.stream() .map(Message::getPayload) .map(x -> (byte[]) x) .map(bytes -> Integer.valueOf(new String(bytes))) .collect(Collectors.toList())) .containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9); assertThat(messages2.stream() .map(Message::getPayload) .map(x -> (byte[]) x) .map(bytes -> Integer.valueOf(new String(bytes))) .collect(Collectors.toList())) .containsExactly(0, 1, 2, 3, 4, 5, 6, 7, 8, 9); }
@Test public void testWithVeryLargeMessage() { Random random = new Random(); String topic = UUID.randomUUID().toString(); Map<String, Object> config = new HashMap<>(); config.put("topic", topic); config.put("host", address); config.put("port", port); config.put("channel-name", topic); config.put("max-message-size", 20 * 1024); MqttSource source = new MqttSource(vertx, new MqttConnectorIncomingConfiguration(new MapBasedConfig(config))); byte[] large = new byte[10 * 1024]; random.nextBytes(large); List<MqttMessage<?>> messages = new ArrayList<>(); PublisherBuilder<MqttMessage<?>> stream = source.getSource(); stream.forEach(messages::add).run(); await().until(source::isSubscribed); new Thread(() -> usage.produce(topic, 10, null, () -> large)).start(); await().atMost(2, TimeUnit.MINUTES).until(() -> messages.size() >= 10); assertThat(messages.stream() .map(Message::getPayload) .map(x -> (byte[]) x) .collect(Collectors.toList())) .contains(large); }