Make read and connection timeout configurable

6 files changed, 51 insertions, 15 deletions

diff --git a/async-http-client-handler/cats/src/main/scala/com/softwaremill/sttp/asynchttpclient/cats/AsyncHttpClientCatsHandler.scala b/async-http-client-handler/cats/src/main/scala/com/softwaremill/sttp/asynchttpclient/cats/AsyncHttpClientCatsHandler.scala
index 18949c0..906f090 100644
--- a/async-http-client-handler/cats/src/main/scala/com/softwaremill/sttp/asynchttpclient/cats/AsyncHttpClientCatsHandler.scala
+++ b/async-http-client-handler/cats/src/main/scala/com/softwaremill/sttp/asynchttpclient/cats/AsyncHttpClientCatsHandler.scala
@@ -16,6 +16,7 @@ import org.asynchttpclient.{
 }
 import org.reactivestreams.Publisher
 
+import scala.concurrent.duration.FiniteDuration
 import scala.language.higherKinds
 
 class AsyncHttpClientCatsHandler[F[_]: Async] private (
@@ -46,8 +47,12 @@ object AsyncHttpClientCatsHandler {
     new FollowRedirectsHandler[F, Nothing](
       new AsyncHttpClientCatsHandler(asyncHttpClient, closeClient))
 
-  def apply[F[_]: Async](): SttpHandler[F, Nothing] =
-    AsyncHttpClientCatsHandler(new DefaultAsyncHttpClient(), closeClient = true)
+  def apply[F[_]: Async](connectionTimeout: FiniteDuration = SttpHandler.DefaultConnectionTimeout)
+    : SttpHandler[F, Nothing] =
+    AsyncHttpClientCatsHandler(
+      new DefaultAsyncHttpClient(
+        AsyncHttpClientHandler.withConnectionTimeout(connectionTimeout)),
+      closeClient = true)
 
   def usingConfig[F[_]: Async](
       cfg: AsyncHttpClientConfig): SttpHandler[F, Nothing] =
diff --git a/async-http-client-handler/fs2/src/main/scala/com/softwaremill/sttp/asynchttpclient/fs2/AsyncHttpClientFs2Handler.scala b/async-http-client-handler/fs2/src/main/scala/com/softwaremill/sttp/asynchttpclient/fs2/AsyncHttpClientFs2Handler.scala
index 56e7b8c..8d38f43 100644
--- a/async-http-client-handler/fs2/src/main/scala/com/softwaremill/sttp/asynchttpclient/fs2/AsyncHttpClientFs2Handler.scala
+++ b/async-http-client-handler/fs2/src/main/scala/com/softwaremill/sttp/asynchttpclient/fs2/AsyncHttpClientFs2Handler.scala
@@ -21,6 +21,7 @@ import org.asynchttpclient.{
 import org.reactivestreams.Publisher
 
 import scala.concurrent.ExecutionContext
+import scala.concurrent.duration.FiniteDuration
 import scala.language.higherKinds
 
 class AsyncHttpClientFs2Handler[F[_]: Effect] private (
@@ -63,11 +64,13 @@ object AsyncHttpClientFs2Handler {
     *           e.g. mapping responses. Defaults to the global execution
     *           context.
     */
-  def apply[F[_]: Effect]()(
+  def apply[F[_]: Effect](connectionTimeout: FiniteDuration = SttpHandler.DefaultConnectionTimeout)(
       implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
     : SttpHandler[F, Stream[F, ByteBuffer]] =
-    AsyncHttpClientFs2Handler[F](new DefaultAsyncHttpClient(),
-                                 closeClient = true)
+    AsyncHttpClientFs2Handler[F](
+      new DefaultAsyncHttpClient(
+        AsyncHttpClientHandler.withConnectionTimeout(connectionTimeout)),
+      closeClient = true)
 
   /**
     * @param ec The execution context for running non-network related operations,
diff --git a/async-http-client-handler/future/src/main/scala/com/softwaremill/sttp/asynchttpclient/future/AsyncHttpClientFutureHandler.scala b/async-http-client-handler/future/src/main/scala/com/softwaremill/sttp/asynchttpclient/future/AsyncHttpClientFutureHandler.scala
index b80a91e..bbcc9c2 100644
--- a/async-http-client-handler/future/src/main/scala/com/softwaremill/sttp/asynchttpclient/future/AsyncHttpClientFutureHandler.scala
+++ b/async-http-client-handler/future/src/main/scala/com/softwaremill/sttp/asynchttpclient/future/AsyncHttpClientFutureHandler.scala
@@ -11,6 +11,7 @@ import org.asynchttpclient.{
 }
 import org.reactivestreams.Publisher
 
+import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.{ExecutionContext, Future}
 
 class AsyncHttpClientFutureHandler private (
@@ -44,10 +45,13 @@ object AsyncHttpClientFutureHandler {
     *           e.g. mapping responses. Defaults to the global execution
     *           context.
     */
-  def apply()(implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
+  def apply(connectionTimeout: FiniteDuration = SttpHandler.DefaultConnectionTimeout)(
+      implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
     : SttpHandler[Future, Nothing] =
-    AsyncHttpClientFutureHandler(new DefaultAsyncHttpClient(),
-                                 closeClient = true)
+    AsyncHttpClientFutureHandler(
+      new DefaultAsyncHttpClient(
+        AsyncHttpClientHandler.withConnectionTimeout(connectionTimeout)),
+      closeClient = true)
 
   /**
     * @param ec The execution context for running non-network related operations,
diff --git a/async-http-client-handler/monix/src/main/scala/com/softwaremill/sttp/asynchttpclient/monix/AsyncHttpClientMonixHandler.scala b/async-http-client-handler/monix/src/main/scala/com/softwaremill/sttp/asynchttpclient/monix/AsyncHttpClientMonixHandler.scala
index 311d3ea..457726a 100644
--- a/async-http-client-handler/monix/src/main/scala/com/softwaremill/sttp/asynchttpclient/monix/AsyncHttpClientMonixHandler.scala
+++ b/async-http-client-handler/monix/src/main/scala/com/softwaremill/sttp/asynchttpclient/monix/AsyncHttpClientMonixHandler.scala
@@ -20,6 +20,7 @@ import org.asynchttpclient.{
 }
 import org.reactivestreams.Publisher
 
+import scala.concurrent.duration.FiniteDuration
 import scala.util.{Failure, Success}
 
 class AsyncHttpClientMonixHandler private (
@@ -61,10 +62,13 @@ object AsyncHttpClientMonixHandler {
     * @param s The scheduler used for streaming request bodies. Defaults to the
     *          global scheduler.
     */
-  def apply()(implicit s: Scheduler = Scheduler.Implicits.global)
+  def apply(connectionTimeout: FiniteDuration = SttpHandler.DefaultConnectionTimeout)(
+      implicit s: Scheduler = Scheduler.Implicits.global)
     : SttpHandler[Task, Observable[ByteBuffer]] =
-    AsyncHttpClientMonixHandler(new DefaultAsyncHttpClient(),
-                                closeClient = true)
+    AsyncHttpClientMonixHandler(
+      new DefaultAsyncHttpClient(
+        AsyncHttpClientHandler.withConnectionTimeout(connectionTimeout)),
+      closeClient = true)
 
   /**
     * @param s The scheduler used for streaming request bodies. Defaults to the
diff --git a/async-http-client-handler/scalaz/src/main/scala/com/softwaremill/sttp/asynchttpclient/scalaz/AsyncHttpClientScalazHandler.scala b/async-http-client-handler/scalaz/src/main/scala/com/softwaremill/sttp/asynchttpclient/scalaz/AsyncHttpClientScalazHandler.scala
index b470606..88555fd 100644
--- a/async-http-client-handler/scalaz/src/main/scala/com/softwaremill/sttp/asynchttpclient/scalaz/AsyncHttpClientScalazHandler.scala
+++ b/async-http-client-handler/scalaz/src/main/scala/com/softwaremill/sttp/asynchttpclient/scalaz/AsyncHttpClientScalazHandler.scala
@@ -15,6 +15,7 @@ import org.asynchttpclient.{
 }
 import org.reactivestreams.Publisher
 
+import scala.concurrent.duration.FiniteDuration
 import scalaz.{-\/, \/-}
 import scalaz.concurrent.Task
 
@@ -42,12 +43,17 @@ object AsyncHttpClientScalazHandler {
     new FollowRedirectsHandler[Task, Nothing](
       new AsyncHttpClientScalazHandler(asyncHttpClient, closeClient))
 
-  def apply(): SttpHandler[Task, Nothing] =
-    AsyncHttpClientScalazHandler(new DefaultAsyncHttpClient(),
-                                 closeClient = true)
+  def apply(connectionTimeout: FiniteDuration = SttpHandler.DefaultConnectionTimeout)
+    : SttpHandler[Task, Nothing] =
+    AsyncHttpClientScalazHandler(
+      new DefaultAsyncHttpClient(
+        AsyncHttpClientHandler.withConnectionTimeout(connectionTimeout)),
+      closeClient = true)
+
   def usingConfig(cfg: AsyncHttpClientConfig): SttpHandler[Task, Nothing] =
     AsyncHttpClientScalazHandler(new DefaultAsyncHttpClient(cfg),
                                  closeClient = true)
+
   def usingClient(client: AsyncHttpClient): SttpHandler[Task, Nothing] =
     AsyncHttpClientScalazHandler(client, closeClient = false)
 }
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 b6c9249..8467399 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
@@ -17,6 +17,7 @@ import org.asynchttpclient.{
   AsyncCompletionHandler,
   AsyncHandler,
   AsyncHttpClient,
+  DefaultAsyncHttpClientConfig,
   HttpResponseBodyPart,
   HttpResponseHeaders,
   HttpResponseStatus,
@@ -28,6 +29,7 @@ import org.asynchttpclient.{
 import org.reactivestreams.{Publisher, Subscriber, Subscription}
 
 import scala.collection.JavaConverters._
+import scala.concurrent.duration.FiniteDuration
 import scala.language.higherKinds
 import scala.util.{Failure, Try}
 
@@ -155,7 +157,10 @@ abstract class AsyncHttpClientHandler[R[_], S](asyncHttpClient: AsyncHttpClient,
   }
 
   private def requestToAsync(r: Request[_, S]): AsyncRequest = {
-    val rb = new RequestBuilder(r.method.m).setUrl(r.uri.toString)
+    val rb = new RequestBuilder(r.method.m)
+      .setUrl(r.uri.toString)
+      .setRequestTimeout(
+        if (r.readTimeout.isFinite()) r.readTimeout.toMillis.toInt else -1)
     r.headers.foreach { case (k, v) => rb.setHeader(k, v) }
     setBody(r, r.body, rb)
     rb.build()
@@ -289,6 +294,15 @@ abstract class AsyncHttpClientHandler[R[_], S](asyncHttpClient: AsyncHttpClient,
   }
 }
 
+object AsyncHttpClientHandler {
+
+  private[asynchttpclient] def withConnectionTimeout(t: FiniteDuration) = {
+    new DefaultAsyncHttpClientConfig.Builder()
+      .setConnectTimeout(t.toMillis.toInt)
+      .build()
+  }
+}
+
 object EmptyPublisher extends Publisher[ByteBuffer] {
   override def subscribe(s: Subscriber[_ >: ByteBuffer]): Unit = {
     s.onComplete()