RxJs comment fusionner deux observable se chevauchent dans une

Question

I ont deux observables:

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

Le premier contient un certain nombre croissant, mais arrête après un certain temps (ce sont les résultats d'un curseur à partir de la base de données) Les seconds sont émettant en continu un nombre croissant. Contient un certain nombre de la première, mais n'arrête pas d'émettre. (Ce sont les nouvelles données insérées dans la base de données)

Je veux que ce deux observables à regarder observable continue comme ceci:

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

Cette observable contient chaque numéro qu'une seule fois, en gardant l'ordre d'émission. Comment peut-il être résolu en utilisant aussi peu de mémoire que possible?

Answer 1

Je pense que la meilleure approche ici est de mettre en tampon b $ jusqu'à ce qu'un flux $ atteigne b $, puis d'émettre tous les éléments bufférisés de b $ et de passer à b $. Quelque chose comme ceci:

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

Vous pouvez le faire en prenant tous les éléments du premier flux concaténé (.concat) avec le second courant, sauf (.skipWhile inclus) éléments avant le dernier (.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>

Aussi, si vous avez des effets secondaires lors de l'abonnement (par exemple, lorsque vous vous abonnez, le nouveau curseur sera créé), vous pouvez partager cet effet secondaire pour tous les abonnés en utilisant par exemple const a$ = fromMarble(a).shareReaplay().

Vous pouvez lire plus sur le partage des effets secondaires:

• dans l'ancienne documentation RxJS v4 - 4.8 Use the publish operator to share side-effects
• et de cet article - RxSwift: share vs replay vs shareReplay