(executor): Provide simple HTTP executor

.github/workflows/ci.yml

@@ -35,7 +35,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        java: ['8', '11', '17']
+        java: ['11', '17']
         scala: ['2.12.16', '2.13.8']
     steps:
       - name: Checkout current branch

build.sbt

@@ -28,18 +28,26 @@ lazy val library =
   project
     .in(file("modules/library"))
     .settings(stdSettings("zio-elasticsearch"))
+    .settings(scalacOptions += "-language:higherKinds")
     .settings(
       libraryDependencies ++= List(
-        "dev.zio" %% "zio-json"        % "0.3.0",
-        "dev.zio" %% "zio-schema"      % "0.3.1",
-        "dev.zio" %% "zio-schema-json" % "0.3.1"
+        "dev.zio"                       %% "zio-json"        % "0.3.0",
+        "dev.zio"                       %% "zio-schema"      % "0.3.1",
+        "dev.zio"                       %% "zio-schema-json" % "0.3.1",
+        "com.softwaremill.sttp.client3" %% "zio"             % "3.8.3",
+        "com.softwaremill.sttp.client3" %% "zio-json"        % "3.8.3"
       )
     )
 
 lazy val example =
   project
     .in(file("modules/example"))
     .settings(stdSettings("example"))
+    .settings(
+      libraryDependencies ++= List(
+        "dev.zio" %% "zio" % "2.0.4"
+      )
+    )
     .dependsOn(library)
     .settings(
       publish / skip := true

modules/example/docker-compose.yml

@@ -0,0 +1,12 @@
+version: '3.8'
+
+services:
+  elasticsearch:
+    image: elasticsearch:7.17.6
+    container_name: zio-elasticsearch-example
+    ports:
+      - "9200:9200"
+    environment:
+      discovery.type: "single-node"
+      xpack.security.enabled: "false"
+      ES_JAVA_OPTS: "-Xms512m -Xmx512m"

modules/example/src/main/scala/example/ExampleApp.scala

@@ -0,0 +1,21 @@
+package example
+
+import sttp.client3.httpclient.zio.HttpClientZioBackend
+import zio._
+import zio.elasticsearch._
+
+object ExampleApp extends ZIOAppDefault {
+
+  override def run: Task[Unit] = {
+    val index   = IndexName("kibana_sample_data_ecommerce")
+    val docId   = DocumentId("wzvbqIQB22WMP-4s4h8K")
+    val routing = Some(Routing("10"))
+
+    (for {
+      _   <- Console.printLine("Welcome to an example app...")
+      _   <- Console.printLine(s"Looking for the document '$docId' in '$index' index...'")
+      res <- ElasticRequest.getById[ExampleDocument](index, docId, routing).execute
+      _   <- Console.printLine(res)
+    } yield ()).provide(ElasticExecutor.local, HttpClientZioBackend.layer())
+  }
+}

modules/example/src/main/scala/example/ExampleDocument.scala

@@ -0,0 +1,9 @@
+package example
+
+import zio.schema.{DeriveSchema, Schema}
+
+final case class ExampleDocument(id: String, name: String, count: Int)
+
+object ExampleDocument {
+  implicit val schema: Schema[ExampleDocument] = DeriveSchema.gen[ExampleDocument]
+}

modules/library/src/main/scala/zio/elasticsearch/Document.scala

@@ -1,5 +1,6 @@
 package zio.elasticsearch
 
+import zio.json.ast.Json
 import zio.schema.Schema
 import zio.schema.codec.JsonCodec.JsonDecoder
 import zio.schema.codec.{DecodeError, JsonCodec}
@@ -12,4 +13,6 @@ private[elasticsearch] object Document {
   def from[A](doc: A)(implicit schema: Schema[A]): Document = Document(
     JsonCodec.jsonEncoder(schema).encodeJson(doc, indent = None).toString
   )
+
+  def from(json: Json): Document = new Document(json.toString)
 }

modules/library/src/main/scala/zio/elasticsearch/DocumentId.scala

@@ -1,3 +1,5 @@
 package zio.elasticsearch
 
-final case class DocumentId(value: String) extends AnyVal
+final case class DocumentId(value: String) extends AnyVal {
+  override def toString: String = value
+}

modules/library/src/main/scala/zio/elasticsearch/ElasticConfig.scala

@@ -0,0 +1,7 @@
+package zio.elasticsearch
+
+final case class ElasticConfig(host: String, port: Int)
+
+object ElasticConfig {
+  lazy val Default: ElasticConfig = ElasticConfig("localhost", 9200)
+}

modules/library/src/main/scala/zio/elasticsearch/ElasticExecutor.scala

@@ -0,0 +1,21 @@
+package zio.elasticsearch
+
+import sttp.client3.SttpBackend
+import zio.{Task, ZIO, ZLayer}
+
+trait ElasticExecutor {
+  def execute[A](request: ElasticRequest[A]): Task[A]
+}
+
+object ElasticExecutor {
+  lazy val live: ZLayer[ElasticConfig with SttpBackend[Task, Any], Throwable, ElasticExecutor] =
+    ZLayer {
+      for {
+        conf <- ZIO.service[ElasticConfig]
+        sttp <- ZIO.service[SttpBackend[Task, Any]]
+      } yield HttpElasticExecutor(conf, sttp)
+    }
+
+  lazy val local: ZLayer[SttpBackend[Task, Any], Throwable, ElasticExecutor] =
+    ZLayer.succeed(ElasticConfig.Default) >>> live
+}

modules/library/src/main/scala/zio/elasticsearch/ElasticRequest.scala

@@ -3,10 +3,13 @@ package zio.elasticsearch
 import zio.elasticsearch.ElasticError.DocumentRetrievingError._
 import zio.elasticsearch.ElasticError._
 import zio.schema.Schema
+import zio.{RIO, ZIO}
 
 sealed trait ElasticRequest[+A] { self =>
-  final def map[B](f: A => B): ElasticRequest[B] = ElasticRequest.Map(self, f)
+  final def execute: RIO[ElasticExecutor, A] =
+    ZIO.serviceWithZIO[ElasticExecutor](_.execute(self))
 
+  final def map[B](f: A => B): ElasticRequest[B] = ElasticRequest.Map(self, f)
 }
 
 object ElasticRequest {

modules/library/src/main/scala/zio/elasticsearch/ElasticResponse.scala

@@ -0,0 +1,14 @@
+package zio.elasticsearch
+
+import zio.json.ast.Json
+import zio.json.{DeriveJsonDecoder, JsonDecoder, jsonField}
+
+final case class ElasticResponse(
+  found: Boolean,
+  @jsonField("_source")
+  source: Json
+)
+
+object ElasticResponse {
+  implicit val decoder: JsonDecoder[ElasticResponse] = DeriveJsonDecoder.gen[ElasticResponse]
+}

modules/library/src/main/scala/zio/elasticsearch/HttpElasticExecutor.scala

@@ -0,0 +1,42 @@
+package zio.elasticsearch
+
+import sttp.client3._
+import sttp.client3.ziojson._
+import sttp.model.Uri
+import zio.elasticsearch.ElasticRequest._
+import zio.{Task, ZIO}
+
+private[elasticsearch] final class HttpElasticExecutor private (config: ElasticConfig, client: SttpBackend[Task, Any])
+    extends ElasticExecutor {
+
+  import HttpElasticExecutor._
+
+  private val basePath = Uri(config.host, config.port)
+
+  override def execute[A](request: ElasticRequest[A]): Task[A] =
+    request match {
+      case _: Create         => ZIO.unit
+      case _: CreateOrUpdate => ZIO.unit
+      case r: GetById        => executeGetById(r)
+      case map @ Map(_, _)   => execute(map.request).map(map.mapper)
+    }
+
+  private def executeGetById(getById: GetById): Task[Option[Document]] = {
+    val uri = uri"$basePath/${getById.index}/$Doc/${getById.id}".withParam("routing", getById.routing.map(_.value))
+    basicRequest
+      .get(uri)
+      .response(asJson[ElasticResponse])
+      .send(client)
+      .map(_.body.toOption)
+      .map(_.flatMap(d => if (d.found) Option(Document.from(d.source)) else None))
+  }
+
+}
+
+private[elasticsearch] object HttpElasticExecutor {
+
+  private final val Doc = "_doc"
+
+  def apply(config: ElasticConfig, client: SttpBackend[Task, Any]) =
+    new HttpElasticExecutor(config, client)
+}

modules/library/src/main/scala/zio/elasticsearch/IndexName.scala

@@ -1,3 +1,5 @@
 package zio.elasticsearch
 
-final case class IndexName(name: String) extends AnyVal
+final case class IndexName(name: String) extends AnyVal {
+  override def toString: String = name
+}