diff options
17 files changed, 228 insertions, 57 deletions
@@ -17,7 +17,7 @@ val query = "http language:scala" // `sort` is removed, as the value is not defined val request = sttp.get(uri"https://api.github.com/search/repositories?q=$query&sort=$sort") -implicit val handler = HttpURLConnectionHandler +implicit val handler = HttpURLConnectionHandler() val response = request.send() // response.header(...): Option[String] @@ -61,7 +61,7 @@ experimenting with sttp by copy-pasting the following: ```scala import $ivy.`com.softwaremill.sttp::core:0.0.11` import com.softwaremill.sttp._ -implicit val handler = HttpURLConnectionHandler +implicit val handler = HttpURLConnectionHandler() sttp.get(uri"http://httpbin.org/ip").send() ``` @@ -113,7 +113,7 @@ value of type `SttpHandler` needs to be in scope to invoke the `send()` on the request: ```scala -implicit val handler = HttpURLConnectionHandler +implicit val handler = HttpURLConnectionHandler() val response: Response[String] = request.send() ``` @@ -214,7 +214,7 @@ in the identity type constructor, which is equivalent to no wrapper at all. To use, add an implicit value: ```scala -implicit val sttpHandler = HttpURLConnectionHandler +implicit val sttpHandler = HttpURLConnectionHandler() ``` ### `AkkaHttpHandler` @@ -417,7 +417,7 @@ a request to decode the response to a specific object. import com.softwaremill.sttp._ import com.softwaremill.sttp.circe._ -implicit val handler = HttpURLConnectionHandler +implicit val handler = HttpURLConnectionHandler() // Assume that there is an implicit circe encoder in scope // for the request Payload, and a decoder for the Response @@ -461,6 +461,26 @@ There are two type aliases for the request template that are used: * `type Request[T, S] = RequestT[Id, T, S]`. A sendable request. * `type PartialRequest[T, S] = RequestT[Empty, T, S]` +## Timeouts + +Sttp supports read and connection timeouts: + * Connection timeout - can be set globally (30 seconds by default) + * Read timeout - can be set per request (1 minute by default) + +How to use: +```scala +import com.softwaremill.sttp._ +import scala.concurrent.duration._ + +// all backends provide a constructor that allows users to specify connection timeout +implicit val handler = HttpURLConnectionHandler(connectionTimeout = 1.minute) + +sttp + .get(uri"...") + .readTimeout(5.minutes) // or Duration.Inf to turn read timeout off + .send() +``` + ## Notes * the encoding for `String`s defaults to `utf-8`. @@ -493,3 +513,4 @@ and pick a task you'd like to work on! * [Omar Alejandro Mainegra Sarduy](https://github.com/omainegra) * [Bjørn Madsen](https://github.com/aeons) * [Piotr Buda](https://github.com/pbuda) +* [Piotr Gabara](https://github.com/bhop) 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..c6b0a2c 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 @@ -33,12 +37,22 @@ class AkkaHttpHandler private (actorSystem: ActorSystem, private implicit val as: ActorSystem = actorSystem private implicit val materializer: ActorMaterializer = ActorMaterializer() + private val connectionSettings = ClientConnectionSettings(actorSystem) + .withConnectingTimeout(connectionTimeout) + + private val connectionPoolSettings = ConnectionPoolSettings(actorSystem) + override def send[T](r: Request[T, S]): Future[Response[T]] = { implicit val ec: ExecutionContext = this.ec + + val settings = connectionPoolSettings + .withConnectionSettings( + connectionSettings.withIdleTimeout(r.options.readTimeout)) + requestToAkka(r) .flatMap(setBodyOnAkka(r, r.body, _)) .toFuture - .flatMap(Http().singleRequest(_)) + .flatMap(req => Http().singleRequest(req, settings = settings)) .flatMap { hr => val code = hr.status.intValue() @@ -271,19 +285,28 @@ object AkkaHttpHandler { private def apply(actorSystem: ActorSystem, ec: ExecutionContext, - terminateActorSystemOnClose: Boolean) + 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 +315,13 @@ 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..9948d42 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,11 @@ 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( + AsyncHttpClientHandler.defaultClient(connectionTimeout.toMillis.toInt), + 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..a33932b 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,12 @@ 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]( + AsyncHttpClientHandler.defaultClient(connectionTimeout.toMillis.toInt), + 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..6e0a22c 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,12 @@ 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( + AsyncHttpClientHandler.defaultClient(connectionTimeout.toMillis.toInt), + 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..7cfcb43 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,12 @@ 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( + AsyncHttpClientHandler.defaultClient(connectionTimeout.toMillis.toInt), + 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..0e80a29 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,16 @@ 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( + AsyncHttpClientHandler.defaultClient(connectionTimeout.toMillis.toInt), + 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..984ecf6 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,8 @@ import org.asynchttpclient.{ AsyncCompletionHandler, AsyncHandler, AsyncHttpClient, + DefaultAsyncHttpClientConfig, + DefaultAsyncHttpClient, HttpResponseBodyPart, HttpResponseHeaders, HttpResponseStatus, @@ -155,7 +157,11 @@ 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 readTimeout = r.options.readTimeout + val rb = new RequestBuilder(r.method.m) + .setUrl(r.uri.toString) + .setRequestTimeout( + if (readTimeout.isFinite()) readTimeout.toMillis.toInt else -1) r.headers.foreach { case (k, v) => rb.setHeader(k, v) } setBody(r, r.body, rb) rb.build() @@ -289,6 +295,16 @@ abstract class AsyncHttpClientHandler[R[_], S](asyncHttpClient: AsyncHttpClient, } } +object AsyncHttpClientHandler { + + private[asynchttpclient] def defaultClient(connectionTimeout: Int): AsyncHttpClient = + new DefaultAsyncHttpClient( + new DefaultAsyncHttpClientConfig.Builder() + .setConnectTimeout(connectionTimeout) + .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..9b73298 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.options.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 020e926..6e76f6f 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 /** @@ -216,6 +216,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] = + this.copy(options = options.copy(readTimeout = t)) + def response[T2, S2 >: S](ra: ResponseAs[T2, S2]): RequestT[U, T2, S2] = this.copy(response = ra) @@ -281,4 +284,4 @@ class SpecifyAuthScheme[U[_], T, +S](hn: String, rt: RequestT[U, T, S]) { rt.header(hn, s"Bearer $token") } -case class RequestOptions(followRedirects: Boolean) +case class RequestOptions(followRedirects: Boolean, readTimeout: Duration) 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..a9950be 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" @@ -55,13 +58,14 @@ package object sttp { * An empty request with no headers. */ val emptyRequest: RequestT[Empty, String, Nothing] = - RequestT[Empty, String, Nothing](None, - None, - NoBody, - Vector(), - asString, - RequestOptions(followRedirects = true), - Map()) + RequestT[Empty, String, Nothing]( + None, + None, + NoBody, + Vector(), + asString, + RequestOptions(followRedirects = true, readTimeout = DefaultReadTimeout), + Map()) /** * A starting request, with the following modifications comparing to @@ -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..e62112a 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.options.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..8d20bad 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 @@ -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,12 @@ 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.defaultClient(DefaultReadTimeout.toMillis, connectionTimeout.toMillis), + 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..3e57930 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,38 @@ abstract class OkHttpHandler[R[_], S](client: OkHttpClient, } object OkHttpHandler { - def buildClientNoRedirects(): OkHttpClient = + + private[okhttp] def defaultClient(readTimeout: Long, + connectionTimeout: Long): OkHttpClient = new OkHttpClient.Builder() .followRedirects(false) .followSslRedirects(false) + .connectTimeout(connectionTimeout, TimeUnit.MILLISECONDS) + .readTimeout(readTimeout, TimeUnit.MILLISECONDS) .build() + + private[okhttp] def updateClientIfCustomReadTimeout[T, S](r: Request[T, S], + client: OkHttpClient): OkHttpClient = { + val readTimeout = r.options.readTimeout + if (readTimeout == DefaultReadTimeout) client + else + client + .newBuilder() + .readTimeout(if (readTimeout.isFinite()) 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 +188,12 @@ 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.defaultClient(DefaultReadTimeout.toMillis, connectionTimeout.toMillis), + closeClient = true) def usingClient(client: OkHttpClient): SttpHandler[Id, Nothing] = OkHttpSyncHandler(client, closeClient = false) @@ -185,7 +210,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 +239,12 @@ 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.defaultClient(DefaultReadTimeout.toMillis, connectionTimeout.toMillis), + 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() """) } |