You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ma...@apache.org on 2013/10/24 00:15:41 UTC
[11/14] git commit: Update MQTTWordCount.scala
Update MQTTWordCount.scala
Project: http://git-wip-us.apache.org/repos/asf/incubator-spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-spark/commit/6ec39829
Tree: http://git-wip-us.apache.org/repos/asf/incubator-spark/tree/6ec39829
Diff: http://git-wip-us.apache.org/repos/asf/incubator-spark/diff/6ec39829
Branch: refs/heads/master
Commit: 6ec39829e9204c742e364d48c23e106625bba17d
Parents: d223d38
Author: Prabeesh K <pr...@gmail.com>
Authored: Fri Oct 18 17:00:28 2013 +0530
Committer: Prabeesh K <pr...@gmail.com>
Committed: Fri Oct 18 17:00:28 2013 +0530
----------------------------------------------------------------------
.../streaming/examples/MQTTWordCount.scala | 29 ++++++++++----------
1 file changed, 14 insertions(+), 15 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-spark/blob/6ec39829/examples/src/main/scala/org/apache/spark/streaming/examples/MQTTWordCount.scala
----------------------------------------------------------------------
diff --git a/examples/src/main/scala/org/apache/spark/streaming/examples/MQTTWordCount.scala b/examples/src/main/scala/org/apache/spark/streaming/examples/MQTTWordCount.scala
index 04e21be..be6587b 100644
--- a/examples/src/main/scala/org/apache/spark/streaming/examples/MQTTWordCount.scala
+++ b/examples/src/main/scala/org/apache/spark/streaming/examples/MQTTWordCount.scala
@@ -22,12 +22,12 @@ import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.dstream.MQTTReceiver
import org.apache.spark.storage.StorageLevel
-import org.eclipse.paho.client.mqttv3.MqttClient;
-import org.eclipse.paho.client.mqttv3.MqttClientPersistence;
-import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence;
-import org.eclipse.paho.client.mqttv3.MqttException;
-import org.eclipse.paho.client.mqttv3.MqttMessage;
-import org.eclipse.paho.client.mqttv3.MqttTopic;
+import org.eclipse.paho.client.mqttv3.MqttClient
+import org.eclipse.paho.client.mqttv3.MqttClientPersistence
+import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence
+import org.eclipse.paho.client.mqttv3.MqttException
+import org.eclipse.paho.client.mqttv3.MqttMessage
+import org.eclipse.paho.client.mqttv3.MqttTopic
/**
* A simple Mqtt publisher for demonstration purposes, repeatedly publishes Space separated String Message "hello mqtt demo for spark streaming"
@@ -47,24 +47,24 @@ object MQTTPublisher {
val Seq(brokerUrl, topic) = args.toSeq
try {
- var peristance:MqttClientPersistence =new MqttDefaultFilePersistence("/tmp");
- client = new MqttClient(brokerUrl, MqttClient.generateClientId(), peristance);
+ var peristance:MqttClientPersistence =new MqttDefaultFilePersistence("/tmp")
+ client = new MqttClient(brokerUrl, MqttClient.generateClientId(), peristance)
} catch {
- case e: MqttException => println("Exception Caught: " + e);
+ case e: MqttException => println("Exception Caught: " + e)
}
- client.connect();
+ client.connect()
val msgtopic: MqttTopic = client.getTopic(topic);
- val msg: String = "hello mqtt demo for spark streaming";
+ val msg: String = "hello mqtt demo for spark streaming"
while (true) {
- val message: MqttMessage = new MqttMessage(String.valueOf(msg).getBytes());
+ val message: MqttMessage = new MqttMessage(String.valueOf(msg).getBytes())
msgtopic.publish(message);
- println("Published data. topic: " + msgtopic.getName() + " Message: " + message);
+ println("Published data. topic: " + msgtopic.getName() + " Message: " + message)
}
- client.disconnect();
+ client.disconnect()
}
}
@@ -109,4 +109,3 @@ object MQTTWordCount {
ssc.start()
}
}
-