RXJS 为套接字创建 Observable答案

【问题标题】：RXJS Create Observable for socketsRXJS 为套接字创建 Observable
【发布时间】：2018-12-13 05:40:03
【问题描述】：

我正在尝试从允许您收听任何 Pusher 频道的服务返回一个 Observable。

工作正常，希望在取消订阅 Observable（使用 createRealtimeObservable(...) 生成）时，我想从推送器频道发送 unsubscribe。

有什么想法吗？

  public realtimeObservable (channelName, eventName): Observable<any> {

        const realtimeObservable$ = new Observable((observer) => {

          const channel = this.pusher.subscribe(`private-organization-${this.authService.userProfile.organization_id}-${channelName}`)
          channel.bind(eventName, (data) => {
            observer.next(data.payload)
          })

        })

        return realtimeObservable$
      }

【问题讨论】：

标签： sockets rxjs reactive-programming pusher

【解决方案1】：

当您自己手动创建 Observable 时，您有责任在取消订阅时清理所有资源。幸运的是，Observable 有一个机制，允许您在创建时返回取消订阅/处置函数：

return Observable.create((obs) => {
  const channel = this.pusher.subscribe(`private-organization-${this.authService.userProfile.organization_id}-${channelName}`)
  channel.bind(eventName, (data) => {
    observer.next(data.payload)
  });

  return () => {
    // unsubscribe event
    channel.unsubscribe(); // NOTE: i do not know the syntax for unsubscribing a Pusher channel, impl as required.
  };
});

注意：Observable.create() 是 new Observable() 的语法糖

【讨论】：