RxJs wie zwei überlappende beobachtbaren in eine

Question

fusionieren Ich habe zwei Observablen:

-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15-| 
-13--14--15--16--17--18--19-----20---------21--------------22------23--24--> 

Die erste enthält eine steigende Zahl, aber stoppt nach einer gewissen Zeit (dies ist das Cursor-Ergebnis aus der Datenbank) Der zweite ist kontinuierlich steigende Anzahl. Enthält eine Nummer von der ersten, aber hören Sie nicht auf zu emittieren. (Dies sind die neu eingefügten Daten in die Datenbank)

Ich möchte diese zwei beobachtbaren eine kontinuierliche wie dies beobachtbare aussehen:

-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15-16-17-18-19-20-21-----22------23--24--> 

Diese beobachtbaren nur einmal jede Zahl enthält, die Emissions Ordnung zu halten.

Wie kann es mit so wenig Speicher wie möglich gelöst werden?

Answer 1

Ich denke, der beste Ansatz hier ist es, b $ zu puffern, bis ein $ -Stream b $ erreicht, dann alle gepufferten Elemente von b $ ausgeben und zu b $ wechseln. Etwas wie folgt aus:

const a = '-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15'; 
 
const b = '-13--14--15--16--17--18--19-----20---------21--------------22------23--24'; 
 

 
const fromMarble = str => Rx.Observable.from(str.split('-')).concatMap(x => Rx.Observable.of(x).delay(1)).filter(v => v.length).map(x => parseInt(x)); 
 

 
const a$ = fromMarble(a).share(); 
 
const b$ = fromMarble(b).share(); 
 

 
const switchingSignal$ = Rx.Observable.combineLatest(a$, b$.take(1), (a, b) => a >= b).filter(x => x).take(1).share(); 
 

 
const distinct$ = Rx.Observable.merge(
 
\t a$.takeUntil(switchingSignal$).map(x => x + '(from a)'), 
 
\t b$.buffer(switchingSignal$).take(1).mergeMap(buffered => Rx.Observable.from(buffered)).map(x => x + '(from b$ buffer)'), 
 
\t b$.skipUntil(switchingSignal$).map(x => x + '(from b$)') 
 
); 
 

 
distinct$.subscribe(console.log);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.0/Rx.js"></script>

Answer 2

Sie können dies tun, indem alle Elemente aus dem ersten Strom unter verketteten (.concat) mit dem zweiten Strom mit Ausnahme von (.skipWhile einschließlich) Elementen vor spätestens einem (.last)

const a = '-1-2-3-4-5-6-7-8-9-10-11-12-13-14-15' 
 
const b = '-13--14--15--16--17--18--19-----20---------21--------------22------23--24' 
 
const fromMarble = str => Rx.Observable.defer(() => { 
 
    console.log('side effect from subscribing to: ' + str); 
 
    return Rx.Observable.from(str.split('-').filter(v => v.length)); 
 
}); 
 

 
const a$ = fromMarble(a); 
 
const b$ = fromMarble(b); 
 

 
const distinct$ = Rx.Observable.concat(
 
    a$, 
 
    a$.last().switchMap(latest => 
 
    // .skipWhile + .skip(1) => skipWhile but inclusive 
 
    b$.skipWhile(v => v !== latest).skip(1) 
 
), 
 
); 
 

 
distinct$.subscribe(e => console.log(e));

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

Auch wenn Sie beim Abonnieren Nebenwirkungen haben (zum Beispiel beim Abonnieren - der neue Cursor wird erstellt), können Sie diesen Nebeneffekt für alle Teilnehmer freigeben, indem Sie beispielsweise const a$ = fromMarble(a).shareReaplay() verwenden.

Sie können mehr über den Austausch Nebenwirkungen lesen:

• in der alten Dokumentation für RxJS v4 - 4.8 Use the publish operator to share side-effects
• und aus diesem Artikel - RxSwift: share vs replay vs shareReplay