beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [22/50] [abbrv] beam git commit: [BEAM-1686] Use random MQTT clientID when not defined to avoid NPE
Date Fri, 17 Mar 2017 21:46:53 GMT
[BEAM-1686] Use random MQTT clientID when not defined to avoid NPE


Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/752ad8a0
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/752ad8a0
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/752ad8a0

Branch: refs/heads/gearpump-runner
Commit: 752ad8a0ec5597ca84c27cb21862e32d05a81420
Parents: fdba784
Author: Borisa Zivkovic <borisa.zivkovic@huawei.com>
Authored: Fri Mar 10 12:20:13 2017 +0000
Committer: Jean-Baptiste Onofré <jbonofre@apache.org>
Committed: Fri Mar 10 17:00:57 2017 +0100

----------------------------------------------------------------------
 .../org/apache/beam/sdk/io/mqtt/MqttIO.java     |  4 ++
 .../org/apache/beam/sdk/io/mqtt/MqttIOTest.java | 61 ++++++++++++++++++++
 2 files changed, 65 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/beam/blob/752ad8a0/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java b/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java
index 26234cf..46f2dcc 100644
--- a/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java
+++ b/sdks/java/io/mqtt/src/main/java/org/apache/beam/sdk/io/mqtt/MqttIO.java
@@ -209,6 +209,10 @@ public class MqttIO {
         String clientId = getClientId() + "-" + UUID.randomUUID().toString();
         LOG.debug("MQTT client id set to {}", clientId);
         client.setClientId(clientId);
+      } else {
+        String clientId = UUID.randomUUID().toString();
+        LOG.debug("MQTT client id set to random value {}", clientId);
+        client.setClientId(clientId);
       }
       return client;
     }

http://git-wip-us.apache.org/repos/asf/beam/blob/752ad8a0/sdks/java/io/mqtt/src/test/java/org/apache/beam/sdk/io/mqtt/MqttIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/mqtt/src/test/java/org/apache/beam/sdk/io/mqtt/MqttIOTest.java b/sdks/java/io/mqtt/src/test/java/org/apache/beam/sdk/io/mqtt/MqttIOTest.java
index 8a82f40..28ca5f7 100644
--- a/sdks/java/io/mqtt/src/test/java/org/apache/beam/sdk/io/mqtt/MqttIOTest.java
+++ b/sdks/java/io/mqtt/src/test/java/org/apache/beam/sdk/io/mqtt/MqttIOTest.java
@@ -27,6 +27,7 @@ import java.util.Set;
 
 import org.apache.activemq.broker.BrokerService;
 import org.apache.activemq.broker.Connection;
+import org.apache.beam.sdk.io.mqtt.MqttIO.Read;
 import org.apache.beam.sdk.testing.PAssert;
 import org.apache.beam.sdk.testing.RunnableOnService;
 import org.apache.beam.sdk.testing.TestPipeline;
@@ -79,6 +80,66 @@ public class MqttIOTest {
 
   @Test(timeout = 60 * 1000)
   @Category(RunnableOnService.class)
+  public void testReadNoClientId() throws Exception {
+    final String topicName = "READ_TOPIC_NO_CLIENT_ID";
+    Read mqttReader = MqttIO.read()
+    .withConnectionConfiguration(
+        MqttIO.ConnectionConfiguration.create(
+            "tcp://localhost:" + port,
+            topicName))
+  .withMaxNumRecords(10);
+    PCollection<byte[]> output = pipeline.apply(mqttReader);
+    PAssert.that(output).containsInAnyOrder(
+        "This is test 0".getBytes(),
+        "This is test 1".getBytes(),
+        "This is test 2".getBytes(),
+        "This is test 3".getBytes(),
+        "This is test 4".getBytes(),
+        "This is test 5".getBytes(),
+        "This is test 6".getBytes(),
+        "This is test 7".getBytes(),
+        "This is test 8".getBytes(),
+        "This is test 9".getBytes()
+    );
+
+    // produce messages on the brokerService in another thread
+    // This thread prevents to block the pipeline waiting for new messages
+    MQTT client = new MQTT();
+    client.setHost("tcp://localhost:" + port);
+    final BlockingConnection publishConnection = client.blockingConnection();
+    publishConnection.connect();
+    Thread publisherThread = new Thread() {
+      public void run() {
+        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(),
+                QoS.AT_LEAST_ONCE, false);
+          }
+        } catch (Exception e) {
+          // nothing to do
+        }
+      }
+    };
+    publisherThread.start();
+    pipeline.run();
+
+    publishConnection.disconnect();
+    publisherThread.join();
+  }
+
+  @Test(timeout = 60 * 1000)
+  @Category(RunnableOnService.class)
   public void testRead() throws Exception {
     PCollection<byte[]> output = pipeline.apply(
         MqttIO.read()


Mime
View raw message