RxJava BehaviorSubject, das den letzten Artikel nicht ausgibt?

Question

Ich habe eine einfache RxJava, mit ReplaySubject, ich könnte das Ergebnis erhalten, wo alle 3 Zahlen gedruckt werden.

val observable : Observable<Int> = Observable.just(1, 2, 3) 
    val subject = ReplaySubject.create<Int>() 
    observable.subscribe(subject) 
    subject.subscribe{ 
     result -> 
     System.out.println("Start $result in Subscription Result") 
    } 

Wenn ich zum Verhalten ändern, erwarte ich, dass die dritte Zahl d 3 gedruckt werden, wie ich immer dachte Verhalten das letzte emittiert Element zu wiederholen ist.

val observable : Observable<Int> = Observable.just(1, 2, 3) 
    val subject = BehaviorSubject.create<Int>() 
    observable.subscribe(subject) 
    subject.subscribe{ 
     result -> 
     System.out.println("Start $result in Subscription Result") 
    } 

Es wird jedoch nichts gedruckt. Warum?

Habe ich hier etwas Wichtiges vermisst? Wenn ja, lassen Sie mich wissen, wie ich den angenommenen letzten Gegenstand (d. H. 3) drucken lassen kann.

Answer 1

Es wird nichts gedruckt, da die Anmeldung bereits beendet wurde. Wenn das Abonnement noch aktiv ist, dann wird 3 gedruckt werden, zum Beispiel:

val o1: Observable<Int> = Observable.just(1, 2, 3) 
val o2: Observable<Int> = Observable.just(4).delay(100,TimeUnit.MILLISECONDS) 
val observable: Observable<Int> = Observable.concat(o1, o2) 
val subject = BehaviorSubject.create<Int>() 
observable.subscribe(subject) 
subject.subscribe{ 
    result -> 
    System.out.println("Start $result in Subscription Result") 
} 
Thread.sleep(1000) 

Sowohl 3 und 4 (nach Verzögerung) gedruckt, mit 3 emittieren als letztes Ereignis vor der Zeichnungs- und 4 wird, nachdem die ausgesandten Abonnement.

Zusätzlich wie durch @akarnokd im Kommentarbereich erklärte ReplaySubject.createWithSize(1) kann immer letzten Punkt Wiedergabe auch nach dem beobachtbaren Abschluss verwendet werden, und wenn einzelnes Element benötigt wird, unabhängig von Strom Vollendungszustand kann observable.takeLast(1).subscribe(subject) verwendet werden, zu gewährleisten:

val observable : Observable<Int> = Observable.just(1, 2, 3) 
val subject = ReplaySubject.createWithSize<Int>(1) 
observable.takeLast(1).subscribe(subject) //can be moved after subject.subscribe as well 
subject.subscribe{ 
    result -> 
    System.out.println("Start $result in Subscription Result") 
}