Angular 4 RxJs WebSocket

Question

I'm trying to understand how to work with angular 4 and websockets. In my older projects with angularjs (1.x) I followed this guide: AngularJs Websocket Service

The solution in this guide is to store a Promise and an Id for each request inside a collection, so when the server sends back a response with this id, I can retrieve the right Promise from the collection and choose to use "notify" if I expect more messages, and "resolve" if is a one shot request-response.

In Angular 4 is preferable to work with RxJs Observable / Subject. A basic use of websocket can be found in this post: Angular (2+) Websockets Tutorial. But how can I achieve an equivalent mechanism from the first post with RxJs?

This is my starting point Service, it is a simple echo client:

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);
  }

}

gwest7 gwest7 · Accepted Answer · 2017-10-10T15:25:59

The first post is making single-request-single-response calls via a socket. I would really suggest to the author to just use HTTP. Your notify requirement is a better application for WebSockets.

I would suggest you multicast (share()) the observable and whoever needs receive data from the WebSocket subscribe. Here is an example with Angular components.

@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
    }
}

And then components subscribe and unsubscribe.

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
        });
    }

}

For more WebSocket RxJs info here is a great video by the legend himself and the source code

Angular 4 RxJs WebSocket

1 Answers