import { Injectable } from '@angular/core' ; import { Subject } from 'rxjs/Subject' ; import { Subscription } from 'rxjs/Subscription' ; import { Observable } from 'rxjs/Observable' ; import { Observer } from 'rxjs/Observer' ; import 'rxjs/add/operator/map' ; import 'rxjs/add/operator/filter' ; import 'rxjs/add/observable/dom/webSocket' ; const CHAT_URL = 'ws://localhost:8080/chat' ; interface Frame { type : string ; data : any ; } ; enum MessageType { Chat = 1 , Nick , Error } export interface Message { msgType ( ) : MessageType ; toString ( ) : string ; } export class ChatMessage implements Message { author : string ; body : string ; msgType ( ) { return MessageType . Chat ; } toString ( ) { return this . author + ': ' + this . body ; } } export class NickMessage implements Message { oldName : string ; newName : string ; msgType ( ) { return MessageType . Nick ; } toString ( ) { if ( this . newName === '' && this . oldName !== '' ) { return this . oldName + ' has left the channel' ; } else if ( this . oldName !== '' ) { return this . oldName + ' has changed nick to ' + this . newName ; } else { return this . newName + ' has joined the channel' ; } } } export class ErrorMessage implements Message { errorCode : string ; message : string ; msgType ( ) { return MessageType . Error ; } toString ( ) { return 'Error (' + this . errorCode + '): ' + this . message ; } } @ Injectable ( ) export class ChatService { public messages : Observable < Message > ; private ws : Subject < any > ; constructor ( ) { this . ws = Observable . webSocket ( CHAT_URL ) ; this . messages = makeHot ( this . ws ) . map ( parseFrame ) . filter ( m => m != null ) ; } sendNickMessage ( msg ) { let frame = { type : 'NickMessage' , data : { new_name : msg . newName , } , } ; this . ws . next ( JSON . stringify ( frame ) ) ; } sendChatMessage ( msg ) { let frame = { type : 'ChatMessage' , data : { body : msg . body , } , } ; this . ws . next ( JSON . stringify ( frame ) ) ; } } function parseFrame ( frame ) { if ( frame . type === 'ChatMessage' ) { let msg = new ChatMessage ( ) ; msg . author = frame . data . author ; msg . body = frame . data . body ; return msg ; } else if ( frame . type === 'NickMessage' ) { let msg = new NickMessage ( ) ; msg . oldName = frame . data . old_name ; msg . newName = frame . data . new_name ; return msg ; } else if ( frame . type === 'ErrorMessage' ) { let msg = new ErrorMessage ( ) ; msg . errorCode = frame . data . error_code ; msg . message = frame . data . message ; return msg ; } return null ; } function makeHot < T > ( cold ) { let subject = new Subject ( ) ; let refs = 0 ; return Observable . create ( ( observer ) => { let coldSub ; if ( refs === 0 ) { coldSub = cold . subscribe ( o => subject . next ( o ) ) ; } refs ++ ; let hotSub = subject . subscribe ( observer ) ; return ( ) => { refs -- ; if ( refs === 0 ) { coldSub . unsubscribe ( ) ; } hotSub . unsubscribe ( ) ; } ; } ) ; }