@Test(timeout = 4000L) public void testMqttSink() throws Exception { final int numSinks = 4; final List<String> outputStream = Arrays.asList( "Lorem ipsum dolor sit amet, consectetur adipiscing elit.", "In in leo nec erat fringilla mattis eu non massa.", "Cras quis diam suscipit, commodo enim id, pulvinar nunc."); final CountDownLatch countDownLatch = new CountDownLatch(numSinks * outputStream.size()); final Map<String, List<String>> topicListMap = new ConcurrentHashMap<>(); final String topic = "topic"; subscriber.connect(); subscriber.setCallback(new TestMqttSubscriber(topicListMap, countDownLatch)); final List<Sink<MqttMessage>> sinks = new LinkedList<>(); for (int i = 0; i < numSinks; i++) { final Sink<MqttMessage> publisher = new MqttSink(MqttUtils.BROKER_URI, topic + i, mqttSharedResource); topicListMap.put(topic + i, new LinkedList<>()); sinks.add(publisher); subscriber.subscribe(topic + i); } outputStream.forEach((output) -> { for (final Sink<MqttMessage> sink : sinks) { sink.handle(new MqttMessage(output.getBytes())); } }); countDownLatch.await(); for (final List<String> received : topicListMap.values()) { Assert.assertEquals(outputStream, received); } for (final Sink<MqttMessage> sink : sinks) { sink.close(); } }
@Inject public MqttSink( @Parameter(MQTTBrokerURI.class) final String brokerURI, @Parameter(MQTTTopic.class) final String topic, final MQTTResource sharedResource) throws IOException, MqttException { this.brokerURI = brokerURI; this.topic = topic; this.mqttClient = sharedResource.getMqttSinkClient(brokerURI, topic); this.resource = sharedResource; }
MqttSink implements Sink<MqttMessage> { @Inject public MqttSink( @Parameter(MQTTBrokerURI.class) final String brokerURI, @Parameter(MQTTTopic.class) final String topic, final MQTTResource sharedResource) throws IOException, MqttException { this.brokerURI = brokerURI; this.topic = topic; this.mqttClient = sharedResource.getMqttSinkClient(brokerURI, topic); this.resource = sharedResource; } }
MqttSink implements Sink<MqttMessage> { @Inject public MqttSink( @Parameter(MQTTBrokerURI.class) final String brokerURI, @Parameter(MQTTTopic.class) final String topic, final MQTTResource sharedResource) throws IOException, MqttException { this.brokerURI = brokerURI; this.topic = topic; this.mqttClient = sharedResource.getMqttSinkClient(brokerURI, topic); this.resource = sharedResource; } @Inject  MqttSink(
      @Parameter(MQTTBrokerURI.class) final String brokerURI,
      @Parameter(MQTTTopic.class) final String topic,
      final MQTTResource sharedResource); }
MqttSink implements Sink<MqttMessage> { @Inject public MqttSink( @Parameter(MQTTBrokerURI.class) final String brokerURI, @Parameter(MQTTTopic.class) final String topic, final MQTTResource sharedResource) throws IOException, MqttException { this.brokerURI = brokerURI; this.topic = topic; this.mqttClient = sharedResource.getMqttSinkClient(brokerURI, topic); this.resource = sharedResource; } @Inject  MqttSink(
      @Parameter(MQTTBrokerURI.class) final String brokerURI,
      @Parameter(MQTTTopic.class) final String topic,
      final MQTTResource sharedResource); @Override void close(); @Override void handle(final MqttMessage input); }
MqttSink implements Sink<MqttMessage> { @Inject public MqttSink( @Parameter(MQTTBrokerURI.class) final String brokerURI, @Parameter(MQTTTopic.class) final String topic, final MQTTResource sharedResource) throws IOException, MqttException { this.brokerURI = brokerURI; this.topic = topic; this.mqttClient = sharedResource.getMqttSinkClient(brokerURI, topic); this.resource = sharedResource; } @Inject  MqttSink(
      @Parameter(MQTTBrokerURI.class) final String brokerURI,
      @Parameter(MQTTTopic.class) final String topic,
      final MQTTResource sharedResource); @Override void close(); @Override void handle(final MqttMessage input);  }