Observable Operator Hinrichtungen

Question

mit dem folgenden Beispiel (Kotlin Code)

val subject = PublishSubject.create<Int>() 

val stream = subject.map { 
    println("mapping") 
    it * 2 
} 

stream.forEach { println("A: $it") } 
stream.forEach { println("B: $it") } 

subject.onNext(1) 
subject.onCompleted() 

Ausgang wird wieder zu verwenden sein

mapping 
A: 2 
mapping 
B: 2 

, was ich will erreichen, dass Quelle beobachtbaren abgebildet einmal werden und alle Teilnehmer Holen Sie sich das Ergebnis, aber nicht die Mapping-Operation für jede einzelne von ihnen ausführen ...

so

mapping 
A: 2 
B: 2 

In meinem Fall habe ich sehr teuer Berechnung geht, wo Latenz und Leistung kritisch, ich habe eine heiße beobachtbaren als Quelle und viele Abonnenten ...

Wie Wiederverwendung wir Betreiber Hinrichtungen? und generell verschiedene Mapping-Operationen?

Answer 1

Sie können cache verwenden das Ergebnis der Quelle beobachtbaren für alle künftigen Teilnehmer cachen:

val stream = subject.map { 
    println("mapping") 
    it * 2 
}.cache() 

Wenn Sie eine subtilere Kontrolle darüber, wie die Dinge wollen zwischengespeichert werden, die replay wert ist in suchen.

Wenn Sie nicht jedes Element der Quelle cachen wollen beobachtbar, sondern nur erneut veröffentlichen neue Objekte können Sie verwenden publish mit autoConnect:

val stream = subject.map { 
    println("mapping") 
    it * 2 
}.publish() 
.autoConnect() 

, die die folgende Sequenz von Ereignissen gegeben:

stream.forEach { println("A: $it") } 
stream.forEach { println("B: $it") } 

subject.onNext(1) 

stream.forEach { println("C: $it") } 
subject.onNext(2) 
subject.onCompleted() 

Würde drucken:

mapping 
A: 2 
B: 2 
mapping 
A: 4 
B: 4 
C: 4 

Answer 2

Ich habe die Lösung gefunden. Um die Ausführung der Pipe wiederzuverwenden, müssen wir sicherstellen, dass es nur einen Abonnenten gibt und dieser Abonnent alle Emissionen vom Ende der Pipe zu den Einträgen aller Abonnenten weiterleitet ... Das klingt sehr nach Subject!

Wenn wir nur 100-mal abonniert werden wir 100 Rohre von der Quelle beginnen zu beobachten, während in diesem Fall haben wir ein Rohr, die Zweige zu 100 winzigen Rohren im Ende davon ...

fun <T> Observable<T>.hub(): Observable<T> { 
    val hub = PublishSubject.create<T>() 
    this.subscribe(hub) 
    return hub 
} 

jetzt können wir das

val subject = PublishSubject.create<Int>() 

val stream = subject.map { 
    println("mapping") 
    it * 2 
} 

val hub = stream.hub() 

hub.subscribe { println("A: $it") } 
hub.subscribe { println("B: $it") } 

subject.onNext(1) 

subject.onCompleted() 

die diese

mapping 
A: 2 
B: 2 

geben Problem gelöst!