RxJava2 flatMap erstellt doppelte Ereignisse

Question

Ich bin relativ neu auf RxJava2 und ich bekomme einige seltsame Verhaltensweisen, so ist es wahrscheinlich, dass ich das Tool auf dem falschen Weg benutze.

Es ist ein ziemlich großes Projekt, aber ich habe das Snippet unten als Mindest reproduzierbaren Code getrennt:

Observable 
    .interval(333, TimeUnit.MILLISECONDS) 
    .flatMap(new Function<Long, ObservableSource<Integer>>() { 
    private Subject<Integer> s = PublishSubject.create(); 
    private int val = 0; 

    @Override public ObservableSource<Integer> apply(Long aLong) throws Exception { 
     val++; 
     s.onNext(val); 
     return s; 
     } 
    }) 
    .subscribe(new Consumer<Integer>() { 
    @Override public void accept(Integer integer) throws Exception { 
     Log.w("value: %s", integer); 
    } 
    }); 

Dieser Code simuliert von meinem rx-Stream-Ereignisse eine .interval und flatMap mit erhalten diese Ereignisse " mach etwas Verarbeitung "und verwendet eine Subject, um Ergebnisse in den Stream zu drücken.

Der Stream ist ein fortlaufender Prozess mit mehreren verschiedenen Ereignissen. Dieser Mindestcode ist albern, weil ich nur auf den apply Callback dränge, aber im wirklichen Fall gibt es mehrere mögliche Momente, in denen ein Push passieren kann und die Anzahl der Ereignisse, die während apply empfangen werden, ist nicht die gleiche Menge das wird über den Betreff gesendet.

Was ich mit diesem Code zu sehen erwarten:

value: 2 // 1 got skipped because onNext is called before there's a subscriber. 
value: 3 
value: 4 
value: 5 
value: 6 ... etc 

, was ich habe eigentlich ist:

value: 2 
value: 3 
value: 3 // 3 twice 
value: 4 
value: 4 
value: 4 // 4 repeated 3 times 
value: 5 
value: 5 
value: 5 
value: 5 // 5 repeated 4 times 
value: 6 
value: 6 
value: 6 
value: 6 
value: 6 // 6 repeated 5 times 
... etc 

Ich habe auch eine Observable<Integer> o = s.share(); und Rückkehr, es zu haben versucht, oder direkt zurück s.share(); mit den gleichen Ergebnissen.

Ich verstehe irgendwie, warum das passiert. Die ObservableSource wird wieder abonniert und wieder n wieder, so dass es mehr Ereignisse auf jeder Schleife gibt.

Die Frage:

Wie kann ich mein erwartetes Verhalten erreichen?

(im Fall war mein erwartetes Verhalten nicht klar, bitte mehr auf den Kommentaren fragen)

Answer 1

Ihre PublishSubject abonnierte mehrfach, einmal aus Intervall pro Stück().

Bearbeiten: Sie müssen jedes Mal eine neue PublishSubject übergeben (wechseln Sie zu BehaviorSubject, wenn Sie die erste/letzte Emission beibehalten möchten); Übergeben Sie das an den lang andauernden Prozess, und stellen Sie sicher, dass sein onComplete ordnungsgemäß aufgerufen wird, wenn der langwierige Prozess beendet wird.

Answer 2

bearbeiten

Nach jüngsten Äußerungen ich mit dieser Art von einer Lösung kommen könnte:

class MyBluetoothClient { 
    private PublishSubject<BTLEEvent> statusPublishSubject = PublishSubject.create() 

    public Observable<BTLEEvent> getEventObservable() { 
    return statusPublishSubject 
    } 

    private void publishEvent(BTLEEvent event) { 
    statusPublishSubject.onNext(event) 
    } 

    public void doStuff1() { 
    // do something that returns: 
    publishEvent(BTLEEvent.someEvent1) 
    } 

    public void doStuff2() { 
    // do something else that eventually yields 
    publishEvent(BTLEEvent.someEvent2) 
    } 
} 

Und Sie es auf diese Weise verwenden:

MyBluetoothClient client = MyBluetoothClient() 
client 
    .getEventObservable() 
    .subscribe(/* */) 

/// 

client.doStuff1() 

/// 

client.doStuff2 

Original-Antwort

Wird dies tun?

Observable 
    .interval(333, TimeUnit.MILLISECONDS) 
    .flatMap(new Function<Long, ObservableSource<Integer>>() { 
    private int val = 0; 

    @Override public ObservableSource<Integer> apply(Long aLong) throws Exception { 
     val++; 
     return Observable.just(val); 
     } 
    }) 
    .subscribe(new Consumer<Integer>() { 
    @Override public void accept(Integer integer) throws Exception { 
     Log.w("value: %s", integer); 
    } 
    }); 

Answer 3

Also hier ist die Antwort, die ich kam. Ich werde @Tassos Antwort als richtig markieren, da er mich auf den richtigen Weg gezeigt hat.

Zuerst brauche ich eine CachedSubject (ein Subjekt, das Elemente zwischenspeichert, während es keine Beobachter gibt und sie versendet, sobald ein Beobachter sich verbindet), dies ist notwendig, um sicherzustellen, dass Emissionen aus dem apply wirklich durchkommen. Die Klasse wickelt meist eine PublishSubject.

class CachedSubject<T> extends Subject<T> { 

     private PublishSubject<T> publishSubject = PublishSubject.create(); 
     private Queue<T> cache = new ConcurrentLinkedQueue<>(); 

     @Override public boolean hasObservers() { 
      return publishSubject.hasObservers(); 
     } 

     @Override public boolean hasThrowable() { 
      return publishSubject.hasThrowable(); 
     } 

     @Override public boolean hasComplete() { 
      return publishSubject.hasComplete(); 
     } 

     @Override public Throwable getThrowable() { 
      return publishSubject.getThrowable(); 
     } 

     @Override protected void subscribeActual(Observer<? super T> observer) { 
      while (cache.size() > 0) { 
       observer.onNext(cache.remove()); 
      } 
      publishSubject.subscribeActual(observer); 
     } 

     @Override public void onSubscribe(Disposable d) { 
      publishSubject.onSubscribe(d); 
     } 

     @Override public void onNext(T t) { 
      if (hasObservers()) { 
       publishSubject.onNext(t); 
      } else { 
       cache.add(t); 
      } 
     } 

     @Override public void onError(Throwable e) { 
      publishSubject.onError(e); 
     } 

     @Override public void onComplete() { 
      publishSubject.onComplete(); 
     } 
    } 

dann verwende ich diese Klasse mit einem switchMap:

Observable 
    .interval(1000, TimeUnit.MILLISECONDS) 
    .switchMap(new Function<Long, ObservableSource<Integer>>() { 

     private Subject<Integer> s = new CachedSubject<>(); 
     private int val = 0; 

     @Override public ObservableSource<Integer> apply(Long aLong) throws Exception { 
     val++; 
     s.onNext(val); 
     return s; 
     } 
    }) 
    .subscribe(new Consumer<Integer>() { 
     @Override public void accept(Integer integer) throws Exception { 
     Log.w("value: %s", integer); 
     } 
    }); 

Diese mich effektiv ermöglicht eine beliebige Anzahl von Veranstaltungen auf dem apply<T t> Methode zu erhalten und haben nur 1 Consumer diesen abonniert, empfängt alle Ereignisse davon.