角度Websocket RxJS/WebSocketSubject

Question

これまでいくつかのangular2 websocket impelmentationsを使用していましたが、私たちはそれに満足していませんでした。だから私たちはRxJsで私たちの財産でそれをやってみることにしました。

これが私たちの最初の試みである：

@Injectable() 
export class WebSocketService{ 

    public createWebsocket(url: string): Subject<MessageEvent> { 
     let socket = new WebSocket(url); 

     let observable = Observable.create(
            (observer: Observer<MessageEvent>) => { 
             socket.onmessage = observer.next.bind(observer); 
             socket.onerror = observer.error.bind(observer); 
             socket.onclose = observer.complete.bind(observer); 

             return socket.close.bind(socket); 
            }); 

     let observer = { 
      next: (data: Object) => { 
       if (socket.readyState === WebSocket.OPEN) { 
        socket.send(JSON.stringify(data)); 
       } 
      } 
     }; 

     return Subject.create(observer, observable); 
    } 
} 

ソケットが開かれ、しばらくの間、うまく機能しています。数秒後にブラウザがソケットを閉じて、サーバー側で終了イベントを取得します。

これは、サーバーのサイトで入手決算理由です： [1006] WebSocketを読むEOF

は、誰でも助けることができていますか？または、誰もWebSocketSubjectの使い方を知っていますか？以下のためのもの

Answer 1

それはまだ関連しているが、私はhttps://github.com/ohjames/rxjs-websockets

私はServerSocketServiceを呼び出したコンポーネント内に行われた修正の一部を持つ角度からのWebSocket接続用の同様のものを使用しなかった場合、私は知っている `tを（ WebSocketベースのエンドポイントに接続しています）つまり、再試行の仕組みと、この例ではQueuingSubjectの代わりにReplaySubjectを使用しました。

@Injectable() 
export class ServerSocket { 
    // private inputStream: QueueingSubject<string>; 
    private inputStream: ReplaySubject<string>; 
    public messages: Observable<string>; 
    private subscription: Subscription; 
    private websocket: WebSocket; 

    public connect() { 
     if (this.messages) { 
      return; 
     } 
     console.log('inside connect'); 
     // this.inputStream = new QueueingSubject<string>(); 
     this.inputStream = new ReplaySubject(); 

     // Using share() causes a single websocket to be created when the first 
     // observer subscribes. This socket is shared with subsequent observers 
     // and closed when the observer count falls to zero. 
     this.messages = websocketConnect(
      'ws://localhost:9097/echo', 
      this.inputStream 
     ).messages.share(); 


     this.messages.retryWhen(errors => errors.delay(1000)).subscribe(message => { 
      console.log('error', message); 
     }); 
    } 

    public send(message: string): void { 
     // If the websocket is not connected then the QueueingSubject will ensure 
     // that messages are queued and delivered when the websocket reconnects. 
     // A regular Subject can be used to discard messages sent when the websocket 
     // is disconnected. 

     this.inputStream.next(message); 

    } 
} 

コンポーネントのOnInitライフサイクル内で、接続を確立し、購読してからメッセージを送信します。コンポーネントがOnDestroyライフサイクルに達すると、登録解除を行うだけでリソースが解放されます。