How to get notified of a observer's unsubscribe action in a custom Observable in RxJava

Question

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)
})

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?

Answer 1

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)));
})

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.

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);
    }
});