socket_connector.js

import Backoff from 'backo2';

const BrowserWebSocket = global.WebSocket || global.MozWebSocket;

const DEF_PING_TIMEOUT = 2 * 60 * 1000;
const PONG = { op: 'PONG' };

export default class Socket {
  constructor(opts = {}) {
    this.uri = this.buildUri(opts);
    this._debug = !!opts.debug;
    this.token = null;
    this.readyState = null;
    this.writable = false;
    this.buffer = [];
    this._pingTimeout = opts.pingTimeout === undefined ? DEF_PING_TIMEOUT : opts.pingTimeout;
    this._reconnection = opts.reconnection !== false;
    this._skipReconnect = false;
    this._reconnectionAttempts = opts.reconnectionAttempts || Infinity;
    this._reconnectionDelay = opts.reconnectionDelay || 1000;
    this._reconnectionDelayMax = opts.reconnectionDelayMax || 5000;
    this._randomizationFactor = opts.randomizationFactor || 0.5;
    this.backoff = new Backoff({
      min: this._reconnectionDelay,
      max: this._reconnectionDelayMax,
      jitter: this._randomizationFactor,
    });

    const eventTypes = 'open,close,error,message,ping,reconnect';
    this.events = eventTypes.split(',').reduce((events, ev) => {
      // eslint-disable-next-line no-param-reassign
      events[ev] = [];
      return events;
    }, {});

    if (opts.autoOpen !== false && BrowserWebSocket) this.open();
  }

  debug(...args) {
    // eslint-disable-next-line no-console
    if (this._debug) console.error(args);
  }

  on(event, listener) {
    if (!this.events[event]) return false;
    this.events[event].push(listener);
    return true;
  }

  once(event, listener) {
    const wrap = (...args) => {
      this.removeListener(event, wrap);
      listener(...args);
    };

    this.on(event, wrap);
  }

  emit(event, ...args) {
    if (!this.events[event]) return false;
    if (this.events[event].length) {
      this.events[event].forEach((listener) => {
        listener(...args);
      });
    }
    return true;
  }

  removeListener(event, listener) {
    if (!this.events[event] || !this.events[event].length) return false;
    this.events[event] = this.events[event].filter(l => l !== listener);
  }

  buildUri({ protocol: _protocol, host: _host, port: _port, path: _path }) {
    const loc = global.location;
    const protocol = _protocol || (loc.protocol === 'https:' ? 'wss' : 'ws');
    const host = _host || loc.hostname || '';
    let port = _port || loc.port || '';

    if (protocol === 'wss' && port === 443 || protocol === 'ws' && port === 80) {
      port = '';
    }

    let path = _path || '/';
    if (path[0] !== '/') path = `/${path}`;

    return `${protocol}://${host}${port ? ':' : ''}${port}${path}`;
  }

  open() {
    if (this.readyState === 'closed' || !this.readyState) {
      this.readyState = 'opening';
      this.ws = new BrowserWebSocket(this.uri);
      this.addEventListeners();
    }
  }

  addEventListeners() {
    this.ws.onopen = () => {
      this.onOpen();
    };
    this.ws.onclose = (ev) => {
      this.onClose(ev.code);
    };
    this.ws.onmessage = (ev) => {
      this.onMessage(ev.data);
    };
    this.ws.onerror = (err) => {
      this.onError(err);
    };
  }

  setPingTimeout() {
    if (!this._pingTimeout) return;
    clearTimeout(this._pingTimer);
    this._pingTimer = setTimeout(() => {
      this.ws.close();
    }, this._pingTimeout);
  }

  close() {
    if (this.ws && this.readyState !== 'closed') {
      this._skipReconnect = true;
      this.ws.close();
    }
  }

  onOpen() {
    this.readyState = 'open';
    this.writable = true;
    this._skipReconnect = false;
    this.emit('open');
    this.setPingTimeout();
    this.flush();
  }

  onClose(reason) {
    this.readyState = 'closed';
    this.writable = false;
    this.backoff.reset();
    clearTimeout(this._pingTimer);
    this.emit('close', reason);

    if (this._reconnection) {
      this.reconnect();
    }
  }

  onError(e) {
    const err = new Error('websocket error');
    err.type = 'TransportError';
    err.description = e;
    this.writable = false;
    this.readyState = 'closed';
    clearTimeout(this._pingTimer);
    this.emit('error', err);

    if (this._reconnection) {
      this.reconnect();
    }
  }

  onMessage(_data) {
    let data = _data;
    if (typeof data === 'string') {
      try {
        data = JSON.parse(data);
      } catch (e) {
        this.debug('Websocket data was not JSON.');
        return;
      }

      if (data.op === 'PING') {
        this.setPingTimeout();
        this.send(PONG);
        this.emit('ping');
        return;
      }

      this.emit('message', data);
    }
  }

  send(msg) {
    if (!this.ws || !this.readyState) return false;

    if (!this.writable) {
      if (msg !== PONG) this.buffer.push(msg);
      return;
    }

    const str = JSON.stringify(msg);
    this.doWrite(str);
  }

  doWrite(packet) {
    // Sometimes the websocket closes but this is hit before the browser
    // has a chance to inform us so we need to wrap this in a try/catch.
    try {
      this.ws.send(packet);
    } catch (e) {
      this.debug('Websocket closed before onclose event.');
    }
  }

  flush() {
    if (this.readyState !== 'open' || !this.writable) return;
    const buffer = this.buffer;
    this.buffer = [];
    buffer.forEach((msg) => {
      this.send(msg);
    });
  }

  reconnect() {
    if (!this._reconnection ||
        this._skipReconnect ||
        this._reconnecting ||
        this.readyState !== 'closed') {
      return;
    }

    clearTimeout(this._pingTimer);
    clearTimeout(this._reconnectTimer);

    if (this.backoff.attempts >= this._reconnectionAttempts) {
      this.backoff.reset();
      this._reconnection = false;
      this.debug('Websocket reconnect failed.');
      return;
    }

    const delay = this.backoff.duration();
    this._reconnecting = true;
    const timer = setTimeout(() => {
      if (!this._reconnection || this._skipReconnect) return;

      this.emit('reconnect', this.backoff.attempts);
      this.open();

      const handleOpen = () => {
        this._reconnecting = false;
        clearTimeout(this._reconnectTimer);
        this.backoff.reset();
      };

      const handleError = () => {
        this._reconnecting = false;
        this.reconnect();
      };

      this.once('open', handleOpen);
      this.once('error', handleError);
    }, delay);

    this._reconnectTimer = timer;
  }
}