RxJava - Verwendung von Themen

Question

Ich verstehe, dass die Verwendung von Themen in der RxJava Welt im Allgemeinen abgeraten wird. Ich versuche, auf Objektänderungen eines Objekts zu beobachten, die mit einem Subjekt leicht erreicht werden können. Ich weiß, dass rx.net Observable.FromEvent() zur Verfügung stellt, um diese Situation leicht zu handhaben, aber ich frage mich, ob es sich um eine Java-Alternative handelt, die keine Subjekte sind.

Observable.create(emitter -> { 
    PropertyObserver x = event -> emitter.onNext(event) 
    yourObject.addObserver(x); 
    // Set cancellable after adding observer, if the observable is 
    // disposed already it will properly remove the listener. 
    emitter.setCancellable(() -> yourObject.removeObserver(x)); 
}); 

Aber Sie müssen sich bewusst sein, dass jedes Mal diese beobachtbaren verwendet wird, um dem zugehörigen Objekt einen neuen Listener fügt hinzu:

Answer 1

Sie können diese Funktionalität durch die Observable.create Methode erreichen. Dies kann vermieden werden, indem die Operatoren share() und publish() angewendet werden, wenn mehrere Beobachter von den Ereignissen des Objekts abhängig sind.

Answer 2

Verwenden Sie create Operator. Hier erhalten Sie ein komplettes Beispiel.

package xdean.stackoverflow.rx; 

import java.util.LinkedList; 
import java.util.List; 

import io.reactivex.Observable; 
import io.reactivex.ObservableEmitter; 
import io.reactivex.disposables.Disposable; 

public class Q47000589 { 
    public static void main(String[] args) { 
    Property<Integer> p = new Property<>(1); 
    Disposable d = p.listen().subscribe(i -> System.out.println("Listen: " + i)); 
    p.set(2); 
    p.set(3); 
    d.dispose(); 
    p.set(4); 
    p.set(5); 
    } 

    public static class Property<T> { 
    T value; 
    List<ObservableEmitter<? super T>> emitters = new LinkedList<>(); 

    public Property(T value) { 
     this.value = value; 
    } 

    T get() { 
     return value; 
    } 

    void set(T t) { 
     this.value = t; 
     emitters.forEach(e -> e.onNext(t)); 
    } 

    Observable<T> listen() { 
     return Observable.create(e -> { 
     emitters.add(e); 
     e.onNext(value); 
     e.setCancellable(() -> emitters.remove(e)); 
     }); 
    } 
    } 
} 

Und Ausgabe lautet:

Listen: 1 
Listen: 2 
Listen: 3