将多个可观察到的数据与不同的操作/操作相结合

Question

我正在构建一个Angular2应用程序，所以我已经习惯了整个系统的可观察性和反应性扩展。我正在使用TypeScript和rxjs。

现在，我有一个可以观察到的，或者，如果你愿意的话，一条流，一些物体的数组。比方说人-物体。现在，我有了另外两个Person对象流，并希望将它们组合起来，因此我得到了一个始终是最新的流：

var people$ = getPeople();                  // Observable<Person[]>
var personAdded$ = eventHub.personAdded;    // Observable<Person>;
var personRemoved$ = eventHub.personRemoved // Observable<Person>;

var allwaysUpToDatePeople$ = people$.doSomeMagic(personAdded$, personRemoved$, ...);

如果人流发出一个数组，比方说，5个人，在那之后，个人加注流会发出一个人，所有的人流都会发出一个6的数组。如果被重组的人流发射一个人，那么allPeople流应该发出一个人-对象的数组，而不是一个刚刚被人重构的流发出的对象。

有没有一种在rxjs中内置的方法来获得这种行为？

Answer 1

我的建议是将action的概念封装到流中，然后将流合并并直接应用于Array。

第一步是定义一些描述您的操作的函数：

function add(people, person) {
  return people.concat([people]);
}

function remove(people, person) {
  const index = people.indexOf(person);
  return index < 0 ? people : people.splice(index, 1);
}

注意:我们避免在适当的位置改变Array，因为它可能会产生不可预见的副作用。纯度要求我们创建数组的副本。

现在，我们可以使用这些函数并将它们提升到流中来创建一个发出函数的Observable：

const added$ = eventHub.personAdded.map(person => people => add(people, person));
const removed$ = eventHub.personRemoved.map(person => people => remove(people, person));

现在我们得到了以：people => people的形式出现的事件，其中输入和输出将是一个人员数组(在本例中，简化为字符串数组)。

现在我们该怎么把这个联系起来？实际上，我们只关心在我们有一个数组将这些事件应用于：

const currentPeople = 

  // Resets this stream if a new set of people comes in
  people$.switchMap(peopleArray => 

    // Merge the actions together 
    Rx.Observable.merge(added$, removed$)

      // Pass in the starting Array and apply each action as it comes in
      .scan((current, op) => op(current), peopleArray)

      // Always emit the starting array first
      .startWith(people)
  )
  // This just makes sure that every new subscription doesn't restart the stream
  // and every subscriber always gets the latest value
  .shareReplay(1);

这种技术有几种优化，这取决于您的需要(例如，避免函数运行，或者使用二进制搜索)，但是我发现上面对于泛型情况来说比较优雅。

Answer 2

您希望合并所有流(Ghostbuster样式)，然后使用扫描操作符来确定状态。扫描操作符的工作方式类似Javascript。

这是一个演示..。

const initialPeople = ['Person 1', 'Person 2', 'Person 3', 'Person 4'];

const initialPeople$ = Rx.Observable.from(initialPeople);

const addPeople = ['Person 5', 'Person 6', 'Person 7'];

const addPeople$ = Rx.Observable.from(addPeople)
            .concatMap(x => Rx.Observable.of(x).delay(1000)); // this just makes it async

const removePeople = ['Person 2x', 'Person 4x'];

const removePeople$ = Rx.Observable.from(removePeople)
                                              .delay(5000)
                                                .concatMap(x => Rx.Observable.of(x).delay(1000));

const mergedStream$ = Rx.Observable.merge(initialPeople$, addPeople$, removePeople$)

mergedStream$
  .scan((acc, stream) => {
        if (stream.includes('x') && acc.length > 0) {
            const index = acc.findIndex(person => person === stream.replace('x', ''))
            acc.splice(index, 1);
        } else {
            acc.push(stream);
        }
      return acc;
  }, [])
  .subscribe(x => console.log(x))

// In the end, ["Person 1", "Person 3", "Person 5", "Person 6", "Person 7"]

http://jsbin.com/rozetoy/edit?js,console

你没提到你的数据结构。我使用"x“作为标志是有点(很多)笨重和问题。但是我认为你知道如何修改扫描操作符来适应你的数据。