使用 RxJS 将初始流与修改流结合起来答案

【问题标题】：Combine initial stream with modify streams with RxJS使用 RxJS 将初始流与修改流结合起来
【发布时间】：2016-06-29 13:01:29
【问题描述】：

我正在尝试考虑功能反应性并将页面上的元素列表设置为流。这是使用 Angular2，但问题应该类似于任何基于流的架构。因此，我目前有两个流，初始流（从 github 获取用户列表的 http 调用）和删除用户流（单击删除用户按钮时发生）。我相信大理石图如下所示：

|[user1,user2,user3]|                       <--- http initial stream
|---------------------x----------x-----...  <--- x denotes user removed

如何组合这些流以使其正常工作？我也在考虑以后有更多的流进行排序和排序。我会以正确的方式解决这个问题吗？这是代码（注意此代码不完整，目前 removeUser$ 不应该与 user$ 交互）：

export class UserGridComponent implements OnInit { 
    public users$: Observable<any>;
    public removeUser$: Subject;

    constructor(private _githubUserService: GithubUserService) { }

    ngOnInit() { 
       this.removeUser$ = new Subject()
           .subscribe((user) => { console.log('next ' + JSON.stringify(user)});
       this.users$ = this._githubUserService.getUsers()
           .map((res) => res.json());
    }
}

这里是Plunker

目前我只记录到控制台单击删除按钮并传递给用户。

这是显示我使用异步管道 (Angular2) 订阅 user$ 的 html 模板：

<md-list>
  <h1 md-header>GitHub Users</h1>
  <md-list-item *ngFor="let user of users$ | async">
    <a href="https://github.com/{{user.login}}" target="_new">
      <img md-list-avatar [src]="user.avatar_url">
    </a>
    <h4>{{user.login}}</h4>
    <button md-icon-button (click)="removeUser$.next(user)">
      <md-icon>cancel</md-icon>
    </button>
  </md-list-item>
</md-list>

【问题讨论】：

我不是 100% 清楚你想要实现什么，但在你的代码中你从未订阅过 users$ 可观察对象，所以它可能仍然很冷。
我确实使用 angular2 异步管道订阅了模板中可观察到的 user$。每当发生 removeUser$ 事件时，我都会尝试从 user$ 中删除用户。我也发一下模板，在plunker里，谢谢。

标签： angular reactive-programming rxjs

【解决方案1】：

ngOnInit() { 
   this.users$ = new BehaviorSubject([]); // init with empty user array
   this._githubUserService.getUsers()
       .map((res) => res.json())
       .subscribe(this.users$);

   this.removeUser$ = new Subject();
   this.removeUser$.subscribe((user) => { 
       this.users$.take(1)  // get current users, as users$ is a BehaviorSubject
           .map(users => users.filter(u => u != user))   // remove `user`
           .subscribe(this.users$); // update users$ stream
   });
}

希望我的cmets可以作为解释。 BehaviorSubject 缓存最后发出的值，所以 this.users$.take(1) 是同步的。我发现自己使用 BehaviorSubject 作为 angular2 ' |异步管道非常频繁。

编辑：removeUser$ 部分的想法相同，但要短一些：

.....
   this.removeUser$.withLatestFrom(user$, 
       (toRemove, users) => users.filter(u => u != toRemove)   // remove `toRemove`
   ).subscribe(this.users$); // update users$ stream

【讨论】：

虽然这可能有效，但它似乎真的很混乱，而且不容易把我的头绕起来......我想知道在这个用例中是否最好保持状态（使用 redux）以便我们可以从流中删除.. 我的其他用例 - 过滤，排序 - 组合流会很好.. 或者如果有更好的方法可以在没有嵌入式用户流的情况下编写它
国营商店确实是这里缺少的部分。我明白您要做什么，不久前我也处于同一点。您确实可以使用纯 rxjs（使用扫描运算符）来做到这一点，但是使用像 ngxs 的 ngrx 这样的状态存储肯定会帮助您进行反应式编程。我会看看是否能找到相同的方法来使用扫描运算符创建一个示例：这是一个很好的运算符。但是你一定要试试ngrx或ngxs。

【解决方案2】：

看看这个 stackblitz：它展示了如何使用 scan 来改变从一个或多个源发出的事件的单个值（可以是任何值...）。在示例中，所有传入操作都来自 modify$ 流，但您不妨将 add$ 和 remove$ 流合并在一起。

https://stackblitz.com/edit/angular-evc48h

但是...我不建议这样做，因为这正是国营商店为您所做的。

所以一定要看看著名的ngrx： https://github.com/ngrx/platform

或者是鲜为人知但更干净的 ngxs： https://github.com/ngxs/store

我经常听到人们声称使用状态存储会引入不必要的复杂性，这对于小而简单的屏幕可能是正确的。但是，您的屏幕越复杂，国营商店就越能帮助您保持优雅和干净。

【讨论】：