good long running tests

Author: navicore

src/main/resources/application.conf

@@ -23,8 +23,8 @@ main {
   mqttClientId = "dtLabClient"
   mqttClientId = ${?MQTT_CLIENT_ID}
 
-  dtlabIngestUri = "http://localhost:8082/dtlab-alligator/extractor/telemetry/CHANGEME"
-  dtlabIngestUri = ${?DTLAB_INGEST_PATH}
+  dtlabIngestUris = "http://localhost:8082/dtlab-alligator/extractor/telemetry/CHANGEME http://localhost:8082/dtlab-alligator/extractor/array/CHANGEME"
+  dtlabIngestUris = ${?DTLAB_INGEST_PATHS}
 
   telemetryContentType = "json"
   telemetryContentType = ${?TELEMETRY_CONTENT_TYPE}

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

@@ -26,7 +26,7 @@ object Conf extends LazyLogging {
   val mqttTopic: String = conf.getString("main.mqttTopic")
   val keyStorePassword: String = conf.getString("main.keyStorePassword")
   val keyStorePath: String = conf.getString("main.keyStorePath")
-  val dtlabIngestUri: String = conf.getString("main.dtlabIngestUri")
+  val dtlabIngestUris: List[String] = conf.getString("main.dtlabIngestUris").split(' ').toList
   import scala.concurrent.duration._
   val webhookTimeoutSeconds: Duration =
     conf.getInt("main.webhookTimeoutSeconds").seconds

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

@@ -40,12 +40,13 @@ object MqttSourceWebhookSinkStream extends LazyLogging {
 
     src
       .runWith(akka.stream.scaladsl.Sink.foreach(m => {
-
+        // todo: move to an async flow that can scale horizontally
         Observer("mqtt_message_processing_by_sink")
         val payload = m.message.payload.map(_.toChar).mkString
         logger.debug(payload)
         PostString(payload) match {
-          case Some(code) if code == StatusCodes.Accepted =>
+          case Some(code)
+              if code == StatusCodes.Accepted || code == StatusCodes.OK =>
             m.ack()
             Observer("mqtt_message_processed_by_sink_fitness")
           case code =>

src/main/scala/somind/dtlab/ingest/mqtt/utils/PostString.scala

@@ -10,21 +10,25 @@ import scala.concurrent.{Await, Future, TimeoutException}
 object PostString extends LazyLogging {
 
   val http: HttpExt = Http(actorSystem)
-  val request: HttpRequest = HttpRequest()
-    .withUri(uri = Uri(dtlabIngestUri))
+  val request: HttpRequest = HttpRequest().withMethod(HttpMethods.POST)
 
   def apply(payload: String): Option[StatusCode] = {
     try {
-      val f = PostString.applyAsync(payload)
-      val r = Await.result(f, webhookTimeoutSeconds)
-      if (!r.status.isSuccess()) {
-        logger.warn(s"post not successful: ${r._3}")
-      }
-      Some(r.status)
+
+      val results = dtlabIngestUris.map(uriString => {
+        val f = PostString.applyAsync(payload, uriString)
+        val r = Await.result(f, webhookTimeoutSeconds)
+        if (!r.status.isSuccess()) {
+          logger.warn(s"post not successful: $r")
+          return None
+        }
+        logger.debug(s"post successful")
+        Some(r.status)
+      })
+      results.last
     } catch {
       case e: TimeoutException =>
-        logger.error(
-          s"remote system is timing out: ${e.getMessage}")
+        logger.error(s"remote system is timing out: ${e.getMessage}")
         System.exit(1)
         None
       case e: Throwable =>
@@ -34,12 +38,14 @@ object PostString extends LazyLogging {
     }
   }
 
-  def applyAsync(telem: String): Future[HttpResponse] = {
+  def applyAsync(telem: String, uriString: String): Future[HttpResponse] = {
 
     val newRequest =
-      request.withEntity(entity = HttpEntity(telemetryContentType, telem))
+      request
+        .withEntity(entity = HttpEntity(telemetryContentType, telem))
+        .withUri(uri = Uri(uriString))
 
-    logger.debug(s"sending request to: " + newRequest)
+    logger.debug(s"posting request to: " + newRequest)
     http.singleRequest(newRequest)
   }