Comment créer mon propre Observable chaud à partir de zéro?

Question

Comment créer mon propre Observable chaud à partir de zéro?

J'aimerais créer ma propre fonction, renvoyant un observable, retournant des emplacements:

public static Observable locationObservable(Context context, String provider, long minTime, float minDistance) {

Ceci est pour Android. Il est recommandé d'utiliser Observable.create() à cette fin, mais l'exemple montre simplement le passage d'une liste constante d'entiers à chaque abonné, ce qui n'est pas chaud.

Si je fais quelque chose d'autre ici, par exemple, me rappeler une liste d'abonnés, alors comment implémenter la désinscription et de nombreuses autres fonctionnalités?

C'est-à-dire, je n'ai absolument aucune idée de ce qu'il faut faire à l'intérieur de l'implémentation de Observable.OnSubscribe()?

Answer 1

Généralement pour créer un observable chaud, vous utilisez une sorte de Subject : PublishSubject, BehaviorSubject, etc.

Consultez des exemples pour BehaviorSubject ici.

class LocationService {
    private Subject subject = BehaviorSubject.create();

    Observable locationObservable(...) {
        return subject;
    }

    void onNewLocationListener(Location newLocation) {
        subject.onNext(newLocation);
    }
}

Answer 2

Il n'est pas recommandé d'écrire le vôtre, du moins jusqu'à ce que vous maîtrisiez ceux existants et ayez besoin d'un modèle de mise en cache/émission particulier non couvert par les 5 valeurs par défaut (Async, Behavior, Publish, Replay, Unicast).

J'ai une série en 3 parties sur le sujet (jeu de mots):

• Partie 1
• Partie 2
• Partie 3

Answer 3

Regardez cet exemple merveilleux pris directement depuis RealmObservableFactory de Realm :

@Override
public Observable from(Realm realm) {
    final RealmConfiguration realmConfig = realm.getConfiguration();
    return Observable.create(new Observable.OnSubscribe() { // créer un nouvel observable
        @Override
        public void call(final Subscriber subscriber) { // ceci est exécuté sur `subscribeOn(Scheduler)`
            final Realm observableRealm = Realm.getInstance(realmConfig);
            final RealmChangeListener listener = new RealmChangeListener() {
                @Override
                public void onChange(Realm realm) {
                    if (!subscriber.isUnsubscribed()) { // vérifiez toujours si le subscriber s'est désabonné !
                        subscriber.onNext(observableRealm);
                    }
                }
            };
            subscriber.add(Subscriptions.create(new Action0() { // ajoutez d'abord la désinscription ! thread spécifié par unsubscribeOn(Scheduler)
                @Override
                public void call() {
                    observableRealm.removeChangeListener(listener); // enlever le listener
                    observableRealm.close();
                }
            }));
            observableRealm.addChangeListener(listener); // ajouter le listener
            subscriber.onNext(observableRealm); // valeur initiale
        }
    });
}

Et lisez les commentaires, c'est un très bon exemple.