RxJava: dlaczego subskrybowanie wspólnych obserwowalnych zmian emitowanych elementów - rx-java, reactive-programming, rx-java2

Natknąłem się na pewne zagadkowe zachowanie, którego nie potrafięwyjaśniać. Przedstawiłem poniższy przykład ze znacznie większego łańcucha rx, więc nie dziw się, dlaczego, do licha, miałbym to zrobić. Chcę tylko dowiedzieć się, dlaczego tak się dzieje! :)

enum class Request {
Request1,
Request2
}


fun main(args: Array<String>) {
val requestStream = PublishSubject.create<Request>()

val stateChanges = requestStream.share()

stateChanges
.delaySubscription(requestStream)
.subscribe({ println("received $it") })

// Comment this and it changes the output!
stateChanges.subscribe()

requestStream.onNext(Request.Request1)
requestStream.onNext(Request.Request2)
}

Więc. jeśli uruchomisz powyższy program, wydrukuje:

received Request1
received Request2

Ale jeśli skomentujesz stateChanges.subscribe(), nagle Request1 gubi się i drukuje tylko to:

received Request2

Możesz to wyjaśnić? Chciałbym również wiedzieć, czy można mieć powyższą konfigurację, aby emitować oba elementy, nawet w przypadku braku dodatkowych subscribe().

Odpowiedzi:

W normalnym przypadku share jest już połączony z requestStream i z powodu delaySubscription(requestStream), share dostaje to sekundę Observer przed requestStream wysyła to Request1. Więc requestStream ma dwa Observersi wysłanie Request1 do pierwszego dodaje kolejne Observer drugiemu konsumentowi - share, dlatego subskrybent końcowy otrzymuje Request1.

W skomentowanym przypadku share nie ma połączenia z requestStream jednak dlatego requestStream może tylko powiadomić delaySubscription. delaySubscription wyzwalacze share który następnie subskrybuje requestStream. Jednak, PublishSubject emituje elementy tylko do bieżącej migawki Observersi nie widzi pierwszego onNext dodał nowy Observer w międzyczasie. Tak więc, Request1 nie osiąga println.

Ten przypadek narożny nie jest obsługiwany przez PublishSubject ponieważ wymagałoby tego onNext pamiętać, które Observer otrzymał już bieżący element i spróbuj ponownie w przypadku zestawu prądu Observer zmienione. Zwiększa to pamięć i czas.