xen-orchestra/packages/xo-lib/index.js

'use strict';

//====================================================================

var assign = require('lodash.assign');
var Bluebird = require('bluebird');
var EventEmitter = require('events').EventEmitter;
var forEach = require('lodash.foreach');
var inherits = require('util').inherits;
var jsonRpc = require('json-rpc');
var MethodNotFound = require('json-rpc/errors').MethodNotFound;
var parseUrl = require('url').parse;
var WebSocket = require('ws');

//====================================================================

function makeDeferred() {
  var resolve, reject;
  var promise = new Bluebird(function (resolve_, reject_) {
    resolve = resolve_;
    reject = reject_;
  });

  return {
    promise: promise,
    reject: reject,
    resolve: resolve,
  };
}

function startsWith(string, target) {
  return (string.lastIndexOf(target, 0) === 0);
}

//====================================================================

// Fix URL if necessary.
var URL_RE = /^(?:(?:http|ws)(s)?:\/\/)?(.*?)\/*(?:\/api\/)?$/;
function fixUrl(url) {
  var matches = URL_RE.exec(url);
  var isSecure = !!matches[1];
  var rest = matches[2];

  return [
    isSecure ? 'wss' : 'ws',
    '://',
    rest,
    '/api/',
  ].join('');
}

//====================================================================

function Xo(url) {
  // Super constructor.
  EventEmitter.call(this);

  // Fix the URL (ensure correct protocol and /api/ path).
  this._url = fixUrl(url);

  // Current status which may be:
  // - disconnected
  // - connecting
  // - connected
  this.status = 'disconnected';

  // Will contains the WebSocket.
  this._socket = null;

  // The JSON-RPC server.
  var this_ = this;
  this._jsonRpc = jsonRpc.createServer(function (message) {
    if (message.type === 'notification') {
      this_.emit('notification', message);
    } else {
      // This object does not support requests.
      throw new MethodNotFound(message.method);
    }
  }).on('data', function (message) {
    this_._socket.send(JSON.stringify(message));
  });
}
inherits(Xo, EventEmitter);

assign(Xo.prototype, {
  close: function () {
    if (this._socket) {
      this._socket.close();
    }
  },

  connect: Bluebird.method(function () {
    if (this._socket) {
      return;
    }
    this.status = 'connecting';

    var deferred = makeDeferred();

    var opts = {};
    if (startsWith(this._url, 'wss')) {
      // Due to imperfect TLS implementation in XO-Server.
      opts.rejectUnauthorized = false;
    }
    var socket = this._socket = new WebSocket(this._url, '', opts);

    // Used to avoid binding listeners to this object.
    var this_ = this;

    // When the socket opens, send any queued requests.
    socket.addEventListener('open', function () {
      this.status = 'connected';

      // Resolves the promise.
      deferred.resolve();
    });

    socket.addEventListener('message', function (message) {
      this_._jsonRpc.write(message.data);
    });

    socket.addEventListener('close', function () {
      this_.status = 'disconnected';
      this_._socket = null;

      this_._jsonRpc.failPendingRequests('connection lost');
    });

    socket.addEventListener('error', function (error) {
      // Fails the connect promise if possible.
      deferred.reject(error);
    });

    return deferred.promise;
  }),

  call: function (method, params) {
    var jsonRpc = this._jsonRpc;

    return this.connect().then(function () {
      return jsonRpc.request(method, params);
    });
  },
});

//====================================================================

exports = module.exports = Xo;
exports.fixUrl = fixUrl;