Make read and connection timeout configurable

17 files changed, 210 insertions, 48 deletions

diff --git a/akka-http-handler/src/main/scala/com/softwaremill/sttp/akkahttp/AkkaHttpHandler.scala b/akka-http-handler/src/main/scala/com/softwaremill/sttp/akkahttp/AkkaHttpHandler.scala
index 2aa4251..da538a1 100644
--- a/akka-http-handler/src/main/scala/com/softwaremill/sttp/akkahttp/AkkaHttpHandler.scala
+++ b/akka-http-handler/src/main/scala/com/softwaremill/sttp/akkahttp/AkkaHttpHandler.scala
@@ -13,18 +13,22 @@ import akka.http.scaladsl.model.headers.{
   `Content-Type`
 }
 import akka.http.scaladsl.model.{Multipart => AkkaMultipart, _}
+import akka.http.scaladsl.settings.ClientConnectionSettings
+import akka.http.scaladsl.settings.ConnectionPoolSettings
 import akka.stream.ActorMaterializer
 import akka.stream.scaladsl.{FileIO, Source, StreamConverters}
 import akka.util.ByteString
 import com.softwaremill.sttp._
 
 import scala.collection.immutable.Seq
+import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.{ExecutionContext, Future}
 import scala.util.{Failure, Success, Try}
 
 class AkkaHttpHandler private (actorSystem: ActorSystem,
                                ec: ExecutionContext,
-                               terminateActorSystemOnClose: Boolean)
+                               terminateActorSystemOnClose: Boolean,
+                               connectionTimeout: FiniteDuration)
     extends SttpHandler[Future, Source[ByteString, Any]] {
 
   // the supported stream type
@@ -35,10 +39,19 @@ class AkkaHttpHandler private (actorSystem: ActorSystem,
 
   override def send[T](r: Request[T, S]): Future[Response[T]] = {
     implicit val ec: ExecutionContext = this.ec
+
+    val connectionSettings = ClientConnectionSettings(actorSystem)
+      .withIdleTimeout(r.readTimeout)
+      .withConnectingTimeout(connectionTimeout)
+
+    val connectionPoolSettings = ConnectionPoolSettings(actorSystem)
+      .withConnectionSettings(connectionSettings)
+
     requestToAkka(r)
       .flatMap(setBodyOnAkka(r, r.body, _))
       .toFuture
-      .flatMap(Http().singleRequest(_))
+      .flatMap(req =>
+        Http().singleRequest(req, settings = connectionPoolSettings))
       .flatMap { hr =>
         val code = hr.status.intValue()
 
@@ -271,19 +284,26 @@ object AkkaHttpHandler {
 
   private def apply(actorSystem: ActorSystem,
                     ec: ExecutionContext,
-                    terminateActorSystemOnClose: Boolean)
-    : SttpHandler[Future, Source[ByteString, Any]] =
+                    terminateActorSystemOnClose: Boolean,
+                    connectionTimeout: FiniteDuration): SttpHandler[Future, Source[ByteString, Any]] =
     new FollowRedirectsHandler(
-      new AkkaHttpHandler(actorSystem, ec, terminateActorSystemOnClose))
+      new AkkaHttpHandler(actorSystem,
+                          ec,
+                          terminateActorSystemOnClose,
+                          connectionTimeout))
 
   /**
     * @param ec The execution context for running non-network related operations,
     *           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, Source[ByteString, Any]] =
-    AkkaHttpHandler(ActorSystem("sttp"), ec, terminateActorSystemOnClose = true)
+    AkkaHttpHandler(ActorSystem("sttp"),
+                    ec,
+                    terminateActorSystemOnClose = true,
+                    connectionTimeout)
 
   /**
     * @param actorSystem The actor system which will be used for the http-client
@@ -292,8 +312,12 @@ object AkkaHttpHandler {
     *           e.g. mapping responses. Defaults to the global execution
     *           context.
     */
-  def usingActorSystem(actorSystem: ActorSystem)(
+  def usingActorSystem(actorSystem: ActorSystem,
+                       connectionTimeout: FiniteDuration = SttpHandler.DefaultConnectionTimeout)(
       implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
     : SttpHandler[Future, Source[ByteString, Any]] =
-    AkkaHttpHandler(actorSystem, ec, terminateActorSystemOnClose = false)
+    AkkaHttpHandler(actorSystem,
+                    ec,
+                    terminateActorSystemOnClose = false,
+                    connectionTimeout)
 }
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()
diff --git a/core/src/main/scala/com/softwaremill/sttp/HttpURLConnectionHandler.scala b/core/src/main/scala/com/softwaremill/sttp/HttpURLConnectionHandler.scala
index 45a0448..24c81c7 100644
--- a/core/src/main/scala/com/softwaremill/sttp/HttpURLConnectionHandler.scala
+++ b/core/src/main/scala/com/softwaremill/sttp/HttpURLConnectionHandler.scala
@@ -11,14 +11,18 @@ import java.util.zip.{GZIPInputStream, InflaterInputStream}
 import scala.annotation.tailrec
 import scala.io.Source
 import scala.collection.JavaConverters._
+import scala.concurrent.duration.{Duration, FiniteDuration}
 
-class HttpURLConnectionHandler extends SttpHandler[Id, Nothing] {
+class HttpURLConnectionHandler private (connectionTimeout: FiniteDuration)
+    extends SttpHandler[Id, Nothing] {
   override def send[T](r: Request[T, Nothing]): Response[T] = {
     val c =
       new URL(r.uri.toString).openConnection().asInstanceOf[HttpURLConnection]
     c.setRequestMethod(r.method.m)
     r.headers.foreach { case (k, v) => c.setRequestProperty(k, v) }
     c.setDoInput(true)
+    c.setReadTimeout(timeout(r.readTimeout))
+    c.setConnectTimeout(timeout(connectionTimeout))
 
     // redirects are handled in SttpHandler
     c.setInstanceFollowRedirects(false)
@@ -68,6 +72,10 @@ class HttpURLConnectionHandler extends SttpHandler[Id, Nothing] {
     }
   }
 
+  private def timeout(t: Duration): Int =
+    if (t.isFinite()) t.toMillis.toInt
+    else 0
+
   private def writeBasicBody(body: BasicRequestBody, os: OutputStream): Unit = {
     body match {
       case StringBody(b, encoding, _) =>
@@ -247,3 +255,11 @@ class HttpURLConnectionHandler extends SttpHandler[Id, Nothing] {
 
   override def close(): Unit = {}
 }
+
+object HttpURLConnectionHandler {
+
+  def apply(connectionTimeout: FiniteDuration = SttpHandler.DefaultConnectionTimeout)
+    : SttpHandler[Id, Nothing] =
+    new FollowRedirectsHandler[Id, Nothing](
+      new HttpURLConnectionHandler(connectionTimeout))
+}
diff --git a/core/src/main/scala/com/softwaremill/sttp/RequestT.scala b/core/src/main/scala/com/softwaremill/sttp/RequestT.scala
index 27ed7f3..b9b49f4 100644
--- a/core/src/main/scala/com/softwaremill/sttp/RequestT.scala
+++ b/core/src/main/scala/com/softwaremill/sttp/RequestT.scala
@@ -6,7 +6,7 @@ import java.nio.file.Path
 import java.util.Base64
 
 import scala.collection.immutable.Seq
-
+import scala.concurrent.duration.Duration
 import scala.language.higherKinds
 
 /**
@@ -34,6 +34,7 @@ case class RequestT[U[_], T, +S](
     body: RequestBody[S],
     headers: Seq[(String, String)],
     response: ResponseAs[T, S],
+    readTimeout: Duration,
     options: RequestOptions,
     tags: Map[String, Any]
 ) {
@@ -216,6 +217,9 @@ case class RequestT[U[_], T, +S](
   def streamBody[S2 >: S](b: S2): RequestT[U, T, S2] =
     copy[U, T, S2](body = StreamBody(b))
 
+  def readTimeout(t: Duration): RequestT[U, T, S] =
+    copy(readTimeout = t)
+
   def response[T2, S2 >: S](ra: ResponseAs[T2, S2]): RequestT[U, T2, S2] =
     this.copy(response = ra)
 
diff --git a/core/src/main/scala/com/softwaremill/sttp/SttpHandler.scala b/core/src/main/scala/com/softwaremill/sttp/SttpHandler.scala
index 248356d..b2019dc 100644
--- a/core/src/main/scala/com/softwaremill/sttp/SttpHandler.scala
+++ b/core/src/main/scala/com/softwaremill/sttp/SttpHandler.scala
@@ -1,6 +1,7 @@
 package com.softwaremill.sttp
 
 import scala.language.higherKinds
+import scala.concurrent.duration._
 
 /**
   * @tparam R The type constructor in which responses are wrapped. E.g. `Id`
@@ -19,3 +20,7 @@ trait SttpHandler[R[_], -S] {
     */
   def responseMonad: MonadError[R]
 }
+
+object SttpHandler {
+  private[sttp] val DefaultConnectionTimeout = 30.seconds
+}
\ No newline at end of file
diff --git a/core/src/main/scala/com/softwaremill/sttp/package.scala b/core/src/main/scala/com/softwaremill/sttp/package.scala
index d64acfe..4ec1331 100644
--- a/core/src/main/scala/com/softwaremill/sttp/package.scala
+++ b/core/src/main/scala/com/softwaremill/sttp/package.scala
@@ -7,6 +7,7 @@ import java.nio.file.Path
 import scala.annotation.{implicitNotFound, tailrec}
 import scala.language.higherKinds
 import scala.collection.immutable.Seq
+import scala.concurrent.duration._
 
 package object sttp {
   type Id[X] = X
@@ -26,6 +27,8 @@ package object sttp {
     */
   type BodySerializer[B] = B => BasicRequestBody
 
+  val DefaultReadTimeout: Duration = 1.minute
+
   // constants
 
   private[sttp] val ContentTypeHeader = "Content-Type"
@@ -60,6 +63,7 @@ package object sttp {
                                      NoBody,
                                      Vector(),
                                      asString,
+                                     DefaultReadTimeout,
                                      RequestOptions(followRedirects = true),
                                      Map())
 
@@ -266,9 +270,4 @@ package object sttp {
   implicit class UriContext(val sc: StringContext) extends AnyVal {
     def uri(args: Any*): Uri = UriInterpolator.interpolate(sc, args: _*)
   }
-
-  // default handler
-
-  val HttpURLConnectionHandler: SttpHandler[Id, Nothing] =
-    new FollowRedirectsHandler[Id, Nothing](new HttpURLConnectionHandler())
 }
diff --git a/core/src/test/scala/com/softwaremill/sttp/RequestTests.scala b/core/src/test/scala/com/softwaremill/sttp/RequestTests.scala
index 5773cb1..6332467 100644
--- a/core/src/test/scala/com/softwaremill/sttp/RequestTests.scala
+++ b/core/src/test/scala/com/softwaremill/sttp/RequestTests.scala
@@ -66,4 +66,8 @@ class RequestTests extends FlatSpec with Matchers {
       .find(_._1.equalsIgnoreCase(ContentLengthHeader))
       .map(_._2) should be(Some("10"))
   }
+
+  "request timeout" should "use default if not overridden" in {
+    sttp.readTimeout should be(DefaultReadTimeout)
+  }
 }
diff --git a/okhttp-handler/monix/src/main/scala/com/softwaremill/sttp/okhttp/monix/OkHttpMonixHandler.scala b/okhttp-handler/monix/src/main/scala/com/softwaremill/sttp/okhttp/monix/OkHttpMonixHandler.scala
index 31f4546..608d499 100644
--- a/okhttp-handler/monix/src/main/scala/com/softwaremill/sttp/okhttp/monix/OkHttpMonixHandler.scala
+++ b/okhttp-handler/monix/src/main/scala/com/softwaremill/sttp/okhttp/monix/OkHttpMonixHandler.scala
@@ -1,7 +1,7 @@
 package com.softwaremill.sttp.okhttp.monix
 
 import java.nio.ByteBuffer
-import java.util.concurrent.ArrayBlockingQueue
+import java.util.concurrent.{ArrayBlockingQueue, TimeUnit}
 
 import com.softwaremill.sttp.{SttpHandler, _}
 import com.softwaremill.sttp.okhttp.{OkHttpAsyncHandler, OkHttpHandler}
@@ -14,6 +14,7 @@ import okhttp3.{MediaType, OkHttpClient, RequestBody => OkHttpRequestBody}
 import okio.BufferedSink
 
 import scala.concurrent.Future
+import scala.concurrent.duration.FiniteDuration
 import scala.util.{Failure, Success, Try}
 
 class OkHttpMonixHandler private (client: OkHttpClient, closeClient: Boolean)(
@@ -84,10 +85,17 @@ object OkHttpMonixHandler {
       implicit s: Scheduler): SttpHandler[Task, Observable[ByteBuffer]] =
     new FollowRedirectsHandler(new OkHttpMonixHandler(client, closeClient)(s))
 
-  def apply()(implicit s: Scheduler = Scheduler.Implicits.global)
+  def apply(connectionTimeout: FiniteDuration = SttpHandler.DefaultConnectionTimeout)(
+      implicit s: Scheduler = Scheduler.Implicits.global)
     : SttpHandler[Task, Observable[ByteBuffer]] =
-    OkHttpMonixHandler(OkHttpHandler.buildClientNoRedirects(),
-                       closeClient = true)(s)
+    OkHttpMonixHandler(
+      OkHttpHandler
+        .defaultBuilder()
+        .connectTimeout(connectionTimeout.toMillis, TimeUnit.MILLISECONDS)
+        .readTimeout(SttpHandler.DefaultConnectionTimeout.toMillis, TimeUnit.MILLISECONDS)
+        .build(),
+      closeClient = true
+    )(s)
 
   def usingClient(client: OkHttpClient)(implicit s: Scheduler =
                                           Scheduler.Implicits.global)
diff --git a/okhttp-handler/src/main/scala/com/softwaremill/sttp/okhttp/OkHttpClientHandler.scala b/okhttp-handler/src/main/scala/com/softwaremill/sttp/okhttp/OkHttpClientHandler.scala
index 2250859..d670b62 100644
--- a/okhttp-handler/src/main/scala/com/softwaremill/sttp/okhttp/OkHttpClientHandler.scala
+++ b/okhttp-handler/src/main/scala/com/softwaremill/sttp/okhttp/OkHttpClientHandler.scala
@@ -2,6 +2,7 @@ package com.softwaremill.sttp.okhttp
 
 import java.io.IOException
 import java.nio.charset.Charset
+import java.util.concurrent.TimeUnit
 
 import com.softwaremill.sttp._
 import ResponseAs.EagerResponseHandler
@@ -20,6 +21,7 @@ import okhttp3.{
 import okio.{BufferedSink, Okio}
 
 import scala.collection.JavaConverters._
+import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.{ExecutionContext, Future}
 import scala.language.higherKinds
 import scala.util.{Failure, Try}
@@ -142,18 +144,34 @@ abstract class OkHttpHandler[R[_], S](client: OkHttpClient,
 }
 
 object OkHttpHandler {
-  def buildClientNoRedirects(): OkHttpClient =
+  def defaultBuilder(): OkHttpClient.Builder =
     new OkHttpClient.Builder()
       .followRedirects(false)
       .followSslRedirects(false)
-      .build()
+
+  def updateClientIfCustomReadTimeout[T, S](
+      r: Request[T, S],
+      client: OkHttpClient): OkHttpClient = {
+    if (r.readTimeout == SttpHandler.DefaultConnectionTimeout) client
+    else
+      client
+        .newBuilder()
+        .readTimeout(if (r.readTimeout.isFinite()) r.readTimeout.toMillis
+                     else 0,
+                     TimeUnit.MILLISECONDS)
+        .build()
+
+  }
 }
 
 class OkHttpSyncHandler private (client: OkHttpClient, closeClient: Boolean)
     extends OkHttpHandler[Id, Nothing](client, closeClient) {
   override def send[T](r: Request[T, Nothing]): Response[T] = {
     val request = convertRequest(r)
-    val response = client.newCall(request).execute()
+    val response = OkHttpHandler
+      .updateClientIfCustomReadTimeout(r, client)
+      .newCall(request)
+      .execute()
     readResponse(response, r.response)
   }
 
@@ -166,9 +184,15 @@ object OkHttpSyncHandler {
     new FollowRedirectsHandler[Id, Nothing](
       new OkHttpSyncHandler(client, closeClient))
 
-  def apply(): SttpHandler[Id, Nothing] =
-    OkHttpSyncHandler(OkHttpHandler.buildClientNoRedirects(),
-                      closeClient = true)
+  def apply(
+      connectionTimeout: FiniteDuration = SttpHandler.DefaultConnectionTimeout)
+    : SttpHandler[Id, Nothing] =
+    OkHttpSyncHandler(
+      OkHttpHandler
+        .defaultBuilder()
+        .connectTimeout(connectionTimeout.toMillis, TimeUnit.MILLISECONDS)
+        .build(),
+      closeClient = true)
 
   def usingClient(client: OkHttpClient): SttpHandler[Id, Nothing] =
     OkHttpSyncHandler(client, closeClient = false)
@@ -185,7 +209,8 @@ abstract class OkHttpAsyncHandler[R[_], S](client: OkHttpClient,
       def success(r: R[Response[T]]) = cb(Right(r))
       def error(t: Throwable) = cb(Left(t))
 
-      client
+      OkHttpHandler
+        .updateClientIfCustomReadTimeout(r, client)
         .newCall(request)
         .enqueue(new Callback {
           override def onFailure(call: Call, e: IOException): Unit =
@@ -213,10 +238,17 @@ object OkHttpFutureHandler {
     new FollowRedirectsHandler[Future, Nothing](
       new OkHttpFutureHandler(client, closeClient))
 
-  def apply()(implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
+  def apply(connectionTimeout: FiniteDuration = SttpHandler.DefaultConnectionTimeout)(
+      implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
     : SttpHandler[Future, Nothing] =
-    OkHttpFutureHandler(OkHttpHandler.buildClientNoRedirects(),
-                        closeClient = true)
+    OkHttpFutureHandler(
+      OkHttpHandler
+        .defaultBuilder()
+        .connectTimeout(connectionTimeout.toMillis, TimeUnit.MILLISECONDS)
+        .readTimeout(SttpHandler.DefaultConnectionTimeout.toMillis, TimeUnit.MILLISECONDS)
+        .build(),
+      closeClient = true
+    )
 
   def usingClient(client: OkHttpClient)(implicit ec: ExecutionContext =
                                           ExecutionContext.Implicits.global)
diff --git a/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala b/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala
index aaa6f4c..f544750 100644
--- a/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala
+++ b/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala
@@ -26,6 +26,8 @@ import org.scalatest.concurrent.{IntegrationPatience, ScalaFutures}
 import org.scalatest.{path => _, _}
 
 import scala.concurrent.ExecutionContext.Implicits.global
+import scala.concurrent.Future
+import scala.concurrent.duration._
 import scala.language.higherKinds
 
 class BasicTests
@@ -163,13 +165,19 @@ class BasicTests
         path("loop") {
           redirect("/redirect/loop", StatusCodes.Found)
         }
+    } ~ pathPrefix("timeout") {
+      complete {
+        akka.pattern.after(1.second, using = actorSystem.scheduler)(
+          Future.successful("Done"))
+      }
     }
 
   override def port = 51823
 
   var closeHandlers: List[() => Unit] = Nil
 
-  runTests("HttpURLConnection")(HttpURLConnectionHandler, ForceWrappedValue.id)
+  runTests("HttpURLConnection")(HttpURLConnectionHandler(),
+                                ForceWrappedValue.id)
   runTests("Akka HTTP")(AkkaHttpHandler.usingActorSystem(actorSystem),
                         ForceWrappedValue.future)
   runTests("Async Http Client - Future")(AsyncHttpClientFutureHandler(),
@@ -211,6 +219,7 @@ class BasicTests
     downloadFileTests()
     multipartTests()
     redirectTests()
+    timeoutTests()
 
     def parseResponseTests(): Unit = {
       name should "parse response as string" in {
@@ -633,6 +642,28 @@ class BasicTests
         resp.history should have size (FollowRedirectsHandler.MaxRedirects)
       }
     }
+
+    def timeoutTests(): Unit = {
+      name should "fail if read timeout is not big enough" in {
+        val request = sttp
+          .get(uri"$endpoint/timeout")
+          .readTimeout(200.milliseconds)
+          .response(asString)
+
+        intercept[Throwable] {
+          request.send().force()
+        }
+      }
+
+      name should "not fail if read timeout is big enough" in {
+        val request = sttp
+          .get(uri"$endpoint/timeout")
+          .readTimeout(5.seconds)
+          .response(asString)
+
+        request.send().force().unsafeBody should be("Done")
+      }
+    }
   }
 
   override protected def afterAll(): Unit = {
diff --git a/tests/src/test/scala/com/softwaremill/sttp/IllTypedTests.scala b/tests/src/test/scala/com/softwaremill/sttp/IllTypedTests.scala
index 4e03fc2..53ea798 100644
--- a/tests/src/test/scala/com/softwaremill/sttp/IllTypedTests.scala
+++ b/tests/src/test/scala/com/softwaremill/sttp/IllTypedTests.scala
@@ -25,7 +25,7 @@ class IllTypedTests extends FlatSpec with Matchers {
     val thrown = intercept[ToolBoxError] {
       EvalScala("""
         import com.softwaremill.sttp._
-        implicit val sttpHandler = HttpURLConnectionHandler
+        implicit val sttpHandler = HttpURLConnectionHandler()
         sttp.send()
         """)
     }
diff --git a/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala b/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala
index 6292ecd..7e4d5a9 100644
--- a/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala
+++ b/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala
@@ -17,7 +17,10 @@ import scala.concurrent.duration._
 import scala.language.higherKinds
 import scalaz._
 
-trait TestHttpServer extends BeforeAndAfterAll with ScalaFutures with TestingPatience {
+trait TestHttpServer
+    extends BeforeAndAfterAll
+    with ScalaFutures
+    with TestingPatience {
   this: Suite =>
   protected implicit val actorSystem: ActorSystem = ActorSystem("sttp-test")
   import actorSystem.dispatcher