/* * (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 RpcBuilder = require('../'); var WebSocketWithReconnection = require('./transports/webSocketWithReconnection'); var OpenViduLogger = require('../../../Logger/OpenViduLogger').OpenViduLogger; Date.now = Date.now || function () { return +new Date; }; var PING_INTERVAL = 5000; var RECONNECTING = 'RECONNECTING'; var CONNECTED = 'CONNECTED'; var DISCONNECTED = 'DISCONNECTED'; var Logger = OpenViduLogger.getInstance(); /** * * heartbeat: interval in ms for each heartbeat message, *
* ws : { * uri : URI to conntect to, * onconnected : callback method to invoke when connection is successful, * ondisconnect : callback method to invoke when the connection is lost (max retries for reconnecting reached), * onreconnecting : callback method to invoke when the client is reconnecting, * onreconnected : callback method to invoke when the client successfully reconnects, * onerror : callback method to invoke when there is an error * }, * rpc : { * requestTimeout : timeout for a request, * sessionStatusChanged: callback method for changes in session status, * mediaRenegotiation: mediaRenegotiation * } **/ function JsonRpcClient(configuration) { var self = this; var wsConfig = configuration.ws; var notReconnectIfNumLessThan = -1; var pingNextNum = 0; var enabledPings = true; var pingPongStarted = false; var pingInterval; var status = DISCONNECTED; var onreconnecting = wsConfig.onreconnecting; var onreconnected = wsConfig.onreconnected; var onconnected = wsConfig.onconnected; var onerror = wsConfig.onerror; configuration.rpc.pull = function (params, request) { request.reply(null, "push"); } wsConfig.onreconnecting = function () { Logger.debug("--------- ONRECONNECTING -----------"); if (status === RECONNECTING) { Logger.error("Websocket already in RECONNECTING state when receiving a new ONRECONNECTING message. Ignoring it"); return; } stopPing(); status = RECONNECTING; if (onreconnecting) { onreconnecting(); } } wsConfig.onreconnected = function () { Logger.debug("--------- ONRECONNECTED -----------"); if (status === CONNECTED) { Logger.error("Websocket already in CONNECTED state when receiving a new ONRECONNECTED message. Ignoring it"); return; } status = CONNECTED; updateNotReconnectIfLessThan(); if (onreconnected) { onreconnected(); } } wsConfig.onconnected = function () { Logger.debug("--------- ONCONNECTED -----------"); if (status === CONNECTED) { Logger.error("Websocket already in CONNECTED state when receiving a new ONCONNECTED message. Ignoring it"); return; } status = CONNECTED; enabledPings = true; usePing(); if (onconnected) { onconnected(); } } wsConfig.onerror = function (error) { Logger.debug("--------- ONERROR -----------"); status = DISCONNECTED; stopPing(); if (onerror) { onerror(error); } } var ws = new WebSocketWithReconnection(wsConfig); Logger.debug('Connecting websocket to URI: ' + wsConfig.uri); var rpcBuilderOptions = { request_timeout: configuration.rpc.requestTimeout, ping_request_timeout: configuration.rpc.heartbeatRequestTimeout }; var rpc = new RpcBuilder(RpcBuilder.packers.JsonRPC, rpcBuilderOptions, ws, function (request) { Logger.debug('Received request: ' + JSON.stringify(request)); try { var func = configuration.rpc[request.method]; if (func === undefined) { Logger.error("Method " + request.method + " not registered in client"); } else { func(request.params, request); } } catch (err) { Logger.error('Exception processing request: ' + JSON.stringify(request)); Logger.error(err); } }); this.send = function (method, params, callback) { var requestTime = Date.now(); rpc.encode(method, params, function (error, result) { if (error) { try { Logger.error("ERROR:" + error.message + " in Request: method:" + method + " params:" + JSON.stringify(params) + " request:" + error.request); if (error.data) { Logger.error("ERROR DATA:" + JSON.stringify(error.data)); } } catch (e) {} error.requestTime = requestTime; } if (callback) { if (result != undefined && result.value !== 'pong') { Logger.debug('Response: ' + JSON.stringify(result)); } callback(error, result); } }); } function updateNotReconnectIfLessThan() { Logger.debug("notReconnectIfNumLessThan = " + pingNextNum + ' (old=' + notReconnectIfNumLessThan + ')'); notReconnectIfNumLessThan = pingNextNum; } function sendPing() { if (enabledPings) { var params = null; if (pingNextNum == 0 || pingNextNum == notReconnectIfNumLessThan) { params = { interval: configuration.heartbeat || PING_INTERVAL }; } pingNextNum++; self.send('ping', params, (function (pingNum) { return function (error, result) { if (error) { Logger.debug("Error in ping request #" + pingNum + " (" + error.message + ")"); if (pingNum > notReconnectIfNumLessThan) { enabledPings = false; updateNotReconnectIfLessThan(); Logger.debug("Server did not respond to ping message #" + pingNum + ". Reconnecting... "); ws.reconnectWs(); } } } })(pingNextNum)); } else { Logger.debug("Trying to send ping, but ping is not enabled"); } } /* * If configuration.hearbeat has any value, the ping-pong will work with the interval * of configuration.hearbeat */ function usePing() { if (!pingPongStarted) { Logger.debug("Starting ping (if configured)") pingPongStarted = true; if (configuration.heartbeat != undefined) { pingInterval = setInterval(sendPing, configuration.heartbeat); sendPing(); } } } function stopPing() { clearInterval(pingInterval); pingPongStarted = false; enabledPings = false; pingNextNum = -1; rpc.cancel(); } this.close = function (code, reason) { Logger.debug("Closing with code: " + code + " because: " + reason); if (pingInterval != undefined) { Logger.debug("Clearing ping interval"); clearInterval(pingInterval); } pingPongStarted = false; enabledPings = false; ws.close(code, reason); } this.reconnect = function () { ws.reconnectWs(); } this.resetPing = function () { enabledPings = true; pingNextNum = 0; usePing(); } this.getReadyState = function () { return ws.getReadyState(); } } module.exports = JsonRpcClient;