Come ottenere la notifica dell'azione di annullamento dell'iscrizione di un osservatore in un Osservabile personalizzato in RxJava

Question

Sto provando a racchiudere un'API basata su listener-pattern su Observable. Il mio codice sembra quasi seguito.

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

Tuttavia, voglio che il mio osservabile di rimuovere immediatamente l'ascoltatore dal existingEventSource sottostante quando l'osservatore chiama subscription.unscribe(). Come potrei raggiungere questo obiettivo?

Answer 1

La classe astratta ha un metodo add che consente di aggiungere Subscription s che verrà annullata con l'utente.

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

Pensate aSubscriber come Observer che ha sottoscritto il tuo Observable; lo chiameremo Subscriber. Finché il numero Subscriber è ancora sottoscritto allo Observable, lo Observable può emettere valori. Ma quando quello Subscriber non sottoscritto dovrebbe fermarsi. Ma se vogliamo ricevere una notifica quando il Subscriber disiscrive, possiamo registrare un Action da eseguire quando succede. Questo è il motivo per cui viene utilizzato il metodo . Come menzionato da @dwursteisen nei commenti; in pratica stai registrando un lambda che verrà eseguito quando il Sottoscrittore non sottoscrive.

È inoltre possibile annullare la sottoscrizione dell'abbonamento su un altro programma di pianificazione. Vedere MainThreadSubscription dal progetto rxanroid per un esempio su come ottenerlo.

Ecco un esempio di come ci si utilizza per annullare la sottoscrizione sul thread principale

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