openvidu/openvidu-browser/src/OpenViduInternal/KurentoUtils/kurento-jsonrpc/index.js

/*
 * (C) Copyright 2014 Kurento (http://kurento.org/)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

var defineProperty_IE8 = false;
if (Object.defineProperty) {
    try {
        Object.defineProperty({}, 'x', {});
    } catch (e) {
        defineProperty_IE8 = true;
    }
}

// https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/Function/bind
if (!Function.prototype.bind) {
    Function.prototype.bind = function (oThis) {
        if (typeof this !== 'function') {
            // closest thing possible to the ECMAScript 5
            // internal IsCallable function
            throw new TypeError('Function.prototype.bind - what is trying to be bound is not callable');
        }

        var aArgs = Array.prototype.slice.call(arguments, 1),
            fToBind = this,
            fNOP = function () {},
            fBound = function () {
                return fToBind.apply(this instanceof fNOP && oThis ? this : oThis, aArgs.concat(Array.prototype.slice.call(arguments)));
            };

        fNOP.prototype = this.prototype;
        fBound.prototype = new fNOP();

        return fBound;
    };
}

var EventEmitter = require('events').EventEmitter;

var inherits = require('inherits');

var packers = require('./packers');
var Mapper = require('./Mapper');

var BASE_TIMEOUT = 5000;

function unifyResponseMethods(responseMethods) {
    if (!responseMethods) return {};

    for (var key in responseMethods) {
        var value = responseMethods[key];

        if (typeof value == 'string')
            responseMethods[key] = {
                response: value
            };
    }

    return responseMethods;
}

function unifyTransport(transport) {
    if (!transport) return;

    // Transport as a function
    if (transport instanceof Function)
        return {
            send: transport
        };

    // WebSocket & DataChannel
    if (transport.send instanceof Function) return transport;

    // Message API (Inter-window & WebWorker)
    if (transport.postMessage instanceof Function) {
        transport.send = transport.postMessage;
        return transport;
    }

    // Stream API
    if (transport.write instanceof Function) {
        transport.send = transport.write;
        return transport;
    }

    // Transports that only can receive messages, but not send
    if (transport.onmessage !== undefined) return;
    if (transport.pause instanceof Function) return;

    throw new SyntaxError('Transport is not a function nor a valid object');
}

/**
 * Representation of a RPC notification
 *
 * @class
 *
 * @constructor
 *
 * @param {String} method -method of the notification
 * @param params - parameters of the notification
 */
function RpcNotification(method, params) {
    if (defineProperty_IE8) {
        this.method = method;
        this.params = params;
    } else {
        Object.defineProperty(this, 'method', {
            value: method,
            enumerable: true
        });
        Object.defineProperty(this, 'params', {
            value: params,
            enumerable: true
        });
    }
}

/**
 * @class
 *
 * @constructor
 *
 * @param {object} packer
 *
 * @param {object} [options]
 *
 * @param {object} [transport]
 *
 * @param {Function} [onRequest]
 */
function RpcBuilder(packer, options, transport, onRequest) {
    var self = this;

    if (!packer) throw new SyntaxError('Packer is not defined');

    if (!packer.pack || !packer.unpack) throw new SyntaxError('Packer is invalid');

    var responseMethods = unifyResponseMethods(packer.responseMethods);

    if (options instanceof Function) {
        if (transport != undefined) throw new SyntaxError("There can't be parameters after onRequest");

        onRequest = options;
        transport = undefined;
        options = undefined;
    }

    if (options && options.send instanceof Function) {
        if (transport && !(transport instanceof Function)) throw new SyntaxError('Only a function can be after transport');

        onRequest = transport;
        transport = options;
        options = undefined;
    }

    if (transport instanceof Function) {
        if (onRequest != undefined) throw new SyntaxError("There can't be parameters after onRequest");

        onRequest = transport;
        transport = undefined;
    }

    if (transport && transport.send instanceof Function)
        if (onRequest && !(onRequest instanceof Function)) throw new SyntaxError('Only a function can be after transport');

    options = options || {};

    EventEmitter.call(this);

    if (onRequest) this.on('request', onRequest);

    if (defineProperty_IE8) this.peerID = options.peerID;
    else
        Object.defineProperty(this, 'peerID', {
            value: options.peerID
        });

    var max_retries = options.max_retries || 0;

    function transportMessage(event) {
        self.decode(event.data || event);
    }

    this.getTransport = function () {
        return transport;
    };
    this.setTransport = function (value) {
        // Remove listener from old transport
        if (transport) {
            // W3C transports
            if (transport.removeEventListener) transport.removeEventListener('message', transportMessage);
            // Node.js Streams API
            else if (transport.removeListener) transport.removeListener('data', transportMessage);
        }

        // Set listener on new transport
        if (value) {
            // W3C transports
            if (value.addEventListener) value.addEventListener('message', transportMessage);
            // Node.js Streams API
            else if (value.addListener) value.addListener('data', transportMessage);
        }

        transport = unifyTransport(value);
    };

    if (!defineProperty_IE8)
        Object.defineProperty(this, 'transport', {
            get: this.getTransport.bind(this),
            set: this.setTransport.bind(this)
        });

    this.setTransport(transport);

    var request_timeout = options.request_timeout || BASE_TIMEOUT;
    var ping_request_timeout = options.ping_request_timeout || request_timeout;
    var response_timeout = options.response_timeout || BASE_TIMEOUT;
    var duplicates_timeout = options.duplicates_timeout || BASE_TIMEOUT;

    var requestID = 0;

    var requests = new Mapper();
    var responses = new Mapper();
    var processedResponses = new Mapper();

    var message2Key = {};

    /**
     * Store the response to prevent to process duplicate request later
     */
    function storeResponse(message, id, dest) {
        var response = {
            message: message,
            /** Timeout to auto-clean old responses */
            timeout: setTimeout(function () {
                responses.remove(id, dest);
            }, response_timeout)
        };

        responses.set(response, id, dest);
    }

    /**
     * Store the response to ignore duplicated messages later
     */
    function storeProcessedResponse(ack, from) {
        var timeout = setTimeout(function () {
            processedResponses.remove(ack, from);
        }, duplicates_timeout);

        processedResponses.set(timeout, ack, from);
    }

    /**
     * Representation of a RPC request
     *
     * @class
     * @extends RpcNotification
     *
     * @constructor
     *
     * @param {String} method -method of the notification
     * @param params - parameters of the notification
     * @param {Integer} id - identifier of the request
     * @param [from] - source of the notification
     */
    function RpcRequest(method, params, id, from, transport) {
        RpcNotification.call(this, method, params);

        this.getTransport = function () {
            return transport;
        };
        this.setTransport = function (value) {
            transport = unifyTransport(value);
        };

        if (!defineProperty_IE8)
            Object.defineProperty(this, 'transport', {
                get: this.getTransport.bind(this),
                set: this.setTransport.bind(this)
            });

        var response = responses.get(id, from);

        /**
         * @constant {Boolean} duplicated
         */
        if (!(transport || self.getTransport())) {
            if (defineProperty_IE8) this.duplicated = Boolean(response);
            else
                Object.defineProperty(this, 'duplicated', {
                    value: Boolean(response)
                });
        }

        var responseMethod = responseMethods[method];

        this.pack = packer.pack.bind(packer, this, id);

        /**
         * Generate a response to this request
         *
         * @param {Error} [error]
         * @param {*} [result]
         *
         * @returns {string}
         */
        this.reply = function (error, result, transport) {
            // Fix optional parameters
            if (error instanceof Function || (error && error.send instanceof Function)) {
                if (result != undefined) throw new SyntaxError("There can't be parameters after callback");

                transport = error;
                result = null;
                error = undefined;
            } else if (result instanceof Function || (result && result.send instanceof Function)) {
                if (transport != undefined) throw new SyntaxError("There can't be parameters after callback");

                transport = result;
                result = null;
            }

            transport = unifyTransport(transport);

            // Duplicated request, remove old response timeout
            if (response) clearTimeout(response.timeout);

            if (from != undefined) {
                if (error) error.dest = from;

                if (result) result.dest = from;
            }

            var message;

            // New request or overriden one, create new response with provided data
            if (error || result != undefined) {
                if (self.peerID != undefined) {
                    if (error) error.from = self.peerID;
                    else result.from = self.peerID;
                }

                // Protocol indicates that responses has own request methods
                if (responseMethod) {
                    if (responseMethod.error == undefined && error)
                        message = {
                            error: error
                        };
                    else {
                        var method = error ? responseMethod.error : responseMethod.response;

                        message = {
                            method: method,
                            params: error || result
                        };
                    }
                } else
                    message = {
                        error: error,
                        result: result
                    };

                message = packer.pack(message, id);
            }

            // Duplicate & not-overriden request, re-send old response
            else if (response) message = response.message;
            // New empty reply, response null value
            else
                message = packer.pack(
                    {
                        result: null
                    },
                    id
                );

            // Store the response to prevent to process a duplicated request later
            storeResponse(message, id, from);

            // Return the stored response so it can be directly send back
            transport = transport || this.getTransport() || self.getTransport();

            if (transport) return transport.send(message);

            return message;
        };
    }
    inherits(RpcRequest, RpcNotification);

    function cancel(message) {
        var key = message2Key[message];
        if (!key) return;

        delete message2Key[message];

        var request = requests.pop(key.id, key.dest);
        if (!request) return;

        clearTimeout(request.timeout);

        // Start duplicated responses timeout
        storeProcessedResponse(key.id, key.dest);
    }

    /**
     * Allow to cancel a request and don't wait for a response
     *
     * If `message` is not given, cancel all the request
     */
    this.cancel = function (message) {
        if (message) return cancel(message);

        for (var message in message2Key) cancel(message);
    };

    this.close = function () {
        // Prevent to receive new messages
        var transport = this.getTransport();
        if (transport && transport.close) transport.close(4003, 'Cancel request');

        // Request & processed responses
        this.cancel();

        processedResponses.forEach(clearTimeout);

        // Responses
        responses.forEach(function (response) {
            clearTimeout(response.timeout);
        });
    };

    /**
     * Generates and encode a JsonRPC 2.0 message
     *
     * @param {String} method -method of the notification
     * @param params - parameters of the notification
     * @param [dest] - destination of the notification
     * @param {object} [transport] - transport where to send the message
     * @param [callback] - function called when a response to this request is
     *   received. If not defined, a notification will be send instead
     *
     * @returns {string} A raw JsonRPC 2.0 request or notification string
     */
    this.encode = function (method, params, dest, transport, callback) {
        // Fix optional parameters
        if (params instanceof Function) {
            if (dest != undefined) throw new SyntaxError("There can't be parameters after callback");

            callback = params;
            transport = undefined;
            dest = undefined;
            params = undefined;
        } else if (dest instanceof Function) {
            if (transport != undefined) throw new SyntaxError("There can't be parameters after callback");

            callback = dest;
            transport = undefined;
            dest = undefined;
        } else if (transport instanceof Function) {
            if (callback != undefined) throw new SyntaxError("There can't be parameters after callback");

            callback = transport;
            transport = undefined;
        }

        if (self.peerID != undefined) {
            params = params || {};

            params.from = self.peerID;
        }

        if (dest != undefined) {
            params = params || {};

            params.dest = dest;
        }

        // Encode message
        var message = {
            method: method,
            params: params
        };

        if (callback) {
            var id = requestID++;
            var retried = 0;

            message = packer.pack(message, id);

            function dispatchCallback(error, result) {
                self.cancel(message);

                callback(error, result);
            }

            var request = {
                message: message,
                callback: dispatchCallback,
                responseMethods: responseMethods[method] || {}
            };

            var encode_transport = unifyTransport(transport);

            function sendRequest(transport) {
                var rt = method === 'ping' ? ping_request_timeout : request_timeout;
                request.timeout = setTimeout(timeout, rt * Math.pow(2, retried++));
                message2Key[message] = {
                    id: id,
                    dest: dest
                };
                requests.set(request, id, dest);

                transport = transport || encode_transport || self.getTransport();
                if (transport) return transport.send(message);

                return message;
            }

            function retry(transport) {
                transport = unifyTransport(transport);

                console.warn(retried + ' retry for request message:', message);

                var timeout = processedResponses.pop(id, dest);
                clearTimeout(timeout);

                return sendRequest(transport);
            }

            function timeout() {
                if (retried < max_retries) return retry(transport);

                var error = new Error('Request has timed out');
                error.request = message;

                error.retry = retry;

                dispatchCallback(error);
            }

            return sendRequest(transport);
        }

        // Return the packed message
        message = packer.pack(message);

        transport = transport || this.getTransport();
        if (transport) return transport.send(message);

        return message;
    };

    /**
     * Decode and process a JsonRPC 2.0 message
     *
     * @param {string} message - string with the content of the message
     *
     * @returns {RpcNotification|RpcRequest|undefined} - the representation of the
     *   notification or the request. If a response was processed, it will return
     *   `undefined` to notify that it was processed
     *
     * @throws {TypeError} - Message is not defined
     */
    this.decode = function (message, transport) {
        if (!message) throw new TypeError('Message is not defined');

        try {
            message = packer.unpack(message);
        } catch (e) {
            // Ignore invalid messages
            return console.debug(e, message);
        }

        var id = message.id;
        var ack = message.ack;
        var method = message.method;
        var params = message.params || {};

        var from = params.from;
        var dest = params.dest;

        // Ignore messages send by us
        if (self.peerID != undefined && from == self.peerID) return;

        // Notification
        if (id == undefined && ack == undefined) {
            var notification = new RpcNotification(method, params);

            if (self.emit('request', notification)) return;
            return notification;
        }

        function processRequest() {
            // If we have a transport and it's a duplicated request, reply inmediatly
            transport = unifyTransport(transport) || self.getTransport();
            if (transport) {
                var response = responses.get(id, from);
                if (response) return transport.send(response.message);
            }

            var idAck = id != undefined ? id : ack;
            var request = new RpcRequest(method, params, idAck, from, transport);

            if (self.emit('request', request)) return;
            return request;
        }

        function processResponse(request, error, result) {
            request.callback(error, result);
        }

        function duplicatedResponse(timeout) {
            console.warn('Response already processed', message);

            // Update duplicated responses timeout
            clearTimeout(timeout);
            storeProcessedResponse(ack, from);
        }

        // Request, or response with own method
        if (method) {
            // Check if it's a response with own method
            if (dest == undefined || dest == self.peerID) {
                var request = requests.get(ack, from);
                if (request) {
                    var responseMethods = request.responseMethods;

                    if (method == responseMethods.error) return processResponse(request, params);

                    if (method == responseMethods.response) return processResponse(request, null, params);

                    return processRequest();
                }

                var processed = processedResponses.get(ack, from);
                if (processed) return duplicatedResponse(processed);
            }

            // Request
            return processRequest();
        }

        var error = message.error;
        var result = message.result;

        // Ignore responses not send to us
        if (error && error.dest && error.dest != self.peerID) return;
        if (result && result.dest && result.dest != self.peerID) return;

        // Response
        var request = requests.get(ack, from);
        if (!request) {
            var processed = processedResponses.get(ack, from);
            if (processed) return duplicatedResponse(processed);

            return console.warn('No callback was defined for this message', message);
        }

        // Process response
        processResponse(request, error, result);
    };
}
inherits(RpcBuilder, EventEmitter);

RpcBuilder.RpcNotification = RpcNotification;

module.exports = RpcBuilder;

var clients = require('./clients');
var transports = require('./clients/transports');

RpcBuilder.clients = clients;
RpcBuilder.clients.transports = transports;
RpcBuilder.packers = packers;