Merge pull request #33 from SoMind/mqtt-merge

refactor

Author: navicore

.gitignore

@@ -1,3 +1,5 @@
+.bsp/
+runme*sh
 target/
 project/target/
 project/project/

README.md

@@ -2,10 +2,35 @@
 
 [![Codacy Badge](https://api.codacy.com/project/badge/Grade/25295eb70b8d463486773644f4b1c215)](https://app.codacy.com/gh/SoMind/dtlab-ingest-mqtt?utm_source=github.com&utm_medium=referral&utm_content=SoMind/dtlab-ingest-mqtt&utm_campaign=Badge_Grade_Dashboard)
 
-## UNDER CONSTRUCTION
-
 Listen for data on an MQTT topic and post each message to a remote HTTP service (probably a dtlab-ingest instance).
 
 ```console
 sbt assembly && MQTT_URL=ssl://YOUR_HOST:8883 MQTT_TOPIC=test/c/ MQTT_CLIENT_ID=YOUR_CLIENT_ID MQTT_USER=YOUR_USER MQTT_PWD=YOUR_PWD java -jar target/scala-2.12/DtLabMqttIngest.jar
-```
+```
+
+### Build and run
+
+```console
+sbt assembly
+java -jar target/scala-2.12/MqttToKafka.jar
+```
+
+### Test and Code Coverage
+
+```console
+sbt clean coverage test
+sbt coverageReport
+open target/scala-2.12/scoverage-report/index.html
+```
+
+### Code Stats
+
+```console
+sbt stats
+```
+### Check dependencies
+
+```console
+sbt dependencyUpdates
+```
+

build.sbt

@@ -1,40 +1,53 @@
-name := "DtLabIngestMqtt"
+name := "DtlabIngestMqtt"
 
 fork := true
 javaOptions in test ++= Seq(
-  "-Xms128M", "-Xmx256M",
-  "-XX:MaxPermSize=256M",
+  "-Xms512M", "-Xmx2048M",
+  "-XX:MaxPermSize=2048M",
   "-XX:+CMSClassUnloadingEnabled"
 )
 
 parallelExecution in test := false
 
+scalacOptions += "-Ypartial-unification"
+
 version := "1.0"
 
-scalaVersion := "2.12.12"
-val akkaVersion = "2.6.10"
+scalaVersion := "2.12.8"
+val akkaVersion = "2.5.21"
 
 libraryDependencies ++=
   Seq(
-
-    "tech.navicore" %% "navipath" % "4.0.2",
+    "com.typesafe" % "config" % "1.3.3",
     "com.typesafe.scala-logging" %% "scala-logging" % "3.9.2",
-    "com.typesafe" % "config" % "1.4.1",
     "ch.qos.logback" % "logback-classic" % "1.2.3",
+    "org.slf4j" % "jul-to-slf4j" % "1.7.30",
+
+    "org.typelevel" %% "cats-core" % "1.6.0",
     "com.typesafe.akka" %% "akka-actor" % akkaVersion,
     "com.typesafe.akka" %% "akka-stream" % akkaVersion,
+
     "com.lightbend.akka" %% "akka-stream-alpakka-mqtt" % "2.0.2",
-    "org.scalatest" %% "scalatest" % "3.2.2" % "test"
 
+    "org.scalatest" %% "scalatest" % "3.0.5" % "test"
   )
 
+dependencyOverrides ++= Seq(
+  "com.typesafe.akka" %% "akka-actor"  % akkaVersion,
+  "com.typesafe.akka" %% "akka-stream" % akkaVersion
+)
+
 mainClass in assembly := Some("somind.dtlab.ingest.mqtt.Main")
-assemblyJarName in assembly := "DtLabIngestMqtt.jar"
+assemblyJarName in assembly := "DtlabIngestMqtt.jar"
 
 assemblyMergeStrategy in assembly := {
   case PathList("reference.conf") => MergeStrategy.concat
   case x if x.endsWith("io.netty.versions.properties") => MergeStrategy.first
-  case PathList("META-INF", _ @ _*) => MergeStrategy.discard
+  //case PathList("META-INF", _ @ _*) => MergeStrategy.discard
+  case PathList("META-INF", "MANIFEST.MF") => MergeStrategy.discard
+  case PathList("META-INF", "ECLIPSE_.RSA") => MergeStrategy.discard
+  case PathList("META-INF", "ECLIPSE_.SF") => MergeStrategy.discard
+  //case x if x.endsWith("MANIFEST.MF") => MergeStrategy.last
   case _ => MergeStrategy.first
 }
 

src/main/resources/application.conf

@@ -1,20 +1,34 @@
 main {
 
-  appName = "DtLabIngestMqtt"
+  appName = "DtlabIngestMqtt"
+  appName = ${?APP_NAME}
 
-}
+  keyStorePath = "~/secrets/keystore.jks"
+  keyStorePath = ${?KEYSTORE_PATH}
+  keyStorePassword = "mysecret"
+  keyStorePassword = ${?KEYSTORE_PASSWORD}
+
+  mqttTopic = "data"
+  mqttTopic = ${?MQTT_TOPIC}
 
-mqtt {
-  url = ${MQTT_URL}
-  user = ${MQTT_USER}
-  pwd = ${MQTT_PWD}
-  topic = ${MQTT_TOPIC}
-  clientId = ${MQTT_CLIENT_ID}
+  mqttUrl = "ssl://my.mqtt.host:12345"
+  mqttUrl = ${MQTT_URL}
+
+  mqttClientId = "dtLabClient"
+  mqttClientId = ${?MQTT_CLIENT_ID}
 }
 
 akka {
+
   # Options: OFF, ERROR, WARNING, INFO, DEBUG
+  loglevel = "ERROR"
   loglevel = ${?AKKA_LOG_LEVEL}
-  loglevel = "INFO"
-}
 
+  actor {
+    serializers {
+    }
+    serialization-bindings {
+    }
+  }
+
+}

src/main/resources/logback.xml

@@ -6,11 +6,11 @@
         </encoder>
     </appender>
 
-    <root level="${SYS_LOG_LEVEL:-warn}">
+    <root level="${SYS_LOG_LEVEL:-error}">
         <appender-ref ref="STDOUT"/>
     </root>
 
-    <logger name="somind" level="${LOG_LEVEL:-debug}"/>
+    <logger name="somind" level="${LOG_LEVEL:-info}"/>
 
 </configuration>
 

src/main/scala/somind/dtlab/ingest/mqtt/Conf.scala

@@ -1,31 +1,23 @@
 package somind.dtlab.ingest.mqtt
 
 import akka.actor.ActorSystem
-import akka.stream._
+import akka.stream.{ActorMaterializer, ActorMaterializerSettings}
 import com.typesafe.config.{Config, ConfigFactory}
 import com.typesafe.scalalogging.LazyLogging
 
-object Conf extends Conf with LazyLogging {
+object Conf extends LazyLogging {
 
-  implicit val actorSystem: ActorSystem = ActorSystem("MqttKafka")
-
-  val decider: Supervision.Decider = { e: Throwable =>
-    logger.error(s"decider can not decide: $e - restarting...", e)
-    Supervision.Restart
-  }
+  InitJavaLogging()
 
+  implicit val actorSystem: ActorSystem = ActorSystem("MqttToKafkaSystem")
   implicit val materializer: ActorMaterializer = ActorMaterializer(
-    ActorMaterializerSettings(actorSystem).withSupervisionStrategy(decider))
-}
-
-trait Conf {
+    ActorMaterializerSettings(actorSystem))
 
   val conf: Config = ConfigFactory.load()
-
-  val mqttClientId: String = conf.getString("mqtt.clientId")
-  val mqttUrl: String = conf.getString("mqtt.url")
-  val mqttUser: String = conf.getString("mqtt.user")
-  val mqttPwd: String = conf.getString("mqtt.pwd")
-  val mqttTopic: String = conf.getString("mqtt.topic")
+  val mqttUrl: String = conf.getString("main.mqttUrl")
+  val mqttClientId: String = conf.getString("main.mqttClientId")
+  val mqttTopic: String = conf.getString("main.mqttTopic")
+  val keyStorePassword: String = conf.getString("main.keyStorePassword")
+  val keyStorePath: String = conf.getString("main.keyStorePath")
 
 }

src/main/scala/somind/dtlab/ingest/mqtt/InitJavaLogging.scala

@@ -0,0 +1,72 @@
+package somind.dtlab.ingest.mqtt
+
+import java.util.logging.Level
+
+import com.typesafe.scalalogging.LazyLogging
+import org.slf4j.LoggerFactory
+
+import scala.sys.env
+
+/**
+  * If you have java a lib dep like eclipse.paho that you want to see logging
+  * from, it sucks to be you.
+  *
+  * You must init the java.util.logging subsystem with jul-to-slf4j to get
+  * Logback and LazyLogging to all play together :(
+  *
+  * dependent on build.sbt libs
+  *  "ch.qos.logback" % "logback-classic" % "?.?.?",
+  *  "org.slf4j" % "jul-to-slf4j" % "?.?.?",
+  */
+object InitJavaLogging extends LazyLogging {
+
+  def apply(): Unit = {
+    import java.util.logging.LogManager
+
+    import org.slf4j.bridge.SLF4JBridgeHandler
+
+    val pin: List[java.util.logging.Logger] = List(
+      java.util.logging.Logger.getLogger(""),
+      java.util.logging.Logger.getLogger("org.eclipse.paho.client.mqttv"),
+      java.util.logging.Logger.getLogger("org.eclipse.paho")
+    )
+
+    def canInstallBridgeHandler = {
+      try {
+        Class.forName("org.slf4j.impl.JDK14LoggerFactory",
+                      false,
+                      this.getClass.getClassLoader)
+        LoggerFactory
+          .getLogger(this.getClass)
+          .warn("Detected [org.slf4j.impl.JDK14LoggerFactory] on classpath. " +
+            "SLF4JBridgeHandler cannot be installed, see: http://www.slf4j.org/legacy.html#julRecursion")
+        false
+      } catch {
+        case _: ClassNotFoundException =>
+          true
+      }
+    }
+
+    val sysLogLevel = env.getOrElse("SYS_LOG_LEVEL", "info") match {
+      case "debug" => Level.FINER
+      case "trace" => Level.FINEST
+      case "error" => Level.WARNING
+      case _       => Level.INFO
+    }
+
+    if (canInstallBridgeHandler) {
+      LogManager.getLogManager.reset()
+      SLF4JBridgeHandler.removeHandlersForRootLogger()
+      SLF4JBridgeHandler.install()
+
+      java.util.logging.Logger
+        .getLogger("")
+        .addHandler(new SLF4JBridgeHandler())
+      pin.foreach(l => l.setLevel(sysLogLevel))
+
+      logger.info("org.slf4j.bridge.SLF4JBridgeHandler installed.")
+    }
+
+  }
+
+}

src/main/scala/somind/dtlab/ingest/mqtt/Main.scala

@@ -1,27 +1,45 @@
 package somind.dtlab.ingest.mqtt
 
 import akka.Done
+import akka.stream.alpakka.mqtt._
+import akka.stream.alpakka.mqtt.scaladsl.{MqttMessageWithAck, MqttSource}
+import akka.stream.scaladsl.Source
 import com.typesafe.scalalogging.LazyLogging
-
-import scala.concurrent.Future
-import scala.util.{Failure, Success}
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence
 import Conf._
 
-import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
 
 object Main extends App with LazyLogging {
 
-  def handleTerminate(result: Future[Done]): Unit = {
-    result onComplete {
-      case Success(_) =>
-        logger.warn("success. but stream should not end!")
-        actorSystem.terminate()
-      case Failure(e) =>
-        logger.error(s"failure. stream should not end! $e", e)
-        actorSystem.terminate()
-    }
-  }
-
-  handleTerminate(Stream())
+  val sslContext = SslContextUtil()
+
+  val connectionSettings = MqttConnectionSettings(
+    mqttUrl,
+    mqttClientId,
+    new MemoryPersistence
+  ).withAutomaticReconnect(true)
+    .withServerUri(mqttUrl)
+    .withSocketFactory(sslContext.getSocketFactory)
+    .withCleanSession(false)
+
+  logger.debug(s"connection settings:\n$connectionSettings")
+
+  val src: Source[MqttMessageWithAck, Future[Done]] =
+    MqttSource.atLeastOnce(
+      connectionSettings.withClientId(
+        clientId = "somind-mqtt-gateway-" + System.nanoTime().toString),
+      MqttSubscriptions(mqttTopic, MqttQoS.AtLeastOnce),
+      bufferSize = 8
+    )
+
+  logger.info("starting...")
+  logger.debug("debugging...")
+  src
+    .map(m => {
+      m.ack()
+      m.message.payload.map(_.toChar).mkString
+    })
+    .runWith(akka.stream.scaladsl.Sink.foreach(println))
 
 }

src/main/scala/somind/dtlab/ingest/mqtt/SslContextUtil.scala

@@ -0,0 +1,31 @@
+package somind.dtlab.ingest.mqtt
+
+import java.io.FileInputStream
+import java.security.KeyStore
+
+import com.typesafe.scalalogging.LazyLogging
+import javax.net.ssl.{KeyManagerFactory, SSLContext, TrustManagerFactory}
+import Conf._
+
+object SslContextUtil extends LazyLogging {
+
+  def apply(): SSLContext = {
+
+    val keyStore = KeyStore.getInstance("pkcs12")
+    keyStore.load(new FileInputStream(keyStorePath),
+                  keyStorePassword.toCharArray)
+    val kmf =
+      KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm)
+    kmf.init(keyStore, keyStorePassword.toCharArray)
+
+    val trustManagerFactory =
+      TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm)
+    trustManagerFactory.init(keyStore)
+
+    val ctx: SSLContext = SSLContext.getInstance("TLS")
+    ctx.init(kmf.getKeyManagers, trustManagerFactory.getTrustManagers, null)
+    ctx
+
+  }
+
+}