scala - 为什么head不取消订阅

Question

假设您在 rxjava-scala-0.18.4

中有以下 Observable

@volatile var dorun = true
var subscriber: Subscriber[String] = null
val myObs = Observable { obs: Subscriber[String] =>
  subscriber = obs
  Subscription { println("unsubscribed"); dorun = false }
}

val sub = myObs.head.subscribe(println(_))

assertTrue(dorun)

subscriber.onNext("hello")
Thread.sleep(500)
assertFalse(dorun)

subscriber.onNext("world")
Thread.sleep(500)
assertFalse(dorun)

第二个断言失败，这意味着 head 没有退订。我对 Observables 的理解是错误的还是应该 head 在第一个元素发出后取消订阅？

Answer 1

看看你的 subscribe() 方法:你循环直到 run 被设置为 false，但唯一的方法是碰巧是关闭订阅。问题是还没有人订阅:循环阻止您返回。 head 运算符无法在第一个项目交付后终止基础订阅，因为它尚未完成订阅。因此，您只会一直循环下去。

一种解决方案是将您的循环移动到在 Schedulers.trampoline() 上安排的操作中。然后事件将在从 subscribe() 返回后的某个时间传递。

此外，在您的 subscribe() 方法中，您似乎需要将新的订阅对象添加到传入的 Subscriber 中，如下所示:

val myObs = Observable {
    obs: rx.lang.scala.Subscriber[String] =>
        ...

        obs.add(
            Subscription {
                dorun = false
                println("unsubscribed")
            }
        )
}