RxJava: pourquoi s'abonner à des modifications observables partagées émet des éléments - rx-java, programmation réactive, rx-java2

Je suis tombé sur un comportement étonnant que je ne peux pasExplique. J'ai distillé l'exemple ci-dessous à partir d'une chaîne beaucoup plus grande, alors ne soyez pas surpris de savoir pourquoi je ferais cela. Je veux juste comprendre pourquoi cela se produit! :)

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)
}

Alors. si vous exécutez le programme ci-dessus, il affichera:

received Request1
received Request2

Mais si vous commentez stateChanges.subscribe(), soudainement Request1 se perd et il n'imprime que ceci:

received Request2

Pouvez-vous l'expliquer? J'aimerais aussi savoir s'il est possible d'avoir la configuration ci-dessus pour émettre les deux éléments même en subscribe().

Réponses:

Dans le cas normal, share est déjà connecté à requestStream et en raison de delaySubscription(requestStream), share obtient la seconde Observer avant requestStream envoie le Request1. Alors requestStream a deux Observers et l'envoi de Request1 au premier ajoute un autre Observer au deuxième consommateur, le sharePar conséquent, l'abonné final reçoit Request1.

Dans le cas commenté, share n'a pas été connecté à requestStream pourtant, donc requestStream peut seulement informer le delaySubscription. delaySubscription déclencheurs share qui souscrit ensuite à requestStream. cependant, PublishSubject émet des éléments uniquement à l'instantané actuel Observers et ne peut pas "voir sa première onNext ajouté un nouveau Observer pendant ce temps. Ainsi, Request1 n’atteint pas le println.

Cette affaire d'angle n'est pas traitée par PublishSubject comme il le faudrait onNext se rappeler lequel Observer a déjà reçu l’élément en cours et réessayez au cas où l’ensemble des éléments en cours Observer modifié. Cela ajoute de la mémoire et du temps supplémentaire.