RxJs管道主题不发射值答案

【问题标题】：RxJs piped subject not emitting valueRxJs管道主题不发射值
【发布时间】：2018-06-18 14:25:51
【问题描述】：

我已经通过 RxJs 支持完成了一些 WebSockets 实现。

listen<T>(action: Action): Observable<T> {
    return this.subject
      .pipe(
          filter(value => value.action === action),
          map(value => {
              delete value.action;
              delete value.id;
              return value;
          })
      );
}

此方法返回 Subject 的管道实例，该实例在 websocket.onmessage 回调中发出所有到达 WebSocket 实例的消息。

ws.onmessage = ({data}) => this.subject.next(JSON.parse(data));

问题是我有两个组件使用这种方法来监听某种类型的消息，而我只能监听一个动作。例如，如果我有这样的东西：

this.websocketService.listen('incomingStream').subscribe(m => console.log(1));
this.websocketService.listen('incomingStream').subscribe(m => console.log(2));

我总是在控制台中得到 1。我想同时多次收听这个事件。

UPD.1

我用 "rxjs": "^6.2.1", "rxjs-compat": "^6.2.1",

UPD.2

我只是返回属性source，它开始正常工作

listen<T>(action: Action): Observable<T> {
    return this.subject
      .pipe(
        filter(value => value.action === action),
        map(value => {
          delete value.action;
          delete value.id;
          return value;
        })
      )
      .source;
  }

我随机做的，现在我不明白为什么它开始起作用了

【问题讨论】：

当我从listen 方法中删除filter 函数时效果很好

标签： angular websocket rxjs

【解决方案1】：

尝试使用share 运算符

listen<T>(action: Action): Observable<T> {
    return this.subject
      .pipe(
          filter(value => value.action === action),
          map(value => {
              delete value.action;
              delete value.id;
              return value;
          }),
           share()
      );
}

【讨论】：

尝试使用 ReplaySubject 代替普通主题
不起作用，这很有趣，因为我试图用简单的例子来演示这个问题，但是所有工作都完美codepen.io/looffee/pen/zapxZe?editors=1111
尝试 publishReplay(1), .refCount()
不起作用，但是当我在 listen 方法中删除 .pipe(...) 和仅 return this.subject; 时，所有观察者都会收到一条消息
那可能是rxjs版本。或者这个过滤器正在过滤所有请求

【解决方案2】：

只需返回属性源，它就开始正常工作了

listen<T>(action: Action): Observable<T> { return this.subject .pipe( filter(value => value.action === action), map(value => { delete value.action; delete value.id; return value; }) ) .source; }

【讨论】：