MqttIO { public static Read read() { return new AutoValue_MqttIO_Read.Builder() .setMaxReadTime(null) .setMaxNumRecords(Long.MAX_VALUE) .build(); } private  MqttIO(); static Read read(); static Write write();  }
@Test(timeout = 60 * 1000) @Ignore("https: public void testReadNoClientId() throws Exception { final String topicName = "READ_TOPIC_NO_CLIENT_ID"; Read mqttReader = MqttIO.read() .withConnectionConfiguration( MqttIO.ConnectionConfiguration.create("tcp: .withMaxNumRecords(10); PCollection<byte[]> output = pipeline.apply(mqttReader); PAssert.that(output) .containsInAnyOrder( "This is test 0".getBytes(StandardCharsets.UTF_8), "This is test 1".getBytes(StandardCharsets.UTF_8), "This is test 2".getBytes(StandardCharsets.UTF_8), "This is test 3".getBytes(StandardCharsets.UTF_8), "This is test 4".getBytes(StandardCharsets.UTF_8), "This is test 5".getBytes(StandardCharsets.UTF_8), "This is test 6".getBytes(StandardCharsets.UTF_8), "This is test 7".getBytes(StandardCharsets.UTF_8), "This is test 8".getBytes(StandardCharsets.UTF_8), "This is test 9".getBytes(StandardCharsets.UTF_8)); MQTT client = new MQTT(); client.setHost("tcp: final BlockingConnection publishConnection = client.blockingConnection(); publishConnection.connect(); Thread publisherThread = new Thread( () -> { try { LOG.info( "Waiting pipeline connected to the MQTT broker before sending " + "messages ..."); boolean pipelineConnected = false; while (!pipelineConnected) { Thread.sleep(1000); for (Connection connection : brokerService.getBroker().getClients()) { if (!connection.getConnectionId().isEmpty()) { pipelineConnected = true; } } } for (int i = 0; i < 10; i++) { publishConnection.publish( topicName, ("This is test " + i).getBytes(StandardCharsets.UTF_8), QoS.EXACTLY_ONCE, false); } } catch (Exception e) { } }); publisherThread.start(); pipeline.run(); publishConnection.disconnect(); publisherThread.join(); }
@Test(timeout = 30 * 1000) @Ignore("https: public void testRead() throws Exception { PCollection<byte[]> output = pipeline.apply( MqttIO.read() .withConnectionConfiguration( MqttIO.ConnectionConfiguration.create("tcp: .withClientId("READ_PIPELINE")) .withMaxReadTime(Duration.standardSeconds(3))); PAssert.that(output) .containsInAnyOrder( "This is test 0".getBytes(StandardCharsets.UTF_8), "This is test 1".getBytes(StandardCharsets.UTF_8), "This is test 2".getBytes(StandardCharsets.UTF_8), "This is test 3".getBytes(StandardCharsets.UTF_8), "This is test 4".getBytes(StandardCharsets.UTF_8), "This is test 5".getBytes(StandardCharsets.UTF_8), "This is test 6".getBytes(StandardCharsets.UTF_8), "This is test 7".getBytes(StandardCharsets.UTF_8), "This is test 8".getBytes(StandardCharsets.UTF_8), "This is test 9".getBytes(StandardCharsets.UTF_8)); MQTT client = new MQTT(); client.setHost("tcp: final BlockingConnection publishConnection = client.blockingConnection(); publishConnection.connect(); Thread publisherThread = new Thread( () -> { try { LOG.info( "Waiting pipeline connected to the MQTT broker before sending " + "messages ..."); boolean pipelineConnected = false; while (!pipelineConnected) { Thread.sleep(1000); for (Connection connection : brokerService.getBroker().getClients()) { if (connection.getConnectionId().startsWith("READ_PIPELINE")) { pipelineConnected = true; } } } for (int i = 0; i < 10; i++) { publishConnection.publish( "READ_TOPIC", ("This is test " + i).getBytes(StandardCharsets.UTF_8), QoS.EXACTLY_ONCE, false); } } catch (Exception e) { } }); publisherThread.start(); pipeline.run(); publisherThread.join(); publishConnection.disconnect(); }
@Test(timeout = 30 * 1000) public void testReceiveWithTimeoutAndNoData() throws Exception { pipeline.apply( MqttIO.read() .withConnectionConfiguration( MqttIO.ConnectionConfiguration.create("tcp: .withClientId("READ_PIPELINE")) .withMaxReadTime(Duration.standardSeconds(2))); pipeline.run(); }