//设置package包名称以及导入依赖的类
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence
import org.eclipse.paho.client.mqttv3.{MqttClient, MqttMessage}

import scala.util.{Failure, Random, Success, Try}


object Publisher extends App {
  // Global information
  val name = "apdl-case-study-from-scratch"

  // MQTT Server info to publish
  val brokerURL = "tcp://mosquitto-mqtt:1883"
  val topicTemperature = s"temp"
  val topicLuminosity = s"light"
  val persistence = new MqttDefaultFilePersistence("tmp")

  Try {
    val client = new MqttClient(brokerURL, MqttClient.generateClientId(), persistence)
    client.connect()

    val topicTemp = client.getTopic(topicTemperature)
    val topicLight = client.getTopic(topicLuminosity)

    while (true) {

      val temp = Random.nextInt(10) + 20
      val light = Random.nextInt(40) + 640
      val msgTemp = new MqttMessage(s"$temp".getBytes)
      val msgLight = new MqttMessage(s"$light".getBytes)

      println(s"Publish $temp to $topicTemperature")
      println(s"Publish $light to $topicLight")

      topicTemp.publish(msgTemp)
      topicLight.publish(msgLight)
      Thread.sleep(1000)
    }
    while(true) {
      println("OK !")
      Thread.sleep(1000)
    }

  } match {
    case Failure(exception) => println(s"ERROR : $exception + ${exception.getCause}")
    case Success(_) => println(s"OK !")
  }
} 

//设置package包名称以及导入依赖的类
package com.murraywilliams

import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
import org.eclipse.paho.client.mqttv3.MqttClient
import org.eclipse.paho.client.mqttv3.MqttConnectOptions
import org.eclipse.paho.client.mqttv3.MqttMessage

object MQTTClient {
  
  private[this] val logger = org.log4s.getLogger
  
  lazy val mqttClient = {
    import XBeeReceiver.conf
    
    try {
      val memoryPersistence = new MemoryPersistence()
      val brokerUrl = conf.getString("mqtt.url")
      val client = new MqttClient(brokerUrl, "xbeeSensorsClientId", memoryPersistence)
      val opts = new MqttConnectOptions()
      opts.setCleanSession(true)
      logger.info(s"Connecting to broker ${brokerUrl}...")
      client.connect(opts)
      logger.info("Connected")
      Some(client)
    } catch {
      case t : Throwable => {
        logger.error(t)("Could not connect to MQTT server.")
        None
      }
    }
  }
  
  lazy val mqttTopicRoot = XBeeReceiver.conf.getString("mqtt.topic")
  
  def mqttSend(topicSuffix : String, message : Any) = {
    val msg = message match {
      case str : String => str.getBytes
      case byteArray : Array[Byte] => byteArray
      case _ => message.toString.getBytes
    }
    mqttClient match {
      case Some(client) if client.isConnected => client.publish(s"${mqttTopicRoot}/${topicSuffix}",new MqttMessage(msg))
      case None => logger.warn("No connection to MQTT broker. Not sending any data.")
    }      
  }
  
} 

//设置package包名称以及导入依赖的类
package com.knoldus

import com.typesafe.config.ConfigFactory
import org.eclipse.paho.client.mqttv3.{MqttClient, MqttException, MqttMessage}
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence

import scala.sys.process._



object MQTTPublisher extends App {

  val config = ConfigFactory.load()
  val url = config.getString("mosquitto-server.url")
  val port = config.getInt("mosquitto-server.port")

  def publishToserver() = {
    println("Hey I am publishing")
    val brokerUrl = s"tcp://$url:$port"
    val topic = "TemperatureEvent"
    val tempCommand = "/opt/vc/bin/vcgencmd measure_temp"
    def getMessage = s"Temperature of CPU at ${System.currentTimeMillis()} is ${tempCommand.!!.split("=")(1)} "
    var client: MqttClient = null
    val persistence = new MemoryPersistence
    try {
      client = new MqttClient(brokerUrl, MqttClient.generateClientId, persistence)
      client.connect()
      val msgTopic = client.getTopic(topic)
      val message = new MqttMessage(getMessage.getBytes("utf-8"))
      while (true) {
        msgTopic.publish(message)
        println(s"Publishing the data topic ${msgTopic.getName} message: ${message.toString}")
        Thread.sleep(1000)
      }
    }
    catch {
      case exception: MqttException => println(s"ExceptionOccured:$exception ")
    }
    finally {
      client.disconnect()
    }
  }

  publishToserver
} 

//设置package包名称以及导入依赖的类
package mqtt

import akka.actor.{ActorPath, ActorSystem}
import mqtt.MqttListenerMessage.ConsumeMessage
import org.eclipse.paho.client.mqttv3.{IMqttDeliveryToken, MqttCallback, MqttMessage}
import play.api.Logger


class MqttDispatchingListener(actorSystem: ActorSystem) extends MqttCallback {
  var listeners: List[ActorPath] = List.empty

  override def deliveryComplete(token: IMqttDeliveryToken): Unit = {}
  override def connectionLost(cause: Throwable): Unit = {}

  def addListener(listener:ActorPath): Unit = {
    listeners = listener :: listeners
  }

  override def messageArrived(topic: String, message: MqttMessage): Unit = {
    val msgStr = new String(message.getPayload)
    Logger.debug(s"Mqtt topic $topic received $msgStr")

    listeners.foreach(actorSystem.actorSelection(_) ! ConsumeMessage(topic, msgStr))
  }
}

object MqttListenerMessage {
  case class Ping()
  case class ConsumeMessage(receivedTopic: String, message: String)
} 

//设置package包名称以及导入依赖的类
package org.apress.prospark

import java.nio.charset.StandardCharsets

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.eclipse.paho.client.mqttv3.MqttClient
import org.eclipse.paho.client.mqttv3.MqttMessage
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
import org.json4s.DefaultFormats
import org.json4s.JField
import org.json4s.JsonAST.JObject
import org.json4s.jvalue2extractable
import org.json4s.jvalue2monadic
import org.json4s.native.JsonMethods.parse
import org.json4s.string2JsonInput

object MqttSinkAppA {

  def main(args: Array[String]) {
    if (args.length != 3) {
      System.err.println(
        "Usage: MqttSinkApp <appname> <outputBrokerUrl> <topic>")
      System.exit(1)
    }

    val Seq(appName, outputBrokerUrl, topic) = args.toSeq

    val conf = new SparkConf()
      .setAppName(appName)
      .setJars(SparkContext.jarOfClass(this.getClass).toSeq)

    val batchInterval = 10

    val ssc = new StreamingContext(conf, Seconds(batchInterval))

    HttpUtils.createStream(ssc, url = "https://query.yahooapis.com/v1/public/yql?q=select%20*%20from%20yahoo.finance.quotes%20where%20symbol%20in%20(%22IBM,GOOG,MSFT,AAPL,FB,ORCL,YHOO,TWTR,LNKD,INTC%22)%0A%09%09&format=json&diagnostics=true&env=http%3A%2F%2Fdatatables.org%2Falltables.env",
      interval = batchInterval)
      .flatMap(rec => {
        val query = parse(rec) \ "query"
        ((query \ "results" \ "quote").children).map(rec => JObject(JField("Timestamp", query \ "created")).merge(rec))
      })
      .map(rec => {
        implicit val formats = DefaultFormats
        rec.children.map(f => f.extract[String]) mkString ","
      })
      .foreachRDD { rdd =>
        val client = new MqttClient(outputBrokerUrl, MqttClient.generateClientId(), new MemoryPersistence())
        client.connect()
        rdd.foreach(rec => client.publish(topic, new MqttMessage(rec.getBytes(StandardCharsets.UTF_8))))
        client.disconnect()
        client.close()
      }

    ssc.start()
    ssc.awaitTermination()
  }

} 

//设置package包名称以及导入依赖的类
package Birdcage


import java.util.Date
import com.sksamuel.elastic4s.ElasticDsl._
import akka.actor.Actor
import com.sksamuel.elastic4s.{ElasticClient, IndexAndType, IndexAndTypes, IndexResult}
import com.typesafe.scalalogging.LazyLogging
import org.eclipse.paho.client.mqttv3.MqttMessage
import scala.concurrent.ExecutionContext.Implicits.global
import concurrent.Future
import scala.util.{Failure, Success}

object MsgToElastic {
  case class MsgMqtt(topic: String, message: MqttMessage, esClient: ElasticClient, indexTypeEs: Tuple2[String, String])
}

class MsgToElastic extends Actor with LazyLogging {

  import MsgToElastic._

  def receive = {
    case MsgMqtt(topic, message, esClient, indexTypeEs) => sentToElastic(topic, message, esClient, indexTypeEs)
    case _ => logger.debug("MsgActor received unsupported message")
  }

  def sentToElastic(topic: String, message: MqttMessage, esClient: ElasticClient, indexTypeEs: Tuple2[String, String]) = {

    val indexType = new IndexAndType(indexTypeEs._1.toLowerCase, indexTypeEs._2.toLowerCase())
    val messageVal = s"%s".format(message)
    val timestamp = new Date().toString

    val resp: Future[IndexResult] = {
      esClient.execute {
        index into IndexAndTypes.apply(indexType)  fields(
          "postcode"   -> topic.takeWhile(_ != '/').toLowerCase(),
          "huisnummer" -> topic.dropWhile(_ != '/').drop(1).takeWhile(_ != '/'),
          "timestamp"  -> timestamp,
          "subject"    -> topic.reverse.takeWhile(_ != '/').reverse,
          "value"      -> messageVal
          )
      }
    }
    resp.onComplete
    {
      case Success(res) => {
        val bufferString = "Msg to Es : %s, %s, %s\n %s\n".format(timestamp, topic, messageVal, resp)
        logger.info(bufferString)
      }
      case Failure(e) => logger.info("An error has occured while sending to Elastic")
    }
  }
} 

//设置package包名称以及导入依赖的类
package io.hydrosphere.mist.master

import io.hydrosphere.mist.Messages.StatusMessages.SystemEvent
import io.hydrosphere.mist.master.interfaces.JsonCodecs
import io.hydrosphere.mist.master.interfaces.async.kafka.TopicProducer
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
import org.eclipse.paho.client.mqttv3.{MqttAsyncClient, MqttConnectOptions, MqttMessage}

trait JobEventPublisher {

  def notify(event: SystemEvent): Unit

  def close(): Unit

}

object JobEventPublisher {

  import JsonCodecs._
  import spray.json._

  def forKafka(host: String, port: Int, topic: String): JobEventPublisher = {
    new JobEventPublisher {
      val producer = TopicProducer(host, port, topic)

      override def notify(event: SystemEvent): Unit = {
        val json = event.toJson
        producer.send("", json)
      }

      override def close(): Unit = {}
    }
  }

  def forMqtt(host: String, port: Int, topic: String): JobEventPublisher = {
    new JobEventPublisher {

      val client = {
        val persistence = new MemoryPersistence
        val uri = s"tcp://$host:$port"
        val client = new MqttAsyncClient(uri, MqttAsyncClient.generateClientId(), persistence)

        val opt = new MqttConnectOptions
        opt.setCleanSession(true)
        client.connect(opt)

        client
      }

      override def notify(event: SystemEvent): Unit = {
        val json = event.toJson
        val message = new MqttMessage(json.getBytes())
        client.publish(topic, message)
      }

      override def close(): Unit = client.close()
    }
  }

} 

//设置package包名称以及导入依赖的类
import com.typesafe.config.ConfigFactory
import org.eclipse.paho.client.mqttv3.{MqttClient, MqttException, MqttMessage}
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence

import scala.sys.process._
import scala.util.Try



object MQTTPublisher extends App {

  val url = Try(ConfigFactory.load().getString("mosquitto-server.url")).toOption.fold("localhost")
  val port = Try(ConfigFactory.load().getInt("mosquitto-server.port")).toOption.fold(1883)

  def publishToserver() = {
    println("Hey I am publishing")
    val brokerUrl = s"tcp://$url:$port"
    val topic = "TemperatureEvent"
    val tempCommand = "/opt/vc/bin/vcgencmd measure_temp"
    def getMessage = s"Temperature of CPU at ${System.currentTimeMillis()} is ${tempCommand.!!.split("=")(1)} "
    var client: MqttClient = null
    val persistence = new MemoryPersistence
    try {
      client = new MqttClient(brokerUrl, MqttClient.generateClientId, persistence)
      client.connect()
      val msgTopic = client.getTopic(topic)
      val message = new MqttMessage(getMessage.getBytes("utf-8"))
      while (true) {
        msgTopic.publish(message)
        println(s"Publishing the data topic ${msgTopic.getName} message: ${message.toString}")
        Thread.sleep(1000)
      }
    }
    catch {
      case exception: MqttException => println(s"ExceptionOccured:$exception ")
    }
    finally {
      client.disconnect()
    }
  }

  publishToserver
} 

//设置package包名称以及导入依赖的类
package service

import com.google.inject.ImplementedBy
import org.eclipse.paho.client.mqttv3.{MqttClient, MqttMessage}
import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence

import scala.concurrent.Future
import scala.util.{Failure, Success, Try}


object PublishService {
  val client = new MqttClient(brokerUrl, MqttClient.generateClientId(), persistence)
  def publish(message: String): Boolean = {
    client.connect()
    val result = Try(client.getTopic(topic+"/set")) match {
      case Success(messageTopic) =>
        val mqttMessage = new MqttMessage(message.getBytes("utf-8"))
        Try(messageTopic.publish(mqttMessage)) match {
          case Success(r) =>
            play.Logger.debug(s"Result of publishing: ${r.getMessage}")
            true
          case Failure(exception) =>

            play.Logger.error(
              s"""
                 |An error occurred while trying to publish the message "$message" to $topic into $brokerUrl:
         """.stripMargin, exception)
            false
        }
      case Failure(exception) =>
        play.Logger.error(s"""
                             |An error occurred while trying to publish the message "$message" to $topic into $brokerUrl:
         """.stripMargin, exception)
        false
    }
    client.disconnect()
    result
  }
} 

//设置package包名称以及导入依赖的类
package service

import controllers.socket.LightSocketActor
import model.Message
import org.eclipse.paho.client.mqttv3.{IMqttDeliveryToken, MqttCallback, MqttClient, MqttMessage}


object SubscribeService {
  val client = new MqttClient(brokerUrl, MqttClient.generateClientId(), persistence)

  def subscribe(): Unit = {
    client.connect()

    // TODO: Change topic
    client.subscribe(topic)

    val callback = new MqttCallback {
      override def deliveryComplete(token: IMqttDeliveryToken): Unit = {
        play.Logger.debug("Delivery complete!")
      }

      override def connectionLost(cause: Throwable): Unit = {
        play.Logger.debug("Connection to socket lost")
        LightSocketActor.sendMessage(Message("Connection lost"))
      }

      override def messageArrived(topic: String, message: MqttMessage): Unit = {
        play.Logger.debug(s"A message arrived: ${new String(message.getPayload)}")
        LightSocketActor.sendMessage(Message(new String(message.getPayload)))
      }
    }

    client.setCallback(callback)
  }
}