netjsagent/lib/wsDataConnection-handler.js

var WebSocket = require('ws');
var agent = require("./agent-setting");
var DataMessageHandler = require("./datamessage-handler");
var util = require('./util');

var backlog = [];
var thisInstance= undefined;

function dataConnHandler(){
    thisInstance= this;
    this.client=null;
    this.timeout=null;
    this.dataMsgHandler=null;
    this.discardedFPLength = 0;
    this.istempDataConn = false;
    this.client;
    this.protocol;
    /*
     * Adding NDCHost & port in constructor, so if control connection is getting close during running test
     * then agent will switch over and will try to connect with backup NDC, but Data and Auto connection will connect with
     * previous NDC, untill control connection with new server has been made
     * */
    this.ndcHost=0 ;this.ndcPort=7892;
}

var closeConnListener =function(err) {
    var self =thisInstance
    util.logger.warn(agent.currentTestRun+" | Data connection, Received socket close event from Host : "+self.ndcHost+" ,Port="+self.ndcPort +", Error :  "+err);
    if(self.istempDataConn) {
        self.closeConnection();
    }
    else{
        if(this.client) {
            self.client.removeAllListeners();
        }
        self.connectToServer();
    }
}

var connectConnListener = function() {
    try {
        var self =thisInstance
        self.timeout = undefined;
        if(!self.istempDataConn){
            util.logger.info(agent.currentTestRun+" | Data Connection established with NDCollector : Socket[addr="+self.ndcHost+",port="+self.ndcPort + ",localport=" +self.getLocalPort()+"]" );
            self.dataMsgHandler = new DataMessageHandler(self);
            if( backlog.length ) {
                for(var i= 0, len= backlog.length; len>i; ++i)
                 self.client.send(backlog[i]);

                backlog.length= 0;
            }
        }
    }
    catch(e){
        util.logger.warn(e);
    }
}


dataConnHandler.prototype.getLocalPort = function(){
    try {
        if(!this.client)return ;
        return this.client._socket.address() ? this.client._socket.address().port : undefined;
    }catch(err){util.logger.warn(err);}
}

dataConnHandler.prototype.createDataConn = function(server,istempDataConn,currProto){
    try {
        this.ndcHost = server.ndcHost;              //Setting new Host port, at time of new Object [connection] creation
        this.ndcPort = currProto.port;
        this.istempDataConn = istempDataConn;       //Flag for New Data Connection.
        this.protocol = currProto.protocol
        var self = this;
        this._connect();
    }catch(err){util.logger.warn(err);}
};


dataConnHandler.prototype.connectToServer=function() {
    try {
        if (!agent.isTestRunning) {
            util.logger.warn(agent.currentTestRun + " | Test run is not running .")
            return;
        }
        var self = this;
        if (self.timeout)
            return;

        self.timeout = setTimeout(function () {
            try {
                self.timeout = undefined;
                self._connect();
                util.logger.warn(agent.currentTestRun + " | Timer for retrying Data connectoion expired. trying to connect with Host : " + self.ndcHost + " ,Port=" + self.ndcPort);
            } catch (e) {
                util.logger.warn(e);
            }
        }, 60000);
    }
    catch(e){
        util.logger.warn(e);
    }
}


dataConnHandler.prototype.closeConnection =function() {
    try {
        /*
         * 1. We are removing close listener befor closing connection because is close listener we are retrying for connection .
         * 2. After closing connection , removing all liteners ,because in close's handshaking, server will send error, so agent should handle
         * exceptions*/
        util.logger.info(agent.currentTestRun + " | Closing the New Data connection .");
        if (this.client) {
            this.client.removeListener('close',closeConnListener)
            this.client.removeListener('open',connectConnListener);
            this.client.close();
            //this.client.removeAllListeners()

            delete this.client;
            this.client = undefined;
        }
        this.ndcHost = 0;
        this.ndcPort = 7892;
        clearTimeout(this.timeout)
        this.timeout = null;
        this.istempDataConn = false;
        this.discardedFPLength = 0;
        delete this.dataMsgHandler;
    }
    catch(e){
        util.logger.warn(e);
    }
}

dataConnHandler.prototype._connect = function() {
    var self = this;
    if(!agent.isTestRunning && !self.istempDataConn) {
        util.logger.warn(agent.currentTestRun+" | Test is not running ,error in making data connection")
        return;
    }
    try {
        var url = this.protocol+'://'+self.ndcHost+':'+self.ndcPort+'/' ;
        var options={}
        if(this.protocol.toLowerCase() == 'wss')
            options.rejectUnauthorized = false
        this.client = new WebSocket(url,options);

        this.client.on('error', function(err) {});
        this.client.on('close',closeConnListener);
        this.client.on('open',connectConnListener);
            }
    catch(err) {
            util.logger.warn(agent.currentTestRun+" | Error in making data connection");
        }
};

dataConnHandler.prototype.write=function(data){
    try {
        if(!this.client ||!data || !data.length)return
        if(this.client.readyState === WebSocket.OPEN) {
            if(!this.istempDataConn) {
                if (this.client.bufferedAmount >= agent.ndDataBufferSize) {
                    if (this.discardedFPLength % 1000 === 0) {
                        util.logger.warn(agent.currentTestRun + " | Discarding Data conn data, Buffer size : ", this.client.bufferSize, " is greater then ndDataBufferSize");
                        this.discardedFPLength = 0
                    }
                    ++this.discardedFPLength;
                    return false
                }
            }
            this.client.send(data);
        }
        else{
            if (backlog.length <= 500)
                backlog.push(data);

            util.logger.warn(agent.currentTestRun+"| client not connected ...........")
            return
        }
    }
    catch(e){
        util.logger.warn(agent.currentTestRun+" | Error in sending data on socket in dataConnHandler ",e);
    }
};

module.exports = dataConnHandler;