0

我正在尝试将响应流订阅者附加到 akka 源。

我的来源似乎可以与一个简单的接收器（如 foreach）一起正常工作 - 但如果我放入一个由订阅者制作的真正接收器，我什么也得不到。

我的背景是：

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{Sink, Source}
import org.reactivestreams.{Subscriber, Subscription}

implicit val system =  ActorSystem.create("test")
implicit val materializer = ActorMaterializer.create(system)

class PrintSubscriber extends Subscriber[String] {
  override def onError(t: Throwable): Unit = {}
  override def onSubscribe(s: Subscription): Unit = {}
  override def onComplete(): Unit = {}
  override def onNext(t: String): Unit = {
    println(t)
  }
}

我的测试用例是：

val subscriber = new PrintSubscriber()
val sink = Sink.fromSubscriber(subscriber)

val source2 = Source.fromIterator(() => Iterator("aaa", "bbb", "ccc"))
val source1 = Source.fromIterator(() => Iterator("xxx", "yyy", "zzz"))
source1.to(sink).run()(materializer)
source2.runForeach(println)

我得到输出：

aaa
bbb
ccc

为什么我没有得到 xxx、yyy 和 zzz？

4

1 回答 1

2

引用以下 Reactive Streams 规范Subscriber：

将订阅者的实例传递给 Publisher.subscribe(Subscriber) 后将收到一次对 onSubscribe(Subscription) 的调用。在调用 Subscription.request(long) 之前不会收到进一步的通知。

要查看某些项目流向订阅者，您可以做出的最小更改是

override def onSubscribe(s: Subscription): Unit = {
  s.request(3)
}

但是，请记住，这不会使其完全符合 Reactive Streams规范。不太容易实现是 Akka-Streams 本身等高级工具包背后的主要原因。

于 2017-02-08T22:03:53.540 回答