Jak otrzymać powiadomienie o akcji wypisania obserwatora w niestandardowym Obserwowalnym w RxJava

Question

Próbuję zawinąć niektóre API oparte na wzorzec słuchacza do obserwowalnego. Mój kod wygląda z grubsza na następujące.

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

Jednakże chcę moje obserwowalne do natychmiastowego usunięcia słuchacza z podstawowej existingEventSource gdy obserwator wzywa subscription.unscribe(). Jak mogę osiągnąć ten cel?

Answer 1

Klasa abstrakcyjna Subscriber ma faktycznie metodę add, która pozwala na dodanie subskrypcji do subskrypcji.

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

Pomyśl aSubscriber jako Observer że subskrybowanego do Observable; będziemy to nazywać Subscriber. Dopóki Subscriber nadal subskrybuje Observable, Observable może emitować wartości. Ale gdy ten Subscriber wyłączy subskrypcję, powinien się zatrzymać. Ale jeśli chcemy otrzymać powiadomienie, gdy rezygnujemy z subskrypcji Subscriber, możemy zarejestrować Action, który zostanie uruchomiony, gdy się to stanie. Do tego służy metoda add. Jak wspomniano przez @dwstwoisen w komentarzach; w zasadzie rejestrujesz lambdę, która zostanie wykonana, gdy subskrybent zrezygnuje z subskrypcji.

Istnieje również możliwość rezygnacji z subskrypcji w innym harmonogramie. Zobacz MainThreadSubscription z projektu rxanroid, aby dowiedzieć się, jak to osiągnąć.

Oto przykład, w jaki sposób chcesz użyć go do wypisywania na głównym wątku

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