Skip to content
This repository has been archived by the owner on Nov 30, 2019. It is now read-only.

Commit

Permalink
[Minor] Fix import order and other coding style
Browse files Browse the repository at this point in the history
fixed import order and other coding style

Author: bilna <bilnap@am.amrita.edu>
Author: Bilna P <bilna.p@gmail.com>

Closes apache#3966 from Bilna/master and squashes the following commits:

5e76f04 [bilna] fix import order and other coding style
5718d66 [bilna] Merge remote-tracking branch 'upstream/master'
ae56514 [bilna] Merge remote-tracking branch 'upstream/master'
acea3a3 [bilna] Adding dependency with scope test
28681fa [bilna] Merge remote-tracking branch 'upstream/master'
fac3904 [bilna] Correction in Indentation and coding style
ed9db4c [bilna] Merge remote-tracking branch 'upstream/master'
4b34ee7 [Bilna P] Update MQTTStreamSuite.scala
04503cf [bilna] Added embedded broker service for mqtt test
89d804e [bilna] Merge remote-tracking branch 'upstream/master'
fc8eb28 [bilna] Merge remote-tracking branch 'upstream/master'
4b58094 [Bilna P] Update MQTTStreamSuite.scala
b1ac4ad [bilna] Added BeforeAndAfter
5f6bfd2 [bilna] Added BeforeAndAfter
e8b6623 [Bilna P] Update MQTTStreamSuite.scala
5ca6691 [Bilna P] Update MQTTStreamSuite.scala
8616495 [bilna] [SPARK-4631] unit test for MQTT
  • Loading branch information
bilna authored and Andrew Or committed Jan 9, 2015
1 parent ae62872 commit 4e1f12d
Showing 1 changed file with 11 additions and 6 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -19,16 +19,19 @@ package org.apache.spark.streaming.mqtt

import java.net.{URI, ServerSocket}

import scala.concurrent.duration._

import org.apache.activemq.broker.{TransportConnector, BrokerService}
import org.apache.spark.util.Utils
import org.eclipse.paho.client.mqttv3._
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence

import org.scalatest.{BeforeAndAfter, FunSuite}
import org.scalatest.concurrent.Eventually
import scala.concurrent.duration._

import org.apache.spark.streaming.{Milliseconds, StreamingContext}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.eclipse.paho.client.mqttv3._
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence
import org.apache.spark.util.Utils

class MQTTStreamSuite extends FunSuite with Eventually with BeforeAndAfter {

Expand All @@ -38,8 +41,9 @@ class MQTTStreamSuite extends FunSuite with Eventually with BeforeAndAfter {
private val freePort = findFreePort()
private val brokerUri = "//localhost:" + freePort
private val topic = "def"
private var ssc: StreamingContext = _
private val persistenceDir = Utils.createTempDir()

private var ssc: StreamingContext = _
private var broker: BrokerService = _
private var connector: TransportConnector = _

Expand Down Expand Up @@ -115,8 +119,9 @@ class MQTTStreamSuite extends FunSuite with Eventually with BeforeAndAfter {
val message: MqttMessage = new MqttMessage(data.getBytes("utf-8"))
message.setQos(1)
message.setRetained(true)
for (i <- 0 to 100)
for (i <- 0 to 100) {
msgTopic.publish(message)
}
}
} finally {
client.disconnect()
Expand Down

0 comments on commit 4e1f12d

Please sign in to comment.