@horizon/client/lib/socket.js

'use strict';

var _classCallCheck2 = require('babel-runtime/helpers/classCallCheck');

var _classCallCheck3 = _interopRequireDefault(_classCallCheck2);

var _possibleConstructorReturn2 = require('babel-runtime/helpers/possibleConstructorReturn');

var _possibleConstructorReturn3 = _interopRequireDefault(_possibleConstructorReturn2);

var _inherits2 = require('babel-runtime/helpers/inherits');

var _inherits3 = _interopRequireDefault(_inherits2);

var _AsyncSubject = require('rxjs/AsyncSubject');

var _BehaviorSubject = require('rxjs/BehaviorSubject');

var _Subject2 = require('rxjs/Subject');

var _Observable = require('rxjs/Observable');

var _merge = require('rxjs/observable/merge');

var _filter = require('rxjs/operator/filter');

var _share = require('rxjs/operator/share');

var _shim = require('./shim.js');

var _serialization = require('./serialization.js');

var _logging = require('./logging.js');

function _interopRequireDefault(obj) { return obj && obj.__esModule ? obj : { default: obj }; }

var PROTOCOL_VERSION = 'rethinkdb-horizon-v0';

// Before connecting the first time
var STATUS_UNCONNECTED = { type: 'unconnected' };
// After the websocket is opened, but before handshake
var STATUS_CONNECTED = { type: 'connected' };
// After the websocket is opened and handshake is completed
var STATUS_READY = { type: 'ready' };
// After unconnected, maybe before or after connected. Any socket level error
var STATUS_ERROR = { type: 'error' };
// Occurs when the socket closes
var STATUS_DISCONNECTED = { type: 'disconnected' };

var ProtocolError = function (_Error) {
  (0, _inherits3.default)(ProtocolError, _Error);

  function ProtocolError(msg, errorCode) {
    (0, _classCallCheck3.default)(this, ProtocolError);

    var _this = (0, _possibleConstructorReturn3.default)(this, _Error.call(this, msg));

    _this.errorCode = errorCode;
    return _this;
  }

  ProtocolError.prototype.toString = function toString() {
    return this.message + ' (Code: ' + this.errorCode + ')';
  };

  return ProtocolError;
}(Error);

// Wraps native websockets with a Subject, which is both an Subscriber
// and an Observable (it is bi-directional after all!). This
// implementation is adapted from Rx.DOM.fromWebSocket and
// RxSocketSubject by Ben Lesh, but it also deals with some simple
// protocol level things like serializing from/to JSON, routing
// request_ids, looking at the `state` field to decide when an
// observable is closed.


var HorizonSocket = function (_Subject) {
  (0, _inherits3.default)(HorizonSocket, _Subject);

  function HorizonSocket(host, secure, path, handshaker) {
    var _context;

    (0, _classCallCheck3.default)(this, HorizonSocket);

    var hostString = 'ws' + (secure ? 's' : '') + '://' + host + '/' + path;
    var msgBuffer = [];
    var ws = void 0,
        handshakeDisp = void 0;
    // Handshake is an asyncsubject because we want it to always cache
    // the last value it received, like a promise
    var handshake = new _AsyncSubject.AsyncSubject();
    var statusSubject = new _BehaviorSubject.BehaviorSubject(STATUS_UNCONNECTED);

    var isOpen = function isOpen() {
      return Boolean(ws) && ws.readyState === _shim.WebSocket.OPEN;
    };

    // Serializes to a string before sending
    function wsSend(msg) {
      var stringMsg = JSON.stringify((0, _serialization.serialize)(msg));
      ws.send(stringMsg);
    }

    // This is the observable part of the Subject. It forwards events
    // from the underlying websocket
    var socketObservable = (_context = _Observable.Observable.create(function (subscriber) {
      ws = new _shim.WebSocket(hostString, PROTOCOL_VERSION);
      ws.onerror = function () {
        // If the websocket experiences the error, we forward it through
        // to the observable. Unfortunately, the event we receive in
        // this callback doesn't tell us much of anything, so there's no
        // reason to forward it on and we just send a generic error.
        statusSubject.next(STATUS_ERROR);
        var errMsg = 'Websocket ' + hostString + ' experienced an error';
        subscriber.error(new Error(errMsg));
      };
      ws.onopen = function () {
        // Send the handshake
        statusSubject.next(STATUS_CONNECTED);
        handshakeDisp = _this2.makeRequest(handshaker()).subscribe(function (x) {
          handshake.next(x);
          handshake.complete();

          handshake.next(STATUS_READY);
        }, function (err) {
          return handshake.error(err);
        }, function () {
          return handshake.complete();
        });
        // Send any messages that have been buffered
        while (msgBuffer.length > 0) {
          var msg = msgBuffer.shift();
          (0, _logging.log)('Sending buffered:', msg);
          wsSend(msg);
        }
      };
      ws.onmessage = function (event) {
        var deserialized = (0, _serialization.deserialize)(JSON.parse(event.data));
        (0, _logging.log)('Received', deserialized);
        subscriber.next(deserialized);
      };
      ws.onclose = function (e) {
        // This will happen if the socket is closed by the server If
        // .close is called from the client (see closeSocket), this
        // listener will be removed
        statusSubject.next(STATUS_DISCONNECTED);
        if (e.code !== 1000 || !e.wasClean) {
          subscriber.error(new Error('Socket closed unexpectedly with code: ' + e.code));
        } else {
          subscriber.complete();
        }
      };
      return function () {
        if (handshakeDisp) {
          handshakeDisp.unsubscribe();
        }
        // This is the "unsubscribe" method on the final Subject
        closeSocket(1000, '');
      };
    }), _share.share).call(_context); // This makes it a "hot" observable, and refCounts it
    // Note possible edge cases: the `share` operator is equivalent to
    // .multicast(() => new Subject()).refCount() // RxJS 5
    // .multicast(new Subject()).refCount() // RxJS 4

    // This is the Subscriber part of the Subject. How we can send stuff
    // over the websocket
    var socketSubscriber = {
      next: function next(messageToSend) {
        // When next is called on this subscriber
        // Note: If we aren't ready, the message is silently dropped
        if (isOpen()) {
          (0, _logging.log)('Sending', messageToSend);
          wsSend(messageToSend); // wsSend serializes to a string
        } else {
            (0, _logging.log)('Buffering', messageToSend);
            msgBuffer.push(messageToSend);
          }
      },
      error: function error(_error) {
        // The subscriber is receiving an error. Better close the
        // websocket with an error
        if (!_error.code) {
          throw new Error('no code specified. Be sure to pass ' + '{ code: ###, reason: "" } to error()');
        }
        closeSocket(_error.code, _error.reason);
      },
      complete: function complete() {
        // complete for the subscriber here is equivalent to "close
        // this socket successfully (which is what code 1000 is)"
        closeSocket(1000, '');
      }
    };

    function closeSocket(code, reason) {
      statusSubject.next(STATUS_DISCONNECTED);
      if (!code) {
        ws.close(); // successful close
      } else {
          ws.close(code, reason);
        }
      ws.onopen = undefined;
      ws.onclose = undefined;
      ws.onmessage = undefined;
    }

    // Subscriptions will be the observable containing all
    // queries/writes/changefeed requests. Specifically, the documents
    // that initiate them, each one with a different request_id

    var _this2 = (0, _possibleConstructorReturn3.default)(this, _Subject.call(this, socketSubscriber, socketObservable));

    var subscriptions = new _Subject2.Subject();
    // Unsubscriptions is similar, only it holds only requests to
    // close a particular request_id on the server. Currently we only
    // need these for changefeeds.
    var unsubscriptions = new _Subject2.Subject();
    var outgoing = _merge.merge.call(_Observable.Observable, subscriptions, unsubscriptions);
    // How many requests are outstanding
    var activeRequests = 0;
    // Monotonically increasing counter for request_ids
    var requestCounter = 0;
    // Unsubscriber for subscriptions/unsubscriptions
    var subDisp = null;
    // Now that super has been called, we can add attributes to this
    _this2.handshake = handshake;
    // Lets external users keep track of the current websocket status
    // without causing it to connect
    _this2.status = statusSubject;

    var incrementActive = function incrementActive() {
      if (++activeRequests === 1) {
        // We subscribe the socket itself to the subscription and
        // unsubscription requests. Since the socket is both an
        // observable and an subscriber. Here it's acting as an subscriber,
        // watching our requests.
        subDisp = outgoing.subscribe(_this2);
      }
    };

    // Decrement the number of active requests on the socket, and
    // close the socket if we're the last request
    var decrementActive = function decrementActive() {
      if (--activeRequests === 0) {
        subDisp.unsubscribe();
      }
    };

    // This is used externally to send requests to the server
    _this2.makeRequest = function (rawRequest) {
      return _Observable.Observable.create(function (reqSubscriber) {
        // Get a new request id
        var request_id = requestCounter++;
        // Add the request id to the request and the unsubscribe request
        // if there is one
        rawRequest.request_id = request_id;
        var unsubscribeRequest = void 0;
        if (rawRequest.type === 'subscribe') {
          unsubscribeRequest = { request_id: request_id, type: 'end_subscription' };
        }
        // First, increment activeRequests and decide if we need to
        // connect to the socket
        incrementActive();

        // Now send the request to the server
        subscriptions.next(rawRequest);

        // Create an observable from the socket that filters by request_id
        var unsubscribeFilter = _filter.filter.call(_this2, function (x) {
          return x.request_id === request_id;
        }).subscribe(function (resp) {
          // Need to faithfully end the stream if there is an error
          if (resp.error !== undefined) {
            reqSubscriber.error(new ProtocolError(resp.error, resp.error_code));
          } else if (resp.data !== undefined || resp.token !== undefined) {
            try {
              reqSubscriber.next(resp);
            } catch (e) {}
          }
          if (resp.state === 'synced') {
            // Create a little dummy object for sync notifications
            reqSubscriber.next({
              type: 'state',
              state: 'synced'
            });
          } else if (resp.state === 'complete') {
            reqSubscriber.complete();
          }
        }, function (err) {
          return reqSubscriber.error(err);
        }, function () {
          return reqSubscriber.complete();
        });
        return function () {
          // Unsubscribe if necessary
          if (unsubscribeRequest) {
            unsubscriptions.next(unsubscribeRequest);
          }
          decrementActive();
          unsubscribeFilter.unsubscribe();
        };
      });
    };
    return _this2;
  }

  return HorizonSocket;
}(_Subject2.Subject);

module.exports = HorizonSocket;
//# sourceMappingURL=socket.js.map