Response.body is now an Either[String, T], to handle cases when the status code isn't 2xx

6 files changed, 55 insertions, 7 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 f6eaf10..fec0eab 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
@@ -28,6 +28,10 @@ class AsyncHttpClientCatsHandler[F[_]: Async] private (
   override protected def publisherToStreamBody(
       p: Publisher[ByteBuffer]): Nothing =
     throw new IllegalStateException("This handler does not support streaming")
+
+  override protected def publisherToString(
+      p: Publisher[ByteBuffer]): F[String] =
+    throw new IllegalStateException("This handler does not support streaming")
 }
 
 object AsyncHttpClientCatsHandler {
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 f7ed00c..dc30925 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
@@ -4,7 +4,7 @@ import java.nio.ByteBuffer
 
 import cats.effect._
 import com.softwaremill.sttp.asynchttpclient.AsyncHttpClientHandler
-import com.softwaremill.sttp.{MonadAsyncError, SttpHandler}
+import com.softwaremill.sttp.{MonadAsyncError, SttpHandler, Utf8, concatByteBuffers}
 import fs2._
 import fs2.interop.reactivestreams._
 import org.asynchttpclient.{
@@ -33,6 +33,15 @@ class AsyncHttpClientFs2Handler[F[_]: Effect] private (
   override protected def publisherToStreamBody(
       p: Publisher[ByteBuffer]): Stream[F, ByteBuffer] =
     p.toStream[F]
+
+  override protected def publisherToString(
+      p: Publisher[ByteBuffer]): F[String] = {
+    val bytes = p
+      .toStream[F]
+      .runFold(ByteBuffer.allocate(0))(concatByteBuffers)
+
+    implicitly[Effect[F]].map(bytes)(bb => new String(bb.array(), Utf8))
+  }
 }
 
 object AsyncHttpClientFs2Handler {
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 d0bc03e..81dee9e 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
@@ -26,6 +26,10 @@ class AsyncHttpClientFutureHandler private (
   override protected def publisherToStreamBody(
       p: Publisher[ByteBuffer]): Nothing =
     throw new IllegalStateException("This handler does not support streaming")
+
+  override protected def publisherToString(
+      p: Publisher[ByteBuffer]): Future[String] =
+    throw new IllegalStateException("This handler does not support streaming")
 }
 
 object AsyncHttpClientFutureHandler {
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 7f01f0c..1ef3973 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
@@ -2,7 +2,7 @@ package com.softwaremill.sttp.asynchttpclient.monix
 
 import java.nio.ByteBuffer
 
-import com.softwaremill.sttp.{MonadAsyncError, SttpHandler}
+import com.softwaremill.sttp.{MonadAsyncError, SttpHandler, Utf8, concatByteBuffers}
 import com.softwaremill.sttp.asynchttpclient.AsyncHttpClientHandler
 import monix.eval.Task
 import monix.execution.{Cancelable, Scheduler}
@@ -31,6 +31,16 @@ class AsyncHttpClientMonixHandler private (
   override protected def publisherToStreamBody(
       p: Publisher[ByteBuffer]): Observable[ByteBuffer] =
     Observable.fromReactivePublisher(p)
+
+  override protected def publisherToString(
+      p: Publisher[ByteBuffer]): Task[String] = {
+
+    val bytes = Observable
+      .fromReactivePublisher(p)
+      .foldLeftL(ByteBuffer.allocate(0))(concatByteBuffers)
+
+    bytes.map(bb => new String(bb.array(), Utf8))
+  }
 }
 
 object AsyncHttpClientMonixHandler {
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 badccb3..4bc169d 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
@@ -26,6 +26,10 @@ class AsyncHttpClientScalazHandler private (asyncHttpClient: AsyncHttpClient,
   override protected def publisherToStreamBody(
       p: Publisher[ByteBuffer]): Nothing =
     throw new IllegalStateException("This handler does not support streaming")
+
+  override protected def publisherToString(
+      p: Publisher[ByteBuffer]): Task[String] =
+    throw new IllegalStateException("This handler does not support streaming")
 }
 
 object AsyncHttpClientScalazHandler {
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 2e5d16b..f8122cc 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
@@ -62,6 +62,8 @@ abstract class AsyncHttpClientHandler[R[_], S](asyncHttpClient: AsyncHttpClient,
 
   protected def publisherToStreamBody(p: Publisher[ByteBuffer]): S
 
+  protected def publisherToString(p: Publisher[ByteBuffer]): R[String]
+
   private def eagerAsyncHandler[T](
       responseAs: ResponseAs[T, S],
       success: R[Response[T]] => Unit,
@@ -134,8 +136,15 @@ abstract class AsyncHttpClientHandler[R[_], S](asyncHttpClient: AsyncHttpClient,
           val baseResponse = readResponseNoBody(builder.build())
           val p = publisher.getOrElse(EmptyPublisher)
           val s = publisherToStreamBody(p)
-          val t = responseAs.responseIsStream(s)
-          success(rm.unit(baseResponse.copy(body = t)))
+          val b = if (codeIsSuccess(baseResponse.code)) {
+            rm.unit(Right(responseAs.responseIsStream(s)))
+          } else {
+            rm.map(publisherToString(p), Left(_: String))
+          }
+
+          success(rm.map(b, { bb: Either[String, T] =>
+            baseResponse.copy(body = bb)
+          }))
         }
       }
 
@@ -223,12 +232,20 @@ abstract class AsyncHttpClientHandler[R[_], S](asyncHttpClient: AsyncHttpClient,
   private def readEagerResponse[T](
       response: AsyncResponse,
       responseAs: ResponseAs[T, S]): R[Response[T]] = {
-    val body = eagerResponseHandler(response).handle(responseAs, rm)
-    rm.map(body, (b: T) => readResponseNoBody(response).copy(body = b))
+    val base = readResponseNoBody(response)
+
+    val body = if (codeIsSuccess(base.code)) {
+      rm.map(eagerResponseHandler(response).handle(responseAs, rm), Right(_: T))
+    } else {
+      rm.map(eagerResponseHandler(response).handle(asString, rm),
+             Left(_: String))
+    }
+
+    rm.map(body, (b: Either[String, T]) => base.copy(body = b))
   }
 
   private def readResponseNoBody(response: AsyncResponse): Response[Unit] = {
-    Response((),
+    Response(Right(()),
              response.getStatusCode,
              response.getHeaders
                .iterator()