Angular 4 RxJs WebSocket

Question

我正在尝试了解如何使用 angular 4 和 websockets。在我使用 angularjs (1.x) 的旧项目中，我遵循了本指南： AngularJs Websocket Service

本指南中的解决方案是将每个请求的 Promise 和 Id 存储在集合中，因此当服务器发送回具有此 id 的响应时，我可以从集合中检索正确的 Promise 并选择使用“通知" 如果我​​期待更多消息，如果是一次性请求响应，则 "resolve"。

在 Angular 4 中最好使用 RxJs Observable / Subject。可以在这篇文章中找到 websocket 的基本用法：Angular (2+) Websockets Tutorial。 但是我怎样才能从第一篇文章中用 RxJs 实现等效的机制呢？

这是我的起点Service，它是一个简单的回显客户端：

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 要求是更好的 WebSocket 应用程序。

我建议您多播 (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