两个可观察物之间的差异 [英] Difference between two observables

问题描述

比方说，我有两个观测值.

Let's say I have two observables.

第一个可观察到的是某些列表的数组:

The first observable is an array of certain listings:

[
    {id: 'zzz', other props here...},
    {id: 'aaa', ...},
    {id: '007', ...}
    ... and more over time
]

第二个可观察的是被忽略列表的数组:

The second observable is an array of ignored listings:

[
    {id: '007'}, // only id, no other props
    {id: 'zzz'}
    ... and more over time
]

结果应该是一个新的可观察列表(第一个可观察)，但不得包含任何被忽略的列表:

The result should be a new observable of listings (first observable) but must not have any of the ignored listings:

[
    {id: 'aaa', other props here...}
    ... and more over time
] 

这是我发布之前所拥有的:

This is what I have now before posting:

obs2.pipe(withLatestFrom(obs1, ? => ?, filter(?));

推荐答案

如果我理解正确，那么您要做的就是

If I'm understanding correctly, what you'll want to do is

1. 汇总一段时间内收到的项目
2. 汇总随时间推移将被忽略的id
3. 最后，由于上述两个流都随时间推移而发出，因此发出的结果列表将不包含被忽略的id.

鉴于以上所述，下面是您可以尝试的一个粗略示例.如底部所述，根据前两个流的节奏，您将获得不同的结果，因为，这就是异步发生的情况.为了证明这一点，我正在模拟随时间推移的事物发射的随机延迟.

Given the above, below is a rough example you could try. As noted towards the bottom, you'll get different results depending on the cadence of the first two streams because, well, thats's what happens with async. To show that, I'm simulating a random delay in the emission of things over time.

希望这会有所帮助！

P.S .:以下为Typescript，假设使用rxjs @ ^ 6.

P.S.: The below is Typescript, assuming rxjs@^6.

import { BehaviorSubject, combineLatest, of, Observable } from "rxjs";
import { delay, map, scan, concatMap } from "rxjs/operators";

/**
 * Data sources
 */

// Just for showcase purposes... Simulates items emitted over time
const simulatedEmitOverTime = <T>() => (source: Observable<T>) =>
  source.pipe(
    concatMap(thing => of(thing).pipe(delay(Math.random() * 1000)))
  );

interface Thing {
  id: string;
}

// Stream of things over time
const thingsOverTime$ = of(
  { id: "zzz" },
  { id: "aaa" },
  { id: "007" }
).pipe(
  simulatedEmitOverTime()
);

// Stream of ignored things over time
const ignoredThingsOverTime$ = of(
  { id: "007" },
  { id: "zzz" }
).pipe(
  simulatedEmitOverTime()
);


/**
 * Somewhere in your app
 */

// Aggregate incoming things
// `scan` takes a reducer-type function
const aggregatedThings$ = thingsOverTime$.pipe(
  scan(
    (aggregatedThings: Thing[], incomingThing: Thing) =>
      aggregatedThings.concat(incomingThing),
    []
  )
);

// Create a Set from incoming ignored thing ids
// A Set will allow for easy filtering over time
const ignoredIds$ = ignoredThingsOverTime$.pipe(
  scan(
    (excludedIdSet, incomingThing: Thing) =>
      excludedIdSet.add(incomingThing.id),
    new Set<string>()
  )
);

// Combine stream and then filter out ignored ids
const sanitizedThings$ = combineLatest(aggregatedThings$, ignoredIds$)
  .pipe(
    map(([things, ignored]) => things.filter(({ id }) => !ignored.has(id)))
  );

// Subscribe where needed
// Note: End result will vary depending on the timing of items coming in
// over time (which is being simulated here-ish)
sanitizedThings$.subscribe(console.log);

这篇关于两个可观察物之间的差异的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！