Adding streaming to the monix async http client handler

1 files changed, 47 insertions, 2 deletions

diff --git a/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala b/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala
index a67be2e..5e6db17 100644
--- a/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala
+++ b/tests/src/test/scala/com/softwaremill/sttp/StreamingTests.scala
@@ -1,11 +1,15 @@
 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.typesafe.scalalogging.StrictLogging
+import monix.reactive.Observable
 import org.scalatest.{BeforeAndAfterAll, FlatSpec, Matchers}
 import org.scalatest.concurrent.{IntegrationPatience, ScalaFutures}
 
@@ -32,12 +36,13 @@ class StreamingTests
   override def port = 51824
 
   akkaStreamingTests()
+  monixStreamingTests()
+
+  val body = "streaming test"
 
   def akkaStreamingTests(): Unit = {
     implicit val handler = AkkaHttpSttpHandler.usingActorSystem(actorSystem)
 
-    val body = "streaming test"
-
     "Akka HTTP" should "stream request body" in {
       val response = sttp
         .post(uri"$endpoint/echo")
@@ -61,4 +66,44 @@ class StreamingTests
       responseBody should be(body)
     }
   }
+
+  def monixStreamingTests(): Unit = {
+    implicit val handler = MonixAsyncHttpClientHandler()
+    import monix.execution.Scheduler.Implicits.global
+
+    val body = "streaming test"
+
+    "Monix Async Http Client" should "stream request body" in {
+      val source = Observable.fromIterable(
+        body.getBytes("utf-8").map(b => ByteBuffer.wrap(Array(b))))
+
+      val response = sttp
+        .post(uri"$endpoint/echo")
+        .streamBody(source)
+        .send()
+        .runAsync
+        .futureValue
+
+      response.body should be(body)
+    }
+
+    it should "receive a stream" in {
+      val response = sttp
+        .post(uri"$endpoint/echo")
+        .body(body)
+        .response(asStream[Observable[ByteBuffer]])
+        .send()
+        .runAsync
+        .futureValue
+
+      val bytes = response.body
+        .flatMap(bb => Observable.fromIterable(bb.array()))
+        .toListL
+        .runAsync
+        .futureValue
+        .toArray
+
+      new String(bytes, "utf-8") should be(body)
+    }
+  }
 }