import { Injectable } from '@angular/core'; import * as Rx from "rxjs/Rx"; import { Observable, Subject } from "rxjs/Rx"; @Injectable({ providedIn: 'root' }) export class WebsocketService { message = ''; public messages: Subject; currentUser = ''; private subject: Rx.Subject; constructor() { } public connect(url): Rx.Subject { if (!this.subject) { this.subject = this.create(url); } return this.subject; } private create(url): Rx.Subject { let ws = new WebSocket(url); let observable = Rx.Observable.create((obs: Rx.Observer) => { ws.onmessage = obs.next.bind(obs); ws.onerror = obs.error.bind(obs); ws.onclose = obs.complete.bind(obs); return ws.close.bind(ws); }); let observer = { next: (data: Object) => { if (ws.readyState === WebSocket.OPEN) { ws.send(JSON.stringify(data)); } } }; return Rx.Subject.create(observer, observable); } }