MqttIO { public static Write write() { return new AutoValue_MqttIO_Write.Builder().setRetained(false).build(); } private  MqttIO(); static Read read(); static Write write();  }
@Test public void testWrite() throws Exception { final int numberOfTestMessages = 200; MQTT client = new MQTT(); client.setHost("tcp: final BlockingConnection connection = client.blockingConnection(); connection.connect(); connection.subscribe(new Topic[] {new Topic(Buffer.utf8("WRITE_TOPIC"), QoS.EXACTLY_ONCE)}); final Set<String> messages = new ConcurrentSkipListSet<>(); Thread subscriber = new Thread( () -> { try { for (int i = 0; i < numberOfTestMessages; i++) { Message message = connection.receive(); messages.add(new String(message.getPayload(), StandardCharsets.UTF_8)); message.ack(); } } catch (Exception e) { LOG.error("Can't receive message", e); } }); subscriber.start(); ArrayList<byte[]> data = new ArrayList<>(); for (int i = 0; i < numberOfTestMessages; i++) { data.add(("Test " + i).getBytes(StandardCharsets.UTF_8)); } pipeline .apply(Create.of(data)) .apply( MqttIO.write() .withConnectionConfiguration( MqttIO.ConnectionConfiguration.create( "tcp: pipeline.run(); subscriber.join(); connection.disconnect(); assertEquals(numberOfTestMessages, messages.size()); for (int i = 0; i < numberOfTestMessages; i++) { assertTrue(messages.contains("Test " + i)); } }