Merge pull request #27 from aeons/feature/fs2

Add fs2 streaming module and refactor streaming tests

11 files changed, 315 insertions, 108 deletions

diff --git a/async-http-client-handler/fs2/src/main/scala/com/softwaremill/sttp/asynchttpclient/fs2/Fs2AsyncHttpClientHandler.scala b/async-http-client-handler/fs2/src/main/scala/com/softwaremill/sttp/asynchttpclient/fs2/Fs2AsyncHttpClientHandler.scala
new file mode 100644
index 0000000..91a3109
--- /dev/null
+++ b/async-http-client-handler/fs2/src/main/scala/com/softwaremill/sttp/asynchttpclient/fs2/Fs2AsyncHttpClientHandler.scala
@@ -0,0 +1,88 @@
+package com.softwaremill.sttp.asynchttpclient.fs2
+
+import java.nio.ByteBuffer
+
+import cats.effect._
+import com.softwaremill.sttp.asynchttpclient.AsyncHttpClientHandler
+import com.softwaremill.sttp.{MonadAsyncError, SttpHandler}
+import fs2._
+import fs2.interop.reactivestreams._
+import org.asynchttpclient.{
+  AsyncHttpClient,
+  AsyncHttpClientConfig,
+  DefaultAsyncHttpClient
+}
+import org.reactivestreams.Publisher
+
+import scala.concurrent.ExecutionContext
+import scala.language.higherKinds
+
+class Fs2AsyncHttpClientHandler[F[_]: Effect] private (
+    asyncHttpClient: AsyncHttpClient,
+    closeClient: Boolean)(implicit ec: ExecutionContext)
+    extends AsyncHttpClientHandler[F, Stream[F, ByteBuffer]](
+      asyncHttpClient,
+      new EffectMonad,
+      closeClient
+    ) {
+
+  override protected def streamBodyToPublisher(
+      s: Stream[F, ByteBuffer]): Publisher[ByteBuffer] =
+    s.toUnicastPublisher
+
+  override protected def publisherToStreamBody(
+      p: Publisher[ByteBuffer]): Stream[F, ByteBuffer] =
+    p.toStream[F]
+}
+
+object Fs2AsyncHttpClientHandler {
+
+  /**
+    * @param ec The execution context for running non-network related operations,
+    *           e.g. mapping responses. Defaults to the global execution
+    *           context.
+    */
+  def apply[F[_]: Effect]()(
+      implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
+    : SttpHandler[F, Stream[F, ByteBuffer]] =
+    new Fs2AsyncHttpClientHandler[F](new DefaultAsyncHttpClient(),
+                                     closeClient = true)
+
+  /**
+    * @param ec The execution context for running non-network related operations,
+    *           e.g. mapping responses. Defaults to the global execution
+    *           context.
+    */
+  def usingConfig[F[_]: Effect](cfg: AsyncHttpClientConfig)(
+      implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
+    : SttpHandler[F, Stream[F, ByteBuffer]] =
+    new Fs2AsyncHttpClientHandler[F](new DefaultAsyncHttpClient(cfg),
+                                     closeClient = true)
+
+  /**
+    * @param ec The execution context for running non-network related operations,
+    *           e.g. mapping responses. Defaults to the global execution
+    *           context.
+    */
+  def usingClient[F[_]: Effect](client: AsyncHttpClient)(
+      implicit ec: ExecutionContext = ExecutionContext.Implicits.global)
+    : SttpHandler[F, Stream[F, ByteBuffer]] =
+    new Fs2AsyncHttpClientHandler[F](client, closeClient = false)
+}
+
+private[fs2] class EffectMonad[F[_]](implicit F: Effect[F])
+    extends MonadAsyncError[F] {
+
+  override def async[T](
+      register: ((Either[Throwable, T]) => Unit) => Unit): F[T] =
+    F.async(register)
+
+  override def unit[T](t: T): F[T] = F.pure(t)
+
+  override def map[T, T2](fa: F[T], f: (T) => T2): F[T2] = F.map(fa)(f)
+
+  override def flatMap[T, T2](fa: F[T], f: (T) => F[T2]): F[T2] =
+    F.flatMap(fa)(f)
+
+  override def error[T](t: Throwable): F[T] = F.raiseError(t)
+}
diff --git a/build.sbt b/build.sbt
index a1896fa..3e2f310 100644
--- a/build.sbt
+++ b/build.sbt
@@ -51,6 +51,7 @@ lazy val rootProject = (project in file("."))
     scalazAsyncHttpClientHandler,
     monixAsyncHttpClientHandler,
     catsAsyncHttpClientHandler,
+    fs2AsyncHttpClientHandler,
     okhttpClientHandler,
     okhttpMonixClientHandler,
     circe,
@@ -121,6 +122,16 @@ lazy val catsAsyncHttpClientHandler: Project = (project in file(
     )
   ) dependsOn asyncHttpClientHandler
 
+lazy val fs2AsyncHttpClientHandler: Project = (project in file(
+  "async-http-client-handler/fs2"))
+  .settings(commonSettings: _*)
+  .settings(
+    name := "async-http-client-handler-fs2",
+    libraryDependencies ++= Seq(
+      "com.github.zainab-ali" %% "fs2-reactive-streams" % "0.2.2"
+    )
+  ) dependsOn asyncHttpClientHandler
+
 lazy val okhttpClientHandler: Project = (project in file(
   "okhttp-client-handler"))
   .settings(commonSettings: _*)
@@ -163,5 +174,8 @@ lazy val tests: Project = (project in file("tests"))
       "ch.qos.logback" % "logback-classic" % "1.2.3"
     ).map(_ % "test"),
     libraryDependencies += "org.scala-lang" % "scala-compiler" % scalaVersion.value % "test"
-  ) dependsOn (core, akkaHttpHandler, futureAsyncHttpClientHandler, scalazAsyncHttpClientHandler,
-monixAsyncHttpClientHandler, catsAsyncHttpClientHandler, okhttpClientHandler, okhttpMonixClientHandler)
+  ) dependsOn (
+  core, akkaHttpHandler, futureAsyncHttpClientHandler, scalazAsyncHttpClientHandler,
+  monixAsyncHttpClientHandler, catsAsyncHttpClientHandler, fs2AsyncHttpClientHandler,
+  okhttpClientHandler, okhttpMonixClientHandler
+)
diff --git a/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala b/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala
index aeee5ed..1594ef3 100644
--- a/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala
+++ b/tests/src/test/scala/com/softwaremill/sttp/BasicTests.scala
@@ -6,18 +6,15 @@ import java.nio.file.Paths
 import java.time.{ZoneId, ZonedDateTime}
 
 import akka.http.scaladsl.coding.{Deflate, Gzip, NoCoding}
-import akka.http.scaladsl.model.{DateTime, FormData}
-import akka.http.scaladsl.model.headers._
 import akka.http.scaladsl.model.headers.CacheDirectives._
+import akka.http.scaladsl.model.headers._
+import akka.http.scaladsl.model.{DateTime, FormData}
 import akka.http.scaladsl.server.Directives._
 import akka.http.scaladsl.server.Route
 import akka.http.scaladsl.server.directives.Credentials
 import akka.util.ByteString
-import com.softwaremill.sttp.akkahttp.AkkaHttpSttpHandler
-import com.typesafe.scalalogging.StrictLogging
-import org.scalatest.concurrent.{IntegrationPatience, ScalaFutures}
-import org.scalatest.{path => _, _}
 import better.files._
+import com.softwaremill.sttp.akkahttp.AkkaHttpSttpHandler
 import com.softwaremill.sttp.asynchttpclient.cats.CatsAsyncHttpClientHandler
 import com.softwaremill.sttp.asynchttpclient.future.FutureAsyncHttpClientHandler
 import com.softwaremill.sttp.asynchttpclient.monix.MonixAsyncHttpClientHandler
@@ -27,8 +24,11 @@ import com.softwaremill.sttp.okhttp.{
   OkHttpFutureClientHandler,
   OkHttpSyncClientHandler
 }
-import scala.concurrent.ExecutionContext.Implicits.global
+import com.typesafe.scalalogging.StrictLogging
+import org.scalatest.concurrent.{IntegrationPatience, ScalaFutures}
+import org.scalatest.{path => _, _}
 
+import scala.concurrent.ExecutionContext.Implicits.global
 import scala.language.higherKinds
 
 class BasicTests
diff --git a/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala b/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala
index 3238c3c..d3c7b89 100644
--- a/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala
+++ b/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala
@@ -1,18 +1,9 @@
 package com.softwaremill.sttp
 
-import java.nio.ByteBuffer
-
 import akka.http.scaladsl.server.Directives._
 import akka.http.scaladsl.server.Route
-import akka.stream.scaladsl.Source
-import akka.util.ByteString
-import com.softwaremill.sttp.akkahttp.AkkaHttpSttpHandler
-import com.softwaremill.sttp.asynchttpclient.monix.MonixAsyncHttpClientHandler
-import com.softwaremill.sttp.okhttp.monix.OkHttpMonixClientHandler
+import com.softwaremill.sttp.streaming._
 import com.typesafe.scalalogging.StrictLogging
-import monix.execution.Scheduler.Implicits.global
-import monix.reactive.Observable
-import org.scalatest.concurrent.{IntegrationPatience, ScalaFutures}
 import org.scalatest.{BeforeAndAfterAll, FlatSpec, Matchers}
 
 import scala.language.higherKinds
@@ -21,16 +12,14 @@ class StreamingTests
     extends FlatSpec
     with Matchers
     with BeforeAndAfterAll
-    with ScalaFutures
     with StrictLogging
-    with IntegrationPatience
-    with ForceWrapped
-    with TestHttpServer {
+    with TestHttpServer
+    with ForceWrapped {
 
   override val serverRoutes: Route =
     path("echo") {
       post {
-        parameterMap { params =>
+        parameterMap { _ =>
           entity(as[String]) { body: String =>
             complete(body)
           }
@@ -38,50 +27,24 @@ class StreamingTests
       }
     }
 
-  type BodyProducer[S] = String => S
-  type BodyConsumer[S] = S => String
-
   override def port = 51824
+
   val body = "streaming test"
 
-  val akkaHandler = AkkaHttpSttpHandler.usingActorSystem(actorSystem)
-  val monixAsyncHttpClient = MonixAsyncHttpClientHandler()
-  val monixOkHttpClient = OkHttpMonixClientHandler()
-
-  val akkaHttpBodyProducer: BodyProducer[Source[ByteString, Any]] = s =>
-    Source.single(ByteString(s))
-  val akkaHttpBodyConsumer: BodyConsumer[Source[ByteString, Any]] =
-    _.runReduce(_ ++ _).futureValue.utf8String
-
-  val monixBodyProducer: BodyProducer[Observable[ByteBuffer]] =
-    s =>
-      Observable.fromIterable(
-        s.getBytes("utf-8").map(b => ByteBuffer.wrap(Array(b))))
-
-  val monixBodyConsumer: BodyConsumer[Observable[ByteBuffer]] = stream =>
-    new String(stream
-                 .flatMap(bb => Observable.fromIterable(bb.array()))
-                 .toListL
-                 .runAsync
-                 .futureValue
-                 .toArray,
-               "utf-8")
-
-  runTests("Akka HTTP", akkaHttpBodyProducer, akkaHttpBodyConsumer)(
-    akkaHandler,
-    ForceWrappedValue.future)
-  runTests("Monix Async Http Client", monixBodyProducer, monixBodyConsumer)(
-    monixAsyncHttpClient,
-    ForceWrappedValue.monixTask)
-  runTests("Monix OkHttp Client", monixBodyProducer, monixBodyConsumer)(
-    monixOkHttpClient,
-    ForceWrappedValue.monixTask)
-
-  def runTests[R[_], S](name: String,
-                        bodyProducer: BodyProducer[S],
-                        bodyConsumer: BodyConsumer[S])(
-      implicit handler: SttpHandler[R, S],
-      forceResponse: ForceWrappedValue[R]): Unit = {
+  var closeHandlers: List[() => Unit] = Nil
+
+  runTests("Akka Http", new AkkaStreamingTests(actorSystem))
+  runTests("Monix Async Http Client", new MonixAHCStreamingTests)
+  runTests("Monix OkHttp", new MonixOKHStreamingTests)
+  runTests("fs2 Async Http Client", new Fs2StreamingTests)
+
+  def runTests[R[_], S](
+      name: String,
+      testStreamingHandler: TestStreamingHandler[R, S]): Unit = {
+    import testStreamingHandler._
+
+    closeHandlers = handler.close _ :: closeHandlers
+
     name should "stream request body" in {
       val response = sttp
         .post(uri"$endpoint/echo")
@@ -89,7 +52,7 @@ class StreamingTests
         .send()
         .force()
 
-      response.body should be(body)
+      response.body shouldBe body
     }
 
     it should "receive a stream" in {
@@ -100,7 +63,7 @@ class StreamingTests
         .send()
         .force()
 
-      bodyConsumer(response.body) should be(body)
+      bodyConsumer(response.body).force() shouldBe body
     }
 
     it should "receive a stream from an https site" in {
@@ -113,13 +76,13 @@ class StreamingTests
         .send()
         .force()
 
-      bodyConsumer(response.body) should include("</div>")
+      bodyConsumer(response.body).force() should include("</div>")
     }
   }
 
   override protected def afterAll(): Unit = {
-    akkaHandler.close()
-    monixAsyncHttpClient.close()
+    closeHandlers.foreach(_())
     super.afterAll()
   }
+
 }
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/AkkaStreamingTests.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/AkkaStreamingTests.scala
new file mode 100644
index 0000000..53fe63e
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/AkkaStreamingTests.scala
@@ -0,0 +1,29 @@
+package com.softwaremill.sttp.streaming
+
+import akka.NotUsed
+import akka.actor.ActorSystem
+import akka.stream.Materializer
+import akka.stream.scaladsl.Source
+import akka.util.ByteString
+import com.softwaremill.sttp.{ForceWrappedValue, SttpHandler}
+import com.softwaremill.sttp.akkahttp.AkkaHttpSttpHandler
+
+import scala.concurrent.Future
+
+class AkkaStreamingTests(actorSystem: ActorSystem)(
+    implicit materializer: Materializer)
+    extends TestStreamingHandler[Future, Source[ByteString, Any]] {
+
+  override implicit val handler: SttpHandler[Future, Source[ByteString, Any]] =
+    AkkaHttpSttpHandler.usingActorSystem(actorSystem)
+
+  override implicit val forceResponse: ForceWrappedValue[Future] =
+    ForceWrappedValue.future
+
+  override def bodyProducer(body: String): Source[ByteString, NotUsed] =
+    Source.single(ByteString(body))
+
+  override def bodyConsumer(stream: Source[ByteString, Any]): Future[String] =
+    stream.map(_.utf8String).runReduce(_ + _)
+
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/Fs2StreamingTests.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/Fs2StreamingTests.scala
new file mode 100644
index 0000000..b9d249b
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/Fs2StreamingTests.scala
@@ -0,0 +1,29 @@
+package com.softwaremill.sttp.streaming
+
+import java.nio.ByteBuffer
+
+import cats.effect._
+import cats.implicits._
+import com.softwaremill.sttp.asynchttpclient.fs2.Fs2AsyncHttpClientHandler
+import com.softwaremill.sttp.{ForceWrappedValue, SttpHandler}
+import fs2._
+
+class Fs2StreamingTests
+    extends TestStreamingHandler[IO, Stream[IO, ByteBuffer]] {
+
+  override implicit val handler: SttpHandler[IO, Stream[IO, ByteBuffer]] =
+    Fs2AsyncHttpClientHandler[IO]()
+
+  override implicit val forceResponse: ForceWrappedValue[IO] =
+    ForceWrappedValue.catsIo
+
+  override def bodyProducer(body: String): Stream[IO, ByteBuffer] =
+    Stream.emits(body.getBytes("utf-8").map(b => ByteBuffer.wrap(Array(b))))
+
+  override def bodyConsumer(stream: Stream[IO, ByteBuffer]): IO[String] =
+    stream
+      .map(bb => Chunk.array(bb.array))
+      .through(text.utf8DecodeC)
+      .runFoldMonoid
+
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixAHCStreamingTests.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixAHCStreamingTests.scala
new file mode 100644
index 0000000..4a4ff96
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixAHCStreamingTests.scala
@@ -0,0 +1,17 @@
+package com.softwaremill.sttp.streaming
+
+import java.nio.ByteBuffer
+
+import com.softwaremill.sttp.SttpHandler
+import com.softwaremill.sttp.asynchttpclient.monix.MonixAsyncHttpClientHandler
+import monix.eval.Task
+import monix.reactive.Observable
+
+class MonixAHCStreamingTests extends MonixBaseHandler {
+
+  import monix.execution.Scheduler.Implicits.global
+
+  override implicit val handler: SttpHandler[Task, Observable[ByteBuffer]] =
+    MonixAsyncHttpClientHandler()
+
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixBaseHandler.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixBaseHandler.scala
new file mode 100644
index 0000000..b34a4ae
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixBaseHandler.scala
@@ -0,0 +1,25 @@
+package com.softwaremill.sttp.streaming
+
+import java.nio.ByteBuffer
+
+import com.softwaremill.sttp.ForceWrappedValue
+import monix.eval.Task
+import monix.reactive.Observable
+
+trait MonixBaseHandler
+    extends TestStreamingHandler[Task, Observable[ByteBuffer]] {
+
+  override implicit def forceResponse: ForceWrappedValue[Task] =
+    ForceWrappedValue.monixTask
+
+  override def bodyProducer(body: String): Observable[ByteBuffer] =
+    Observable.fromIterable(
+      body.getBytes("utf-8").map(b => ByteBuffer.wrap(Array(b))))
+
+  override def bodyConsumer(stream: Observable[ByteBuffer]): Task[String] =
+    stream
+      .flatMap(bb => Observable.fromIterable(bb.array()))
+      .toListL
+      .map(bs => new String(bs.toArray, "utf8"))
+
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixOKHStreamingTests.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixOKHStreamingTests.scala
new file mode 100644
index 0000000..04666be
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/MonixOKHStreamingTests.scala
@@ -0,0 +1,17 @@
+package com.softwaremill.sttp.streaming
+
+import java.nio.ByteBuffer
+
+import com.softwaremill.sttp.SttpHandler
+import com.softwaremill.sttp.okhttp.monix.OkHttpMonixClientHandler
+import monix.eval.Task
+import monix.reactive.Observable
+
+class MonixOKHStreamingTests extends MonixBaseHandler {
+
+  import monix.execution.Scheduler.Implicits.global
+
+  override implicit val handler: SttpHandler[Task, Observable[ByteBuffer]] =
+    OkHttpMonixClientHandler()
+
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/streaming/TestStreamingHandler.scala b/tests/src/test/scala/com/softwaremill/sttp/streaming/TestStreamingHandler.scala
new file mode 100644
index 0000000..b786e49
--- /dev/null
+++ b/tests/src/test/scala/com/softwaremill/sttp/streaming/TestStreamingHandler.scala
@@ -0,0 +1,15 @@
+package com.softwaremill.sttp.streaming
+
+import com.softwaremill.sttp.{ForceWrappedValue, SttpHandler}
+
+import scala.language.higherKinds
+
+trait TestStreamingHandler[R[_], S] {
+  implicit def handler: SttpHandler[R, S]
+
+  implicit def forceResponse: ForceWrappedValue[R]
+
+  def bodyProducer(body: String): S
+
+  def bodyConsumer(stream: S): R[String]
+}
diff --git a/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala b/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala
index 59e1612..407aaca 100644
--- a/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala
+++ b/tests/src/test/scala/com/softwaremill/sttp/testHelpers.scala
@@ -7,12 +7,13 @@ import akka.actor.ActorSystem
 import akka.http.scaladsl.Http
 import akka.http.scaladsl.server.Route
 import akka.stream.ActorMaterializer
-import org.scalatest.{BeforeAndAfterAll, Suite}
 import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.exceptions.TestFailedException
 import org.scalatest.matchers.{MatchResult, Matcher}
+import org.scalatest.{BeforeAndAfterAll, Suite}
 
 import scala.concurrent.Future
+import scala.concurrent.duration._
 import scala.language.higherKinds
 import scalaz._
 
@@ -36,45 +37,54 @@ trait TestHttpServer extends BeforeAndAfterAll with ScalaFutures {
   def port: Int
 }
 
-trait ForceWrapped extends ScalaFutures { this: Suite =>
-  trait ForceWrappedValue[R[_]] {
-    def force[T](wrapped: R[T]): T
+trait ForceWrappedValue[R[_]] {
+  def force[T](wrapped: R[T]): T
+}
+
+object ForceWrappedValue extends ScalaFutures {
+  override implicit val patienceConfig: PatienceConfig =
+    PatienceConfig(timeout = 5.seconds, interval = 150.milliseconds)
+
+  val id = new ForceWrappedValue[Id] {
+    override def force[T](wrapped: Id[T]): T =
+      wrapped
   }
-  object ForceWrappedValue {
-    val id = new ForceWrappedValue[Id] {
-      override def force[T](wrapped: Id[T]): T =
-        wrapped
-    }
-    val future = new ForceWrappedValue[Future] {
-      override def force[T](wrapped: Future[T]): T =
-        try {
-          wrapped.futureValue
-        } catch {
-          case e: TestFailedException if e.getCause != null => throw e.getCause
-        }
-    }
-    val scalazTask = new ForceWrappedValue[scalaz.concurrent.Task] {
-      override def force[T](wrapped: scalaz.concurrent.Task[T]): T =
-        wrapped.unsafePerformSyncAttempt match {
-          case -\/(error) => throw error
-          case \/-(value) => value
-        }
-    }
-    val monixTask = new ForceWrappedValue[monix.eval.Task] {
-      import monix.execution.Scheduler.Implicits.global
-
-      override def force[T](wrapped: monix.eval.Task[T]): T =
-        try {
-          wrapped.runAsync.futureValue
-        } catch {
-          case e: TestFailedException => throw e.getCause
-        }
-    }
-    val catsIo = new ForceWrappedValue[cats.effect.IO] {
-      override def force[T](wrapped: cats.effect.IO[T]): T =
-        wrapped.unsafeRunSync
-    }
+  val future = new ForceWrappedValue[Future] {
+
+    override def force[T](wrapped: Future[T]): T =
+      try {
+        wrapped.futureValue
+      } catch {
+        case e: TestFailedException if e.getCause != null => throw e.getCause
+      }
+  }
+  val scalazTask = new ForceWrappedValue[scalaz.concurrent.Task] {
+    override def force[T](wrapped: scalaz.concurrent.Task[T]): T =
+      wrapped.unsafePerformSyncAttempt match {
+        case -\/(error) => throw error
+        case \/-(value) => value
+      }
+  }
+  val monixTask = new ForceWrappedValue[monix.eval.Task] {
+    import monix.execution.Scheduler.Implicits.global
+
+    override def force[T](wrapped: monix.eval.Task[T]): T =
+      try {
+        wrapped.runAsync.futureValue
+      } catch {
+        case e: TestFailedException => throw e.getCause
+      }
   }
+  val catsIo = new ForceWrappedValue[cats.effect.IO] {
+    override def force[T](wrapped: cats.effect.IO[T]): T =
+      wrapped.unsafeRunSync
+  }
+}
+
+trait ForceWrapped extends ScalaFutures { this: Suite =>
+  type ForceWrappedValue[R[_]] = com.softwaremill.sttp.ForceWrappedValue[R]
+  val ForceWrappedValue: com.softwaremill.sttp.ForceWrappedValue.type =
+    com.softwaremill.sttp.ForceWrappedValue
 
   implicit class ForceDecorator[R[_], T](wrapped: R[T]) {
     def force()(implicit fwv: ForceWrappedValue[R]): T = fwv.force(wrapped)