Implemented ElasticSearchLogSource.

Author: Alex Horsman

instrumentation/elasticsearch/src/main/scala/net/cakesolutions/testkit/logging/elasticsearch/ElasticSearchLogLineSource.scala

@@ -0,0 +1,66 @@
+// Copyright 2018 Cake Solutions Limited
+
+package net.cakesolutions.testkit.logging.elasticsearch
+
+import scala.concurrent.Future
+import com.sksamuel.elastic4s.aws.{Aws4ElasticClient, Aws4ElasticConfig}
+import com.sksamuel.elastic4s.http.RequestFailure
+import com.sksamuel.elastic4s.http.ElasticDsl._
+import com.sksamuel.elastic4s.searches.SearchDefinition
+import com.typesafe.scalalogging.Logger
+import monix.execution.{Cancelable, Scheduler}
+import monix.reactive.{Observable, OverflowStrategy}
+import net.cakesolutions.testkit.config.Configuration.Logging
+
+import scala.util.control.NonFatal
+
+final class ElasticSearchLogSource(
+    search: SearchDefinition,
+    config: Aws4ElasticConfig
+) {
+  import ElasticSearchLogSource._
+
+  private val logger = Logger(Logging.name)
+
+  def source()(implicit scheduler: Scheduler): Observable[String] =
+    Observable.create[String](
+      OverflowStrategy.Unbounded
+    ) { subscriber =>
+      try {
+        scheduler.execute {
+          new Runnable {
+            override def run(): Unit = {
+              val awsElasticClient = Aws4ElasticClient.apply(config)
+              Observable
+                .fromFuture(awsElasticClient.execute(search).flatMap {
+                  case Left(failure) =>
+                    Future.failed(
+                      ElasticSearchRequestFailureException(failure)
+                    )
+                  case Right(result) =>
+                    Future.successful(result)
+                })
+                .flatMap { result =>
+                  Observable.fromIterable(
+                    result.result.hits.hits.map(_.sourceAsString)
+                  )
+                }
+            }
+          }
+        }
+      } catch {
+        case NonFatal(error) =>
+          logger.error("Log parsing error", error)
+          subscriber.onError(error)
+      }
+      Cancelable(() => subscriber.onComplete())
+    }
+}
+
+object ElasticSearchLogSource {
+  case class ElasticSearchRequestFailureException(
+      requestFailure: RequestFailure
+  ) extends Exception(
+        s"ElasticSearch request failed: ${requestFailure.error.reason}"
+      )
+}