Explicit effect types

build.sbt

@@ -198,16 +198,18 @@ lazy val cats_effect_2 = (project in file("elastic4s-effect-cats-2"))
   .settings(libraryDependencies += cats2)
 
 lazy val zio_1 = (project in file("elastic4s-effect-zio-1"))
-  .dependsOn(core, testkit % "test")
+  .dependsOn(core, clientsttp % "test", testkit % "test")
   .settings(name := "elastic4s-effect-zio-1")
   .settings(scala3Settings)
   .settings(libraryDependencies ++= Dependencies.zio1)
+  .settings(libraryDependencies ++= Seq(Dependencies.sttpZio1Backend % "test"))
 
 lazy val zio = (project in file("elastic4s-effect-zio"))
-  .dependsOn(core, testkit % "test")
+  .dependsOn(core, clientsttp % "test", testkit % "test")
   .settings(name := "elastic4s-effect-zio")
   .settings(scala3Settings)
   .settings(libraryDependencies ++= Dependencies.zio)
+  .settings(libraryDependencies ++= Seq(Dependencies.sttpZioBackend % "test"))
 
 lazy val scalaz = (project in file("elastic4s-effect-scalaz"))
   .dependsOn(core)
@@ -302,7 +304,7 @@ lazy val clientsttp = (project in file("elastic4s-client-sttp"))
   .dependsOn(core, testkit % "test")
   .settings(name := "elastic4s-client-sttp")
   .settings(scala3Settings)
-  .settings(libraryDependencies ++= Seq(sttp, asyncHttpClientBackendFuture))
+  .settings(libraryDependencies ++= Seq(sttp, sttpFutureBackend))
 
 lazy val clientakka = (project in file("elastic4s-client-akka"))
   .dependsOn(core, testkit % "test")

elastic4s-client-akka/src/main/scala/com/sksamuel/elastic4s/akka/AkkaHttpClient.scala

@@ -23,7 +23,7 @@ class AkkaHttpClient private[akka] (
   blacklist: Blacklist,
   httpPoolFactory: HttpPoolFactory
 )(implicit system: ActorSystem)
-    extends ElasticHttpClient {
+    extends ElasticHttpClient[Future] {
 
   import AkkaHttpClient._
   import system.dispatcher
@@ -200,30 +200,16 @@ class AkkaHttpClient private[akka] (
     }
   }
 
-  private[akka] def sendAsync(
+  override def send(
     request: ElasticRequest
   ): Future[ElasticHttpResponse] = {
     queueRequestWithRetry(request)
   }
 
-  override def send(
-    request: ElasticRequest,
-    callback: Either[Throwable, ElasticHttpResponse] => Unit
-  ): Unit = {
-    sendAsync(request).onComplete {
-      case Success(r) => callback(Right(r))
-      case Failure(e) => callback(Left(e))
-    }
-  }
-
-  def shutdown(): Future[Unit] = {
+  override def close(): Future[Unit] = {
     httpPoolFactory.shutdown()
   }
 
-  override def close(): Unit = {
-    shutdown()
-  }
-
   private def toRequest(request: ElasticRequest,
                         host: String): Try[HttpRequest] = Try {
     val httpRequest = HttpRequest(

elastic4s-client-akka/src/test/scala/com/sksamuel/elastic4s/akka/AkkaHttpClientMockTest.scala

@@ -68,7 +68,7 @@ class AkkaHttpClientMockTest
         .thenReturn(Success(HttpResponse().withEntity("ok")))
 
       client
-        .sendAsync(ElasticRequest("GET", "/test"))
+        .send(ElasticRequest("GET", "/test"))
         .futureValue shouldBe ElasticResponse(
         200,
         Some(ElasticEntity.StringEntity("ok", None)),
@@ -102,7 +102,7 @@ class AkkaHttpClientMockTest
         .thenReturn(Success(HttpResponse(StatusCodes.BadGateway)))
 
       client
-        .sendAsync(ElasticRequest("GET", "/test"))
+        .send(ElasticRequest("GET", "/test"))
         .futureValue shouldBe ElasticResponse(
         502,
         Some(ElasticEntity.StringEntity("", None)),
@@ -141,7 +141,7 @@ class AkkaHttpClientMockTest
         .thenReturn(Success(HttpResponse().withEntity("host2")))
 
       client
-        .sendAsync(ElasticRequest("GET", "/test"))
+        .send(ElasticRequest("GET", "/test"))
         .futureValue
     }
 
@@ -177,7 +177,7 @@ class AkkaHttpClientMockTest
         .thenReturn(Success(HttpResponse().withEntity("host2")))
 
       client
-        .sendAsync(ElasticRequest("GET", "/test"))
+        .send(ElasticRequest("GET", "/test"))
         .futureValue
     }
 
@@ -207,7 +207,7 @@ class AkkaHttpClientMockTest
         .thenReturn(Success(HttpResponse().withEntity("host2")))
 
       client
-        .sendAsync(ElasticRequest("GET", "/test"))
+        .send(ElasticRequest("GET", "/test"))
         .futureValue shouldBe ElasticResponse(
         200,
         Some(ElasticEntity.StringEntity("host2", None)),

elastic4s-client-akka/src/test/scala/com/sksamuel/elastic4s/akka/AkkaHttpClientTest.scala

@@ -31,14 +31,16 @@ class AkkaHttpClientTest extends AnyFlatSpec with Matchers with DockerTests with
         deleteIndex("testindex")
       }.await
 
-      akkaClient.shutdown().await
+      akkaClient.close().await
       system.terminate().await
     }
   }
 
   private lazy val akkaClient = AkkaHttpClient(AkkaHttpClientSettings(List(s"$elasticHost:$elasticPort")))
 
-  override val client = ElasticClient(akkaClient)
+  def mkAkkaBasedClient(implicit executor: Executor[Future]): ElasticClient[Future] = ElasticClient(akkaClient)
+
+  override lazy val client = mkAkkaBasedClient
 
   "AkkaHttpClient" should "support utf-8" in {
 
@@ -106,14 +108,12 @@ class AkkaHttpClientTest extends AnyFlatSpec with Matchers with DockerTests with
   }
 
   it should "propagate headers if included" in {
-    implicit val executor: Executor[Future] = new Executor[Future] {
-      override def exec(client: HttpClient, request: ElasticRequest): Future[HttpResponse] = {
-        val cred = Base64.getEncoder.encodeToString("user123:pass123".getBytes(StandardCharsets.UTF_8))
-        Executor.FutureExecutor.exec(client, request.copy(headers = Map("Authorization" -> s"Basic $cred")))
-      }
+    implicit val executor: Executor[Future] = (client: HttpClient[Future], request: ElasticRequest) => {
+      val cred = Base64.getEncoder.encodeToString("user123:pass123".getBytes(StandardCharsets.UTF_8))
+      client.send(request.copy(headers = Map("Authorization" -> s"Basic $cred")))
     }
 
-    client.execute {
+    mkAkkaBasedClient.execute {
       catHealth()
     }.await.result.status shouldBe "401"
   }

elastic4s-client-esjava/src/main/scala/com/sksamuel/elastic4s/http/JavaClient.scala

@@ -12,6 +12,7 @@ import org.elasticsearch.client.RestClientBuilder.{HttpClientConfigCallback, Req
 import org.elasticsearch.client.{Request, ResponseException, ResponseListener, RestClient}
 import org.slf4j.{Logger, LoggerFactory}
 
+import scala.concurrent.{ExecutionContext, Future, Promise}
 import scala.io.{Codec, Source}
 import scala.language.higherKinds
 
@@ -20,7 +21,7 @@ case class JavaClientExceptionWrapper(t: Throwable) extends RuntimeException(t)
 /**
   * An implementation of HttpClient that wraps the Elasticsearch Java Rest Client
   */
-class JavaClient(client: RestClient) extends HttpClient {
+class JavaClient(client: RestClient)(implicit ec: ExecutionContext) extends HttpClient[Future] {
 
   def apacheEntity(entity: HttpEntity): AbstractHttpEntity = entity match {
     case e: HttpEntity.StringEntity =>
@@ -58,16 +59,18 @@ class JavaClient(client: RestClient) extends HttpClient {
     HttpResponse(r.getStatusLine.getStatusCode, entity, headers)
   }
 
-  override def send(req: ElasticRequest, callback: Either[Throwable, HttpResponse] => Unit): Unit = {
+  override def send(req: ElasticRequest): Future[HttpResponse] = {
     if (logger.isDebugEnabled) {
       logger.debug("Executing elastic request {}", Show[ElasticRequest].show(req))
     }
 
+    val promise = Promise[HttpResponse]()
+
     val l = new ResponseListener {
-      override def onSuccess(r: org.elasticsearch.client.Response): Unit = callback(Right(fromResponse(r)))
+      override def onSuccess(r: org.elasticsearch.client.Response): Unit = promise.success(fromResponse(r))
       override def onFailure(e: Exception): Unit = e match {
-        case re: ResponseException => callback(Right(fromResponse(re.getResponse)))
-        case t => callback(Left(JavaClientExceptionWrapper(t)))
+        case re: ResponseException => promise.success(fromResponse(re.getResponse))
+        case t => promise.failure(JavaClientExceptionWrapper(t))
       }
     }
 
@@ -78,9 +81,11 @@ class JavaClient(client: RestClient) extends HttpClient {
     req.headers.foreach((optBuilder.addHeader _).tupled)
     request.setOptions(optBuilder)
     client.performRequestAsync(request, l)
+
+    promise.future
   }
 
-  override def close(): Unit = client.close()
+  override def close(): Future[Unit] = Future(client.close())
 
   private def isEntityGziped(entity: org.apache.http.HttpEntity): Boolean = {
     Option(entity.getContentEncoding).flatMap(x => Option(x.getValue)).contains("gzip")
@@ -97,27 +102,33 @@ object JavaClient {
     * @param client the Java client to wrap
     * @return newly created Scala client
     */
-  def fromRestClient(client: RestClient): JavaClient = new JavaClient(client)
+  def fromRestClient(client: RestClient)(implicit ec: ExecutionContext): JavaClient = new JavaClient(client)
 
   /**
     * Creates a new [[ElasticClient]] using the elasticsearch Java API rest client
     * as the underlying client. Optional callbacks can be passed in to configure the client.
     */
-  def apply(props: ElasticProperties): JavaClient =
+  def apply(props: ElasticProperties)(implicit ec: ExecutionContext): JavaClient =
     apply(props, NoOpRequestConfigCallback, NoOpHttpClientConfigCallback)
 
   /**
     * Creates a new [[ElasticClient]] using the elasticsearch Java API rest client
     * as the underlying client. Optional callbacks can be passed in to configure the client.
     */
-  def apply(props: ElasticProperties, requestConfigCallback: RequestConfigCallback): JavaClient =
+  def apply(
+    props: ElasticProperties,
+    requestConfigCallback: RequestConfigCallback
+  )(implicit ec: ExecutionContext): JavaClient =
     apply(props, requestConfigCallback, NoOpHttpClientConfigCallback)
 
   /**
     * Creates a new [[ElasticClient]] using the elasticsearch Java API rest client
     * as the underlying client. Optional callbacks can be passed in to configure the client.
     */
-  def apply(props: ElasticProperties, httpClientConfigCallback: HttpClientConfigCallback): JavaClient =
+  def apply(
+    props: ElasticProperties,
+    httpClientConfigCallback: HttpClientConfigCallback
+  )(implicit ec: ExecutionContext): JavaClient =
     apply(props, NoOpRequestConfigCallback, httpClientConfigCallback)
 
   /**
@@ -126,7 +137,7 @@ object JavaClient {
     */
   def apply(props: ElasticProperties,
             requestConfigCallback: RequestConfigCallback,
-            httpClientConfigCallback: HttpClientConfigCallback): JavaClient = {
+            httpClientConfigCallback: HttpClientConfigCallback)(implicit ec: ExecutionContext): JavaClient = {
     val hosts = props.endpoints.map {
       case ElasticNodeEndpoint(protocol, host, port, _) => new HttpHost(host, port, protocol)
     }

elastic4s-client-pekko/src/main/scala/com/sksamuel/elastic4s/pekko/PekkoHttpClient.scala

@@ -23,7 +23,7 @@ class PekkoHttpClient private[pekko](
   blacklist: Blacklist,
   httpPoolFactory: HttpPoolFactory
 )(implicit system: ActorSystem)
-    extends ElasticHttpClient {
+    extends ElasticHttpClient[Future] {
 
   import PekkoHttpClient._
   import system.dispatcher
@@ -200,30 +200,14 @@ class PekkoHttpClient private[pekko](
     }
   }
 
-  private[pekko] def sendAsync(
-    request: ElasticRequest
-  ): Future[ElasticHttpResponse] = {
+  override def send(request: ElasticRequest): Future[ElasticHttpResponse] = {
     queueRequestWithRetry(request)
   }
 
-  override def send(
-    request: ElasticRequest,
-    callback: Either[Throwable, ElasticHttpResponse] => Unit
-  ): Unit = {
-    sendAsync(request).onComplete {
-      case Success(r) => callback(Right(r))
-      case Failure(e) => callback(Left(e))
-    }
-  }
-
-  def shutdown(): Future[Unit] = {
+  override def close(): Future[Unit] = {
     httpPoolFactory.shutdown()
   }
 
-  override def close(): Unit = {
-    shutdown()
-  }
-
   private def toRequest(request: ElasticRequest,
                         host: String): Try[HttpRequest] = Try {
     val httpRequest = HttpRequest(

elastic4s-client-pekko/src/test/scala/com/sksamuel/elastic4s/pekko/PekkoHttpClientMockTest.scala

@@ -68,7 +68,7 @@ class PekkoHttpClientMockTest
         .thenReturn(Success(HttpResponse().withEntity("ok")))
 
       client
-        .sendAsync(ElasticRequest("GET", "/test"))
+        .send(ElasticRequest("GET", "/test"))
         .futureValue shouldBe ElasticResponse(
         200,
         Some(ElasticEntity.StringEntity("ok", None)),
@@ -102,7 +102,7 @@ class PekkoHttpClientMockTest
         .thenReturn(Success(HttpResponse(StatusCodes.BadGateway)))
 
       client
-        .sendAsync(ElasticRequest("GET", "/test"))
+        .send(ElasticRequest("GET", "/test"))
         .futureValue shouldBe ElasticResponse(
         502,
         Some(ElasticEntity.StringEntity("", None)),
@@ -141,7 +141,7 @@ class PekkoHttpClientMockTest
         .thenReturn(Success(HttpResponse().withEntity("host2")))
 
       client
-        .sendAsync(ElasticRequest("GET", "/test"))
+        .send(ElasticRequest("GET", "/test"))
         .futureValue
     }
 
@@ -177,7 +177,7 @@ class PekkoHttpClientMockTest
         .thenReturn(Success(HttpResponse().withEntity("host2")))
 
       client
-        .sendAsync(ElasticRequest("GET", "/test"))
+        .send(ElasticRequest("GET", "/test"))
         .futureValue
     }
 
@@ -207,7 +207,7 @@ class PekkoHttpClientMockTest
         .thenReturn(Success(HttpResponse().withEntity("host2")))
 
       client
-        .sendAsync(ElasticRequest("GET", "/test"))
+        .send(ElasticRequest("GET", "/test"))
         .futureValue shouldBe ElasticResponse(
         200,
         Some(ElasticEntity.StringEntity("host2", None)),

elastic4s-client-pekko/src/test/scala/com/sksamuel/elastic4s/pekko/PekkoHttpClientTest.scala

@@ -31,14 +31,17 @@ class PekkoHttpClientTest extends AnyFlatSpec with Matchers with DockerTests wit
         deleteIndex("testindex")
       }.await
 
-      pekkoClient.shutdown().await
+      pekkoClient.close().await
       system.terminate().await
     }
   }
 
   private lazy val pekkoClient = PekkoHttpClient(PekkoHttpClientSettings(List(s"$elasticHost:$elasticPort")))
 
-  override val client = ElasticClient(pekkoClient)
+  def mkPekkoBasedClient(implicit executor: Executor[Future]): ElasticClient[Future] =
+    ElasticClient(pekkoClient)
+
+  override lazy val client = mkPekkoBasedClient
 
   "PekkoHttpClient" should "support utf-8" in {
 
@@ -106,14 +109,12 @@ class PekkoHttpClientTest extends AnyFlatSpec with Matchers with DockerTests wit
   }
 
   it should "propagate headers if included" in {
-    implicit val executor: Executor[Future] = new Executor[Future] {
-      override def exec(client: HttpClient, request: ElasticRequest): Future[HttpResponse] = {
-        val cred = Base64.getEncoder.encodeToString("user123:pass123".getBytes(StandardCharsets.UTF_8))
-        Executor.FutureExecutor.exec(client, request.copy(headers = Map("Authorization" -> s"Basic $cred")))
-      }
+    implicit val executor: Executor[Future] = (client: HttpClient[Future], request: ElasticRequest) => {
+      val cred = Base64.getEncoder.encodeToString("user123:pass123".getBytes(StandardCharsets.UTF_8))
+      client.send(request.copy(headers = Map("Authorization" -> s"Basic $cred")))
     }
 
-    client.execute {
+    mkPekkoBasedClient.execute {
       catHealth()
     }.await.result.status shouldBe "401"
   }