如何在RxJava中的自定义Observable中获得观察者取消订阅操作的通知 [英] How to get notified of a observer's unsubscribe action in a custom Observable in RxJava

问题描述

我正在尝试将一些基于侦听器模式的API包装到Observable中.我的代码大致如下所示.

I'm trying to wrap some listener-pattern based API to an Observable. My code roughly looks like following.

def myObservable = Observable.create({ aSubscriber ->
    val listener = {event -> 
      aSubscriber.onNext(event);                
    }
    existingEventSource.addListener(listener)
})

但是，当观察者调用subscription.unscribe()时，我希望我的可观察对象立即从基础的existingEventSource中删除侦听器.我怎样才能实现这个目标?

However, I want my observable to immediately remove the listener from the underlying existingEventSource when the observer calls subscription.unscribe(). How could I achieve this goal?

推荐答案

Subscriber抽象类实际上具有方法add，通过该方法，您可以添加Subscription，而订阅者将不对其进行订阅.

The Subscriber abstract class actually has a method add which lets you add Subscriptions which will be unsubscribed with the subscriber.

def myObservable = Observable.create({ aSubscriber ->
    val listener = {event -> 
      aSubscriber.onNext(event);                
    }
    existingEventSource.addListener(listener)

    // Adds a lambda to be executed when the Subscriber un-subscribes from your Observable
    aSubscriber.add(Subscriptions.create(() -> existingEventSource.removeListener(listener)));
})

将aSubscriber视为订阅了Observable的Observer；我们将其称为Subscriber.只要Subscriber仍订阅Observable，Observable即可发出值.但是，当该Subscriber取消订阅时，它应该停止.但是，如果我们希望在Subscriber取消订阅时得到通知，我们可以注册Action在它发生时运行.这就是add方法的用途.正如@dwursteisen在评论中提到的那样；您基本上是在注册一个lambda，该lambda将在订阅者取消订阅时执行.

Think of aSubscriber as the Observer that subscribed to your Observable; we'll call it a Subscriber. As long as the Subscriber is still subscribed to the Observable, the Observable can emit values. But when that Subscriber un-subscribed then it should stop. But if we want to get notified when the Subscriber unsubscribes we can register an Action to be run when it happens. This is what the add method is used for. As mentioned by @dwursteisen in the comments; you're basically registering a lambda that will be executed when the Subscriber un-subscribes.

也可以使订阅退订在其他调度程序上.请参阅 MainThreadSubscription rxanroid项目中有关如何实现该目标的示例.

It's also possible to have the Subscription be unsubscribe on a different Scheduler. See MainThreadSubscription from the rxanroid project for an example of how to achieve that .

以下是您如何使用它来取消订阅主线程的示例

Here's an example of how you'd use it for unsubscribing on the main thread

aSubscriber.add(new MainThreadSubscription() {
    @Override
    protected void onUnsubscribe() {
        existingEventSource.removeListener(listener);
    }
});

这篇关于如何在RxJava中的自定义Observable中获得观察者取消订阅操作的通知的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！