Rx.Observable.webSocket（）在重新連接後立即完成？

Question

與Rx.Observable.webSocket公開的Subject有點麻煩。雖然WebSocket在complete之後確實重新連接，但後續對Subject的訂閱也會立即完成，而不會推送通過套接字發送的下一條消息。

我想我錯過了一些基本的關於這應該如何工作。

下面是一個requirebin /貼圖，希望能更好地說明我的意思，以及我期待的行爲。認爲這將是一個超級簡單的東西，我忽略了。

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

我結束了不使用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) 
 
})

Answer 2

另一個巧妙的解決方案是使用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