tcp_transport.js

"use strict";
/**
 * @module opcua.transport
 */


// system requires
const EventEmitter = require("events").EventEmitter;
const assert = require("node-opcua-assert").assert;
const _ = require("underscore");
const util = require("util");

// opcua requires
const PacketAssembler = require("node-opcua-packet-assembler").PacketAssembler;

const writeTCPMessageHeader = require("./tools").writeTCPMessageHeader;


const readRawMessageHeader =  require("./message_builder_base").readRawMessageHeader;


const buffer_utils = require("node-opcua-buffer-utils");
const createFastUninitializedBuffer = buffer_utils.createFastUninitializedBuffer;


const debug = require("node-opcua-debug");
const debugLog = debug.make_debugLog(__filename);
const doDebug = debug.checkDebugFlag(__filename);

let fakeSocket = {invalid: true};

exports.setFakeTransport = function (socket_like_mock) {
    fakeSocket = socket_like_mock;
};

exports.getFakeTransport = function () {
    if (fakeSocket.invalid){
        throw new Error("getFakeTransport: setFakeTransport must be called first  - BadProtocolVersionUnsupported");
    }
    return fakeSocket;
};

/**
 * TCP_transport
 *
 * @class TCP_transport
 * @constructor
 * @extends EventEmitter
 */
function TCP_transport() {

    /**
     * timeout
     * @property [timeout=30000]
     * @type {number}
     */
    this.timeout = 30000; // 30 seconds timeout

    this._socket = null;

    /**
     * @property headerSize the size of the header in bytes
     * @type {number}
     * @default  8
     */
    this.headerSize = 8;

    /**
     * @property protocolVersion indicates the version number of the OPCUA protocol used
     * @type {number}
     * @default  0
     */
    this.protocolVersion = 0;

    this.__disconnecting__ = false;

    this.bytesWritten = 0;
    this.bytesRead = 0;

    this._the_callback = null;


    /***
     * @property chunkWrittenCount
     * @type {number}
     */
    this.chunkWrittenCount= 0;
    /***
     * @property chunkReadCount
     * @type {number}
     */
    this.chunkReadCount= 0;
}


util.inherits(TCP_transport, EventEmitter);


/**
 * ```createChunk``` is used to construct a pre-allocated chunk to store up to ```length``` bytes of data.
 * The created chunk includes a prepended header for ```chunk_type``` of size ```self.headerSize```.
 *
 * @method createChunk
 * @param msg_type
 * @param chunk_type {String} chunk type. should be 'F' 'C' or 'A'
 * @param length
 * @return {Buffer} a buffer object with the required length representing the chunk.
 *
 * Note:
 *  - only one chunk can be created at a time.
 *  - a created chunk should be committed using the ```write``` method before an other one is created.
 */
TCP_transport.prototype.createChunk = function (msg_type, chunk_type, length) {

    assert(msg_type === "MSG");
    assert(this._pending_buffer === undefined, "createChunk has already been called ( use write first)");

    const total_length = length + this.headerSize;
    const buffer = createFastUninitializedBuffer(total_length);
    writeTCPMessageHeader("MSG", chunk_type, total_length, buffer);

    this._pending_buffer = buffer;

    return buffer;
};

let counter =0;
function record(data,extra) {

    extra = extra || "";

    const c = ("00000000" +counter.toString(10)).substr(-8);
    data._serialNumber = c;
    data.info = data.info || "";
    const l ="";// "-L="+("00000000" +data.length).substr(-8);
    require("fs").writeFileSync("data"+c+l+data.info+extra+".org",data,"binary");
    counter++;
}


TCP_transport.prototype._write_chunk = function (message_chunk) {

    if (this._socket) {
        this.bytesWritten += message_chunk.length;
        this.chunkWrittenCount ++;
        this._socket.write(message_chunk);
    }
};

/**
 * write the message_chunk on the socket.
 * @method write
 * @param message_chunk {Buffer}
 *
 * Notes:
 *  - the message chunk must have been created by ```createChunk```.
 *  - once a message chunk has been written, it is possible to call ```createChunk``` again.
 *
 */
TCP_transport.prototype.write = function (message_chunk) {

    assert((this._pending_buffer === undefined) || this._pending_buffer === message_chunk, " write should be used with buffer created by createChunk");

    const header = readRawMessageHeader(message_chunk);
    assert(header.length === message_chunk.length);
    assert(["F", "C", "A"].indexOf(header.messageHeader.isFinal) !== -1);

    this._write_chunk(message_chunk);

    this._pending_buffer = undefined;
};


function _fulfill_pending_promises(err, data) {

    const self = this;

    _cleanup_timers.call(self);

    const the_callback = self._the_callback;
    self._the_callback = null;

    if (the_callback) {
        the_callback(err, data);
        return true;
    }
    return false;

}

function _on_message_received(message_chunk) {

    const self = this;
    const has_callback = _fulfill_pending_promises.call(self, null, message_chunk);
    self.chunkReadCount ++;

    if (!has_callback) {
        /**
         * notify the observers that a message chunk has been received
         * @event message
         * @param message_chunk {Buffer} the message chunk
         */
        self.emit("message", message_chunk);
    }
}


function _cleanup_timers() {

    const self = this;
    if (self._timerId) {
        clearTimeout(self._timerId);
        this._timerId = null;
    }
}

function _start_timeout_timer() {

    const self = this;
    assert(!self._timerId, "timer already started");
    self._timerId = setTimeout(function () {
        self._timerId =null;
        _fulfill_pending_promises.call(self, new Error("Timeout in waiting for data on socket ( timeout was = " + self.timeout + " ms )"));
    }, self.timeout);

}

TCP_transport.prototype.on_socket_closed = function(err) {

    const self = this;
    if (self._on_socket_closed_called) {
        return;
    }
    assert(!self._on_socket_closed_called);
    self._on_socket_closed_called = true; // we don't want to send close event twice ...
    /**
     * notify the observers that the transport layer has been disconnected.
     * @event socket_closed
     * @param err the Error object or null
     */
    self.emit("socket_closed", err || null);
};

TCP_transport.prototype.on_socket_ended = function(err) {
    const self = this;
    assert(!self._on_socket_ended_called);
    self._on_socket_ended_called = true; // we don't want to send close event twice ...
    /**
     * notify the observers that the transport layer has been disconnected.
     * @event close
     * @param err the Error object or null
     */
    self.emit("close", err || null);
};

TCP_transport.prototype._on_socket_ended_message =  function(err) {

    const self = this;
    if (self.__disconnecting__) {
        return;
    }
    self._on_socket_ended = null;
    self._on_data_received = null;

    debugLog("Transport Connection ended".red + " " + self.name);
    assert(!self.__disconnecting__);
    err = err || new Error("_socket has been disconnected by third party");

    self.on_socket_ended(err);

    self.__disconnecting__ = true;

    debugLog(" bytesRead    = ", self.bytesRead);
    debugLog(" bytesWritten = ", self.bytesWritten);
    _fulfill_pending_promises.call(self, new Error("Connection aborted - ended by server : " + (err ? err.message : "")));
};


/**
 * @method _install_socket
 * @param socket {Socket}
 * @protected
 */
TCP_transport.prototype._install_socket = function (socket) {

    assert(socket);
    const self = this;

    self.name = " Transport " + counter;
    counter += 1;

    self._socket = socket;

    // install packet assembler ...
    self.packetAssembler = new PacketAssembler({
        readMessageFunc: readRawMessageHeader,
        minimumSizeInBytes: self.headerSize
    });

    self.packetAssembler.on("message", function (message_chunk) {
        _on_message_received.call(self, message_chunk);
    });


    self._socket.on("data", function (data) {
        self.bytesRead += data.length;
        if (data.length > 0) {
            self.packetAssembler.feed(data);
        }

    }).on("close", function (had_error) {
        // istanbul ignore next
        if (doDebug) {
            debugLog(" SOCKET CLOSE : ".red, "had_error =".yellow,had_error.toString().cyan,self.name);
        }
        if (self._socket ) {
            debugLog("  remote address = ",self._socket.remoteAddress, " " , self._socket.remoteFamily, " ",self._socket.remotePort);
        }
        if (had_error) {
            if (self._socket) {
                self._socket.destroy();
            }
        }
        const err = had_error ? new Error("ERROR IN SOCKET") : null;
        self.on_socket_closed(err);

    }).on("end", function (err) {

        // istanbul ignore next
        if (doDebug) {
            debugLog(" SOCKET END : ".red, err ? err.message.yellow : "null", self._socket.name, self.name);
        }
        self._on_socket_ended_message(err);

    }).on("error", function (err) {
        // istanbul ignore next
        if (doDebug) {
            debugLog(" SOCKET ERROR : ".red, err.message.yellow, self._socket.name, self.name);
        }
        // node The "close" event will be called directly following this event.
    });

    const do_destroy_on_timeout =false;
    if (do_destroy_on_timeout) {
        // set socket timeout
        debugLog("setting client/server socket timeout to ",self.timeout);
        self._socket.setTimeout(self.timeout,function(){
            console.log(" connection has timed out (timeout =",self.timeout,")");
            self._socket.destroy();
        });

    }

};


/**
 * @method _install_one_time_message_receiver
 *
 * install a one time message receiver callback
 *
 * Rules:
 * * TCP_transport will not emit the ```message``` event, while the "one time message receiver" is in operation.
 * * the TCP_transport will wait for the next complete message chunk and call the provided callback func
 *   ```callback(null,messageChunk);```
 * * if a messageChunk is not received within ```TCP_transport.timeout``` or if the underlying socket reports an error,
 *    the callback function will be called with an Error.
 *
 * @param callback {Function} the callback function
 * @param callback.err {null|Error}
 * @param callback.messageChunk {Buffer|null}
 * @protected
 */
TCP_transport.prototype._install_one_time_message_receiver = function (callback) {

    const self = this;
    assert(!self._the_callback, "callback already set");
    assert(_.isFunction(callback));
    self._the_callback = callback;
    _start_timeout_timer.call(self);

};


/**
 * disconnect the TCP layer and close the underlying socket.
 * The ```"close"``` event will be emitted to the observers with err=null.
 *
 * @method disconnect
 * @async
 * @param callback
 */
TCP_transport.prototype.disconnect = function (callback) {

    assert(_.isFunction(callback), "expecting a callback function, but got " + callback);

    const self = this;
    if (self.__disconnecting__) {
        callback();
        return;
    }

    assert(!self.__disconnecting__, "TCP Transport has already been disconnected");
    self.__disconnecting__ = true;

    assert(!self._the_callback, "disconnect shall not be called while the 'one time message receiver' is in operation");
    _cleanup_timers.call(self);

    if (self._socket) {
        self._socket.end();
        self._socket.destroy();
//Xx        self._socket.removeAllListeners();
        self._socket = null;
    }

    setImmediate(function () {
        self.on_socket_ended(null);
        callback();
    });

};

TCP_transport.prototype.isValid = function() {
    const self = this;
    return self._socket !== null&& !self._socket.destroyed && !self.__disconnecting__;
};
exports.TCP_transport = TCP_transport;