如何用sttp读取响应作为Observable[String]

Question

我用的是sttp客户端。我希望将响应作为字符串除以行(如Observable[String] )。

在这里，sttp流api：

import java.nio.ByteBuffer

import com.softwaremill.sttp._
import com.softwaremill.sttp.okhttp.monix.OkHttpMonixBackend
import monix.eval.Task
import monix.reactive.Observable

implicit val sttpBackend = OkHttpMonixBackend()

val res: Task[Response[Observable[ByteBuffer]]] = sttp
  .post(uri"someUri")
  .response(asStream[Observable[ByteBuffer]])
  .send()

那我怎么能得到Observable[String]呢？

这里有一些想法：

1.有一个简单的方法可以通过线观察到split吗？

2.，或者也许我可以从响应中获得原始的InputStream，这样我就可以轻松地拆分它，但是我无法找到使用类似asStream[InputStream]的方法。

3.还是只使用http后端witout sttp层？

Answer 1

您的基本问题是如何将Observable[ByteBuffer]转换为Observable[String]，其中每个String都是一行，对吗？

您可以使用bufferWithSelector(selector: Observable[S]): Observable[Seq[A]]方法。这种方法将缓冲可观测到的，直到选择器可观测发射一个元素。

我用Int做了一个小例子：

import monix.reactive.Observable
import monix.execution.Scheduler.Implicits.global
import scala.concurrent.duration._

val source = Observable.range(0, 1000, 1)
  .delayOnNext(100.milliseconds)

val selector = source.filter(_ % 10 == 0)

val buffered = source.bufferWithSelector(selector)
  .map(_.foldLeft("")((s, i) => s + i.toString)) // This folds the Seq[Int] into a String for display purposes

buffered.foreach(println)

试试看！

当然，这有一个主要的缺点:底层可观察到的source将被评估两次。您可以通过修改上面的示例来看到这一点：

// Start writing your ScalaFiddle code here

import monix.reactive.Observable
import monix.execution.Scheduler.Implicits.global
import scala.concurrent.duration._

val source = Observable.range(0, 1000, 1)
  .delayOnNext(100.milliseconds)
  .map {x => println(x); x}  // <------------------

val selector = source.filter(_ % 10 == 0)

val buffered = source.bufferWithSelector(selector)
  .map(_.foldLeft("")((s, i) => s + i.toString))

buffered.foreach(println)

这个会把每个数字打印两次。

要解决这个问题，您必须将可观察到的source转换为可观察的热点：

import monix.reactive.Observable
import monix.execution.Scheduler.Implicits.global
import scala.concurrent.duration._

val source = Observable.range(0, 1000, 1)
  .delayOnNext(100.milliseconds)
  .map {x => println(x); x}
  .publish // <-----------------------------

// source is now a ConnectableObservable and will start emitting elements
// once you call source.connect()

val selector = source.filter(_ % 10 == 0)

val buffered = source.bufferWithSelector(selector)
  .map(_.foldLeft("")((s, i) => s + i.toString))

buffered.foreach(println)

source.connect() // <---------------------------

试试看！

唯一需要做的是修改选择器，使其仅在遇到行提要时才发出项。

我建议先将Observable[ByteBuffer]分解为一个Observable[Byte] (使用flatMap)以避免头痛。