| import { Subject, AnonymousSubject } from '../../Subject'; |
| import { Subscriber } from '../../Subscriber'; |
| import { Observable } from '../../Observable'; |
| import { Subscription } from '../../Subscription'; |
| import { ReplaySubject } from '../../ReplaySubject'; |
| const DEFAULT_WEBSOCKET_CONFIG = { |
| url: '', |
| deserializer: (e) => JSON.parse(e.data), |
| serializer: (value) => JSON.stringify(value), |
| }; |
| const WEBSOCKETSUBJECT_INVALID_ERROR_OBJECT = 'WebSocketSubject.error must be called with an object with an error code, and an optional reason: { code: number, reason: string }'; |
| export class WebSocketSubject extends AnonymousSubject { |
| constructor(urlConfigOrSource, destination) { |
| super(); |
| if (urlConfigOrSource instanceof Observable) { |
| this.destination = destination; |
| this.source = urlConfigOrSource; |
| } |
| else { |
| const config = this._config = Object.assign({}, DEFAULT_WEBSOCKET_CONFIG); |
| this._output = new Subject(); |
| if (typeof urlConfigOrSource === 'string') { |
| config.url = urlConfigOrSource; |
| } |
| else { |
| for (let key in urlConfigOrSource) { |
| if (urlConfigOrSource.hasOwnProperty(key)) { |
| config[key] = urlConfigOrSource[key]; |
| } |
| } |
| } |
| if (!config.WebSocketCtor && WebSocket) { |
| config.WebSocketCtor = WebSocket; |
| } |
| else if (!config.WebSocketCtor) { |
| throw new Error('no WebSocket constructor can be found'); |
| } |
| this.destination = new ReplaySubject(); |
| } |
| } |
| lift(operator) { |
| const sock = new WebSocketSubject(this._config, this.destination); |
| sock.operator = operator; |
| sock.source = this; |
| return sock; |
| } |
| _resetState() { |
| this._socket = null; |
| if (!this.source) { |
| this.destination = new ReplaySubject(); |
| } |
| this._output = new Subject(); |
| } |
| multiplex(subMsg, unsubMsg, messageFilter) { |
| const self = this; |
| return new Observable((observer) => { |
| try { |
| self.next(subMsg()); |
| } |
| catch (err) { |
| observer.error(err); |
| } |
| const subscription = self.subscribe(x => { |
| try { |
| if (messageFilter(x)) { |
| observer.next(x); |
| } |
| } |
| catch (err) { |
| observer.error(err); |
| } |
| }, err => observer.error(err), () => observer.complete()); |
| return () => { |
| try { |
| self.next(unsubMsg()); |
| } |
| catch (err) { |
| observer.error(err); |
| } |
| subscription.unsubscribe(); |
| }; |
| }); |
| } |
| _connectSocket() { |
| const { WebSocketCtor, protocol, url, binaryType } = this._config; |
| const observer = this._output; |
| let socket = null; |
| try { |
| socket = protocol ? |
| new WebSocketCtor(url, protocol) : |
| new WebSocketCtor(url); |
| this._socket = socket; |
| if (binaryType) { |
| this._socket.binaryType = binaryType; |
| } |
| } |
| catch (e) { |
| observer.error(e); |
| return; |
| } |
| const subscription = new Subscription(() => { |
| this._socket = null; |
| if (socket && socket.readyState === 1) { |
| socket.close(); |
| } |
| }); |
| socket.onopen = (e) => { |
| const { _socket } = this; |
| if (!_socket) { |
| socket.close(); |
| this._resetState(); |
| return; |
| } |
| const { openObserver } = this._config; |
| if (openObserver) { |
| openObserver.next(e); |
| } |
| const queue = this.destination; |
| this.destination = Subscriber.create((x) => { |
| if (socket.readyState === 1) { |
| try { |
| const { serializer } = this._config; |
| socket.send(serializer(x)); |
| } |
| catch (e) { |
| this.destination.error(e); |
| } |
| } |
| }, (e) => { |
| const { closingObserver } = this._config; |
| if (closingObserver) { |
| closingObserver.next(undefined); |
| } |
| if (e && e.code) { |
| socket.close(e.code, e.reason); |
| } |
| else { |
| observer.error(new TypeError(WEBSOCKETSUBJECT_INVALID_ERROR_OBJECT)); |
| } |
| this._resetState(); |
| }, () => { |
| const { closingObserver } = this._config; |
| if (closingObserver) { |
| closingObserver.next(undefined); |
| } |
| socket.close(); |
| this._resetState(); |
| }); |
| if (queue && queue instanceof ReplaySubject) { |
| subscription.add(queue.subscribe(this.destination)); |
| } |
| }; |
| socket.onerror = (e) => { |
| this._resetState(); |
| observer.error(e); |
| }; |
| socket.onclose = (e) => { |
| this._resetState(); |
| const { closeObserver } = this._config; |
| if (closeObserver) { |
| closeObserver.next(e); |
| } |
| if (e.wasClean) { |
| observer.complete(); |
| } |
| else { |
| observer.error(e); |
| } |
| }; |
| socket.onmessage = (e) => { |
| try { |
| const { deserializer } = this._config; |
| observer.next(deserializer(e)); |
| } |
| catch (err) { |
| observer.error(err); |
| } |
| }; |
| } |
| _subscribe(subscriber) { |
| const { source } = this; |
| if (source) { |
| return source.subscribe(subscriber); |
| } |
| if (!this._socket) { |
| this._connectSocket(); |
| } |
| this._output.subscribe(subscriber); |
| subscriber.add(() => { |
| const { _socket } = this; |
| if (this._output.observers.length === 0) { |
| if (_socket && _socket.readyState === 1) { |
| _socket.close(); |
| } |
| this._resetState(); |
| } |
| }); |
| return subscriber; |
| } |
| unsubscribe() { |
| const { _socket } = this; |
| if (_socket && _socket.readyState === 1) { |
| _socket.close(); |
| } |
| this._resetState(); |
| super.unsubscribe(); |
| } |
| } |
| //# sourceMappingURL=WebSocketSubject.js.map |