Angular 4 RxJs WebSocket

Question

我想了解如何使用angular 4和websockets。在我與angularjs（1.x的）舊項目我遵循這個指南： AngularJs Websocket Service

本指南中的解決方案是存儲並承諾爲集合內的每個請求的ID，所以當服務器發回一個響應使用這個ID，我可以從集合中檢索正確的Promise，並且如果我期望更多的消息，則選擇使用「notify」;如果是單次請求響應，則選擇「resolve」。

在Angular 4中最好與RxJs Observable/Subject一起工作。 websocket的基本使用可以在這篇文章中找到：Angular (2+) Websockets Tutorial。 但是我怎樣才能實現RxJs的第一篇文章的等價機制？

這是我的出發點服務，這是一個簡單的echo客戶端：

import { Injectable } from '@angular/core'; 
import { Observable, Subject } from 'rxjs/Rx'; 
import * as Rx from 'rxjs/Rx'; 

@Injectable() 
export class WebsocketService { 

    private socketUrl: any = 'wss://echo.websocket.org'; 
    private subject: Rx.Subject<MessageEvent>;  
    private ws : any; 
    public messages: Subject<any>; 

    constructor() { 
     this.messages = <Subject<any>>this.connect() 
      .map((response: MessageEvent): any => { 
       let data = JSON.parse(response.data); 
       return data; 
      }); 
    } 

    connect() : Rx.Subject<MessageEvent> { 
    if (!this.subject) { 
     this.subject = this.create(this.socketUrl); 
     console.log("Successfully connected: " + this.socketUrl); 
    } 
    return this.subject; 
    } 

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

    let observable = Rx.Observable.create(
    (obs: Rx.Observer<MessageEvent>) => { 
     this.ws.onmessage = obs.next.bind(obs); 
     this.ws.onerror = obs.error.bind(obs); 
     this.ws.onclose = obs.complete.bind(obs); 
     return this.ws.close.bind(this.ws); 
    }) 
    let observer = { 
     next: (data: Object) => { 
      if (this.ws.readyState === WebSocket.OPEN) { 
       this.ws.send(JSON.stringify(data)); 
      } 
     } 
    } 
    return Rx.Subject.create(observer, observable); 
    } 

} 

Answer 1

的第一篇文章是通過插座進行單請求單響應呼叫。我真的建議作者只使用HTTP。您的notify要求是WebSockets的更好的應用程序。

我建議你多播（share()）觀察者和誰需要從WebSocket訂閱接收數據。這裏是一個Angular組件的例子。

@Injectable() 
export class ChatService { 
    public messages: Subject<Message>; 
    public shared: Observable<Message>; 

    constructor(wsService: WebsocketService) { 
     this.messages = <Subject<Message>>wsService 
      .connect(CHAT_URL) 
      .map((response: MessageEvent): Message => { 
       let data = JSON.parse(response.data); 
       return { 
        author: data.author, 
        message: data.message 
       } 
      }); 
     this.shared = this.messages.share(); //multicast 
    } 
} 

然後組件訂閱和退訂。

export class SomeComponent { 

    private componentId:number = this.getUniqueId(); 
    private subscription; 

    constructor(private chatService: ChatService) { } 

    ngOnInit(){ 
     this.subscription = this.chatService.shared.subscribe(msg => { //subscribe to multicast obs. 
      console.log("Response from websocket: " + msg); 
      if (msg.comp_id == this.componentId) { //comp_id returned by server 
       //message I am interested in 
      } 
     }); 
    } 
    ngOnDestroy(){ 
     this.subscription.unsubscribe(); 
    } 

    private message = { 
     componentId: this.componentId, 
     messageId:this.getUniqueId(), 
     author: 'tutorialedge', 
     message: 'this is a test message' 
    } 

    sendMsg(message:string) { 
     this.chatService.messages.next({ 
      componentId: this.componentId, 
      messageId: this.getUniqueId(), 
      author: 'tutorialedge',//? 
      message: message 
     }); 
    } 

} 

對於這裏更多的WebSocket RxJs信息是通過great video傳說他和你已經嘗試什麼source code