Angular 2在組件間共享websocket服務

Question

我正在構建一個使用angular 2的web應用程序，在這個應用程序中我想讓多個組件監聽同一個服務。這個服務返回一個可觀測值，從websocket返回輸入數據。我編寫了基於this示例的代碼。

現在的問題是： 的數據是從家庭成分通過（使用的WebSockets）服務發送到服務器並返回數據。但是，只有home.component中的觀察者纔會被調用（id：room.created和data），而不是導航欄中的那個。

有人能告訴我爲什麼不是兩個都叫？我還嘗試將消息$。訂閱添加到app.component，但無濟於事。

現在，讓我們來看看代碼。

返回可觀察對象的消息服務。組件使用此服務來發送和接收消息。

@Injectable() 
export class MessageService { 
    private _messages: Rx.Subject<Message>; 
    messages$: Rx.Observable<Message>; 

    constructor(wsService: SocketService, private configuration: Configuration) { 
     console.log('messag eservice'); 
     this._messages = <Rx.Subject<Message>>wsService 
     .connect() 
     .map((response: MessageEvent): Message => { 
      let data = JSON.parse(response.data); 
      return { 
       id: data.id, 
       data: data.data, 
      } 
     }); 

     this.messages$ = this._messages.asObservable(); 
    } 

    public send(message: Message): void { 
     this._messages.next(message); 
    } 
} 

一個套接字服務，它創建一個websocket連接並綁定自己以輸入和輸出這個套接字。

import { Injectable } from '@angular/core'; 
import * as Rx from "rxjs/Rx"; 
import { Configuration } from '../app.constants'; 

@Injectable() 
export class SocketService { 
    private subject: Rx.Subject<MessageEvent>; 

    constructor(private configuration: Configuration){}; 

    public connect(wsNamespace = ''): Rx.Subject<MessageEvent> { 
     var url = this.configuration.wsUrl + wsNamespace; 
     if(!this.subject) { 
      this.subject = this.create(url); 
     } 
     return this.subject; 
    } 

    private create(url): Rx.Subject<MessageEvent> { 
     let ws = new WebSocket(url); 

     // bind ws events to observable (streams) 
     let observable = Rx.Observable.create((obs: Rx.Observer<MessageEvent>) => { 
      ws.onmessage = obs.next.bind(obs); 
      ws.onerror = obs.error.bind(obs); 
      ws.onclose = obs.complete.bind(obs); 

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

     // on obs next (send something in the stream) send it using ws. 
     let observer = { 
      next: (data: Object) => { 
       if (ws.readyState === WebSocket.OPEN) { 
        ws.send(JSON.stringify(data)); 
       } 
      }, 
     }; 

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

的應用程序組件具有以下提供商：

providers: [MessageService, SocketService, Configuration, AuthService] 

我實例在我的主app.component的供應商，以確保郵件和插座服務不實例化兩次。

我home.component看起來是這樣的（這是一個網頁使用路由加載）：

import { Component, OnInit } from '@angular/core'; 
import { Subject } from 'rxjs'; 
import { Router } from '@angular/router'; 
import { MessageService } from '../../services/message.service'; 

@Component({ 
    ... 
    providers: [] 
}) 
export class HomeComponent implements OnInit { 
    constructor(private router: Router, private messageService: MessageService) {} 

    ngOnInit(): void { 
    this.messageService.send({ 
     id: 'room.create', 
     data: {'name': 'Blaat'} 
    }); 

    this.messageService.messages$.subscribe(msg => { 
     console.log(msg); 
     if(msg.id == 'room.created') { 
      // navigate naar games! 
     } 
    }); 
    } 

} 

我的導航欄組件看起來像這樣（指令）：

import { Component, OnInit } from '@angular/core'; 
import { MessageService } from '../../services/message.service'; 

@Component({ 
    moduleId: module.id, 
    selector: 'navbar', 
    templateUrl: 'navbar.component.html', 
    styleUrls: ['navbar.component.css'] 
}) 
export class Navbar implements OnInit { 

    constructor(private messageService: MessageService) { } 

    ngOnInit() { 

    this.messageService.messages$.subscribe(msg => { 
     console.log(msg); 
     if(msg.id == 'room.created') { 
      // navigate naar games! 
     } 
    }); 
    } 

} 

Answer 1

看來你可觀察的創建函數被多次調用，最有可能是兩個組件=>兩個訂閱=>兩個可觀察的創建函數調用。所以最新的可觀察的創建fn覆蓋了先前可觀察到的對websocket onmessage，onerror和onclose的回調。你應該多播基礎的observable，以防止（共享操作員應該做的伎倆）。

 // bind ws events to observable (streams) 
     let observable = Rx.Observable.create((obs: Rx.Observer<MessageEvent>) => { 
      ws.onmessage = obs.next.bind(obs); 
      ws.onerror = obs.error.bind(obs); 
      ws.onclose = obs.complete.bind(obs); 

      return ws.close.bind(ws); 
     }).share(); 

更多有用的如何做到這一點的資源妥善 https://github.com/ReactiveX/rxjs/blob/master/src/observable/dom/WebSocketSubject.ts https://github.com/blesh/RxSocketSubject