Closing the async http client if created by sttp

4 files changed, 37 insertions, 18 deletions

diff --git a/async-http-client-handler/future/src/main/scala/com/softwaremill/sttp/asynchttpclient/future/FutureAsyncHttpClientHandler.scala b/async-http-client-handler/future/src/main/scala/com/softwaremill/sttp/asynchttpclient/future/FutureAsyncHttpClientHandler.scala
index a2e49a2..11808d3 100644
--- a/async-http-client-handler/future/src/main/scala/com/softwaremill/sttp/asynchttpclient/future/FutureAsyncHttpClientHandler.scala
+++ b/async-http-client-handler/future/src/main/scala/com/softwaremill/sttp/asynchttpclient/future/FutureAsyncHttpClientHandler.scala
@@ -15,10 +15,12 @@ import org.reactivestreams.Publisher
 
 import scala.concurrent.{ExecutionContext, Future, Promise}
 
-class FutureAsyncHttpClientHandler private (asyncHttpClient: AsyncHttpClient)(
-    implicit ec: ExecutionContext)
+class FutureAsyncHttpClientHandler private (
+    asyncHttpClient: AsyncHttpClient,
+    closeClient: Boolean)(implicit ec: ExecutionContext)
     extends AsyncHttpClientHandler[Future, Nothing](asyncHttpClient,
-                                                    new FutureMonad()) {
+                                                    new FutureMonad(),
+                                                    closeClient) {
 
   override protected def streamBodyToPublisher(
       s: Nothing): Publisher[ByteBuffer] = s // nothing is everything
@@ -38,7 +40,8 @@ object FutureAsyncHttpClientHandler {
   def apply()(
       implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
     : FutureAsyncHttpClientHandler =
-    new FutureAsyncHttpClientHandler(new DefaultAsyncHttpClient())
+    new FutureAsyncHttpClientHandler(new DefaultAsyncHttpClient(),
+                                     closeClient = true)
 
   /**
     * @param ec The execution context for running non-network related operations,
@@ -48,7 +51,8 @@ object FutureAsyncHttpClientHandler {
   def usingConfig(cfg: AsyncHttpClientConfig)(
       implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
     : FutureAsyncHttpClientHandler =
-    new FutureAsyncHttpClientHandler(new DefaultAsyncHttpClient())
+    new FutureAsyncHttpClientHandler(new DefaultAsyncHttpClient(),
+                                     closeClient = true)
 
   /**
     * @param ec The execution context for running non-network related operations,
@@ -58,7 +62,7 @@ object FutureAsyncHttpClientHandler {
   def usingClient(client: AsyncHttpClient)(implicit ec: ExecutionContext =
                                              ExecutionContext.Implicits.global)
     : FutureAsyncHttpClientHandler =
-    new FutureAsyncHttpClientHandler(client)
+    new FutureAsyncHttpClientHandler(client, closeClient = false)
 }
 
 private[future] class FutureMonad(implicit ec: ExecutionContext)
diff --git a/async-http-client-handler/monix/src/main/scala/com/softwaremill/sttp/asynchttpclient/monix/MonixAsyncHttpClientHandler.scala b/async-http-client-handler/monix/src/main/scala/com/softwaremill/sttp/asynchttpclient/monix/MonixAsyncHttpClientHandler.scala
index c77e6d9..8e6c70d 100644
--- a/async-http-client-handler/monix/src/main/scala/com/softwaremill/sttp/asynchttpclient/monix/MonixAsyncHttpClientHandler.scala
+++ b/async-http-client-handler/monix/src/main/scala/com/softwaremill/sttp/asynchttpclient/monix/MonixAsyncHttpClientHandler.scala
@@ -18,11 +18,13 @@ import org.reactivestreams.Publisher
 
 import scala.util.{Failure, Success}
 
-class MonixAsyncHttpClientHandler private (asyncHttpClient: AsyncHttpClient)(
-    implicit scheduler: Scheduler)
+class MonixAsyncHttpClientHandler private (
+    asyncHttpClient: AsyncHttpClient,
+    closeClient: Boolean)(implicit scheduler: Scheduler)
     extends AsyncHttpClientHandler[Task, Observable[ByteBuffer]](
       asyncHttpClient,
-      TaskMonad) {
+      TaskMonad,
+      closeClient) {
 
   override protected def streamBodyToPublisher(
       s: Observable[ByteBuffer]): Publisher[ByteBuffer] = {
@@ -42,7 +44,8 @@ object MonixAsyncHttpClientHandler {
     */
   def apply()(implicit s: Scheduler = Scheduler.Implicits.global)
     : MonixAsyncHttpClientHandler =
-    new MonixAsyncHttpClientHandler(new DefaultAsyncHttpClient())
+    new MonixAsyncHttpClientHandler(new DefaultAsyncHttpClient(),
+                                    closeClient = true)
 
   /**
     * @param s The scheduler used for streaming request bodies. Defaults to the
@@ -51,7 +54,8 @@ object MonixAsyncHttpClientHandler {
   def usingConfig(cfg: AsyncHttpClientConfig)(implicit s: Scheduler =
                                                 Scheduler.Implicits.global)
     : MonixAsyncHttpClientHandler =
-    new MonixAsyncHttpClientHandler(new DefaultAsyncHttpClient())
+    new MonixAsyncHttpClientHandler(new DefaultAsyncHttpClient(),
+                                    closeClient = true)
 
   /**
     * @param s The scheduler used for streaming request bodies. Defaults to the
@@ -60,7 +64,7 @@ object MonixAsyncHttpClientHandler {
   def usingClient(client: AsyncHttpClient)(implicit s: Scheduler =
                                              Scheduler.Implicits.global)
     : MonixAsyncHttpClientHandler =
-    new MonixAsyncHttpClientHandler(client)
+    new MonixAsyncHttpClientHandler(client, closeClient = false)
 }
 
 private[monix] object TaskMonad extends MonadAsyncError[Task] {
diff --git a/async-http-client-handler/scalaz/src/main/scala/com/softwaremill/sttp/asynchttpclient/scalaz/ScalazAsyncHttpClientHandler.scala b/async-http-client-handler/scalaz/src/main/scala/com/softwaremill/sttp/asynchttpclient/scalaz/ScalazAsyncHttpClientHandler.scala
index 0460fff..f4de2eb 100644
--- a/async-http-client-handler/scalaz/src/main/scala/com/softwaremill/sttp/asynchttpclient/scalaz/ScalazAsyncHttpClientHandler.scala
+++ b/async-http-client-handler/scalaz/src/main/scala/com/softwaremill/sttp/asynchttpclient/scalaz/ScalazAsyncHttpClientHandler.scala
@@ -16,8 +16,11 @@ import org.reactivestreams.Publisher
 import scalaz.{-\/, \/-}
 import scalaz.concurrent.Task
 
-class ScalazAsyncHttpClientHandler private (asyncHttpClient: AsyncHttpClient)
-    extends AsyncHttpClientHandler[Task, Nothing](asyncHttpClient, TaskMonad) {
+class ScalazAsyncHttpClientHandler private (asyncHttpClient: AsyncHttpClient,
+                                            closeClient: Boolean)
+    extends AsyncHttpClientHandler[Task, Nothing](asyncHttpClient,
+                                                  TaskMonad,
+                                                  closeClient) {
 
   override protected def streamBodyToPublisher(
       s: Nothing): Publisher[ByteBuffer] = s // nothing is everything
@@ -29,11 +32,13 @@ class ScalazAsyncHttpClientHandler private (asyncHttpClient: AsyncHttpClient)
 
 object ScalazAsyncHttpClientHandler {
   def apply(): ScalazAsyncHttpClientHandler =
-    new ScalazAsyncHttpClientHandler(new DefaultAsyncHttpClient())
+    new ScalazAsyncHttpClientHandler(new DefaultAsyncHttpClient(),
+                                     closeClient = true)
   def usingConfig(cfg: AsyncHttpClientConfig): ScalazAsyncHttpClientHandler =
-    new ScalazAsyncHttpClientHandler(new DefaultAsyncHttpClient())
+    new ScalazAsyncHttpClientHandler(new DefaultAsyncHttpClient(),
+                                     closeClient = true)
   def usingClient(client: AsyncHttpClient): ScalazAsyncHttpClientHandler =
-    new ScalazAsyncHttpClientHandler(client)
+    new ScalazAsyncHttpClientHandler(client, closeClient = false)
 }
 
 private[scalaz] object TaskMonad extends MonadAsyncError[Task] {
diff --git a/async-http-client-handler/src/main/scala/com/softwaremill/sttp/asynchttpclient/AsyncHttpClientHandler.scala b/async-http-client-handler/src/main/scala/com/softwaremill/sttp/asynchttpclient/AsyncHttpClientHandler.scala
index f89c85a..1e924a0 100644
--- a/async-http-client-handler/src/main/scala/com/softwaremill/sttp/asynchttpclient/AsyncHttpClientHandler.scala
+++ b/async-http-client-handler/src/main/scala/com/softwaremill/sttp/asynchttpclient/AsyncHttpClientHandler.scala
@@ -30,7 +30,8 @@ import scala.language.higherKinds
 
 abstract class AsyncHttpClientHandler[R[_], S](
     asyncHttpClient: AsyncHttpClient,
-    rm: MonadAsyncError[R])
+    rm: MonadAsyncError[R],
+    closeClient: Boolean)
     extends SttpHandler[R, S] {
 
   override def send[T](r: Request[T, S]): R[Response[T]] = {
@@ -216,6 +217,11 @@ abstract class AsyncHttpClientHandler[R[_], S](
             "Requested a streaming response, trying to read eagerly."))
     }
   }
+
+  def close(): Unit = {
+    if (closeClient)
+      asyncHttpClient.close()
+  }
 }
 
 trait MonadAsyncError[R[_]] {