Rx.Observable.webSocket() 重新连接后立即完成？

Question

在使用Rx.Observable.webSocket 公开的Subject 时遇到了一些麻烦。虽然 WebSocket 在complete 之后确实重新连接，但对Subject 的后续订阅也会立即完成，而不是推送通过套接字发送的下一条消息。

我认为我遗漏了一些关于它应该如何工作的基本知识。

这是一个 requirebin/paste，我希望能更好地说明我的意思，以及我所期望的行为。认为这将是我忽略的超级简单的事情。

Requirebin

var Rx = require('rxjs')

var subject = Rx.Observable.webSocket('wss://echo.websocket.org')

subject.next(JSON.stringify('one'))

subject.subscribe(
    function (msg) {
        console.log('a', msg)
    }, 
    null,
    function () {
        console.log('a complete')
    }
)

setTimeout(function () {
    subject.complete()
}, 1000)

setTimeout(function () {
    subject.next(JSON.stringify('two'))
}, 3000)

setTimeout(function () {
  subject.next(JSON.stringify('three'))

  subject.subscribe(
      function (msg) {
          // Was hoping to get 'two' and 'three'
          console.log('b', msg)
      }, 
      null,
      function () {
          // Instead, we immediately get here.
          console.log('b complete')
      }
  )
}, 5000)

Answer 1

另一个巧妙的解决方案是在WebSocketSubject 上使用包装器。

class RxWebsocketSubject<T> extends Subject<T> {
  private reconnectionObservable: Observable<number>;
  private wsSubjectConfig: WebSocketSubjectConfig;
  private socket: WebSocketSubject<any>;
  private connectionObserver: Observer<boolean>;
  public connectionStatus: Observable<boolean>;

  defaultResultSelector = (e: MessageEvent) => {
    return JSON.parse(e.data);
  }

  defaultSerializer = (data: any): string => {
    return JSON.stringify(data);
  }

  constructor(
    private url: string,
    private reconnectInterval: number = 5000,
    private reconnectAttempts: number = 10,
    private resultSelector?: (e: MessageEvent) => any,
    private serializer?: (data: any) => string,
    ) {
    super();

    this.connectionStatus = new Observable((observer) => {
      this.connectionObserver = observer;
    }).share().distinctUntilChanged();

    if (!resultSelector) {
      this.resultSelector = this.defaultResultSelector;
    }
    if (!this.serializer) {
      this.serializer = this.defaultSerializer;
    }

    this.wsSubjectConfig = {
      url: url,
      closeObserver: {
        next: (e: CloseEvent) => {
          this.socket = null;
          this.connectionObserver.next(false);
        }
      },
      openObserver: {
        next: (e: Event) => {
          this.connectionObserver.next(true);
        }
      }
    };
    this.connect();
    this.connectionStatus.subscribe((isConnected) => {
      if (!this.reconnectionObservable && typeof(isConnected) == "boolean" && !isConnected) {
        this.reconnect();
      }
    });
  }

  connect(): void {
    this.socket = new WebSocketSubject(this.wsSubjectConfig);
    this.socket.subscribe(
      (m) => {
        this.next(m);
      },
      (error: Event) => {
        if (!this.socket) {
          this.reconnect();
        }
      });
  }

  reconnect(): void {
    this.reconnectionObservable = Observable.interval(this.reconnectInterval)
      .takeWhile((v, index) => {
        return index < this.reconnectAttempts && !this.socket
    });
    this.reconnectionObservable.subscribe(
      () => {
        this.connect();
      },
      null,
      () => {
        this.reconnectionObservable = null;
        if (!this.socket) {
          this.complete();
          this.connectionObserver.complete();
        }
      });
  }

  send(data: any): void {
    this.socket.next(this.serializer(data));
  }
}

更多信息请参考以下文章和源代码：
Auto WebSocket reconnection with RxJS
GitHub - Full working rxjs websocket example

Answer 2

我最终没有使用 Rx.Observable.webSocket，而是选择了observable-socket 和一些代码来在套接字关闭后重新连接：

requirebin

const observableSocket = require('observable-socket')
const Rx = require('rxjs')
const EventEmitter = require('events')

function makeObservableLoop (socketEmitter, send, receive) {
    socketEmitter.once('open', function onSocketEmit (wSocket) {
        const oSocket = observableSocket(wSocket)
        const sendSubscription = send.subscribe(msg => oSocket.next(msg))

        oSocket.subscribe(
            function onNext (msg) {
                receive.next(msg)
            },

            function onError (err) {
                error(err)
                sendSubscription.unsubscribe()

                makeObservableLoop(socketEmitter, send, receive)
            },

            function onComplete () {
                sendSubscription.unsubscribe()

                makeObservableLoop(socketEmitter, send, receive)
            }
        )
    })
}

function makeSocketLoop (emitter) {
  const websocket = new WebSocket('wss://echo.websocket.org')

  function onOpen () {
    emitter.emit('open', websocket)
    
    setTimeout(function () {
      websocket.close()
    }, 5000)
  }

  function onClose () {
    makeSocketLoop(emitter)
  }
  
  websocket.onopen = onOpen
  websocket.onclose = onClose
}

function init (socketEmitter) {
    const _send = new Rx.Subject()
    const _receive = new Rx.Subject()

    makeObservableLoop(socketEmitter, _send, _receive)

    const send = msg => _send.next(JSON.stringify(msg))
    const receive = _receive.asObservable()

    return {
        send: send,
        read: receive,
    }
}

const emitter = new EventEmitter()

makeSocketLoop(emitter)
const theSubjectz = init(emitter)

setInterval(function () {
  theSubjectz.send('echo, you there?')
}, 1000)

theSubjectz.read.subscribe(function (el) {
  console.log(el)
})