async-http-client-backend/fs2/src/test/scala/com/softwaremill/sttp/asynchttpclient/fs2/AsyncHttpClientFs2HttpStreamingTest.scala


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

package com.softwaremill.sttp.asynchttpclient.fs2

import java.nio.ByteBuffer

import cats.effect.IO
import cats.instances.string._
import com.softwaremill.sttp.SttpBackend
import com.softwaremill.sttp.testing.streaming.{StreamingTest, TestStreamingBackend}
import com.softwaremill.sttp.testing.ConvertToFuture
import fs2.{Chunk, Stream, text}
import scala.concurrent.Future

class AsyncHttpClientFs2HttpStreamingTest extends StreamingTest[IO, Stream[IO, ByteBuffer]] {

  override val testStreamingBackend: TestStreamingBackend[IO, Stream[IO, ByteBuffer]] =
    new AsyncHttpClientFs2TestStreamingBackend
}

class AsyncHttpClientFs2TestStreamingBackend extends TestStreamingBackend[IO, Stream[IO, ByteBuffer]] {

  override implicit val backend: SttpBackend[IO, Stream[IO, ByteBuffer]] =
    AsyncHttpClientFs2Backend[IO]()

  override implicit val convertToFuture: ConvertToFuture[IO] =
    new ConvertToFuture[IO] {
      override def toFuture[T](value: IO[T]): Future[T] =
        value.unsafeToFuture()
    }

  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)
      .compile
      .foldMonoid

}