larvit/smpp-js
3
0
Fork 0
Code Issues Pull Requests Actions Activity

Dependency updates and new API for logging

Browse Source
This commit is contained in:
Lilleman auf Larv
2018-08-22 17:05:29 +02:00
parent d9d3babe3b
commit 81be0a9aea
9 changed files with 198 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/session.js
+85 -73
View File
@@ -1,13 +1,14 @@
'use strict';
const topLogPrefix = 'larvitsmpp: lib/session.js: ',
LUtils = require('larvitutils'),
lUtils = new LUtils(),
events = require('events'),
moment = require('moment'),
utils = require('./utils'),
async = require('async'),
defs = require('./defs'),
uuid = require('uuid/v1'),
log = require('winston');
uuid = require('uuid/v1');
/**
* Send a response to an sms
@@ -29,7 +30,7 @@ function smsResp(status, cb) {
}
if (typeof cb !== 'function') {
cb = function () {};
cb = function () {};
}
if (sms.smsId === undefined) {
@@ -48,7 +49,7 @@ function smsResp(status, cb) {
if (sms.pduObjs === undefined) {
const err = new Error('No pdu objects found to base return PDU upon');
log.warn(logPrefix + err.message);
sms.log.warn(logPrefix + err.message);
return cb(err);
}
@@ -77,11 +78,11 @@ function smsResp(status, cb) {
}
function incOurSeqNr() {
this.ourSeqNr = this.ourSeqNr + 1;
this.ourSeqNr = this.ourSeqNr + 1;
// If we pass the maximum, start over at 1
if (this.ourSeqNr > 2147483646) {
this.ourSeqNr = 1;
this.ourSeqNr = 1;
}
}
@@ -90,14 +91,15 @@ function incOurSeqNr() {
* Always use this function to close the socket so we get it on log
*/
function closeSocket() {
const logPrefix = topLogPrefix + 'closeSocket() - ';
const logPrefix = topLogPrefix + 'closeSocket() - ',
that = this;
log.verbose(logPrefix + 'Closing socket for ' + this.sock.remoteAddress + ':' + this.sock.remotePort);
if (this.enqLinkTimer) {
log.debug(logPrefix + 'enqLinkTimer found, clearing.');
clearTimeout(this.enqLinkTimer);
that.log.verbose(logPrefix + 'Closing socket for ' + this.sock.remoteAddress + ':' + this.sock.remotePort);
if (that.enqLinkTimer) {
that.log.debug(logPrefix + 'enqLinkTimer found, clearing.');
clearTimeout(that.enqLinkTimer);
}
this.sock.destroy();
that.sock.destroy();
}
/**
@@ -113,7 +115,7 @@ function sockWrite(pdu, closeAfterSend) {
if ( ! Buffer.isBuffer(pdu)) {
utils.objToPdu(pdu, function (err, buffer) {
if (err) {
log.warn(logPrefix + 'Could not convert PDU to buffer');
that.log.warn(logPrefix + 'Could not convert PDU to buffer');
return that.closeSocket();
}
@@ -123,9 +125,9 @@ function sockWrite(pdu, closeAfterSend) {
}
try {
log.verbose(logPrefix + 'sending PDU. SeqNr: ' + pdu.readUInt32BE(12) + ' cmd: ' + defs.cmdsById[pdu.readUInt32BE(4)].command + ' cmdStatus: ' + defs.errorsById[parseInt(pdu.readUInt32BE(8))] + ' hex: ' + pdu.toString('hex'));
that.log.verbose(logPrefix + 'sending PDU. SeqNr: ' + pdu.readUInt32BE(12) + ' cmd: ' + defs.cmdsById[pdu.readUInt32BE(4)].command + ' cmdStatus: ' + defs.errorsById[parseInt(pdu.readUInt32BE(8))] + ' hex: ' + pdu.toString('hex'));
} catch (err) {
log.error(logPrefix + 'PDU buffer is invalid. Buffer hex: "' + pdu.toString('hex') + '"');
that.log.error(logPrefix + 'PDU buffer is invalid. Buffer hex: "' + pdu.toString('hex') + '"');
return;
}
@@ -161,7 +163,7 @@ function send(pdu, closeAfterSend, cb) {
// Make sure the sequence number is set and is correct
pduObj.seqNr = this.ourSeqNr;
log.debug(logPrefix + 'Sending PDU to remote. pduObj: ' + JSON.stringify(pduObj));
that.log.debug(logPrefix + 'Sending PDU to remote. pduObj: ' + JSON.stringify(pduObj));
// If closeAndSend is omitted, put cb in its place
if (typeof closeAfterSend === 'function') {
@@ -177,13 +179,13 @@ function send(pdu, closeAfterSend, cb) {
// Response PDUs are not allowed with the send() command, they should use the sendReturn()
if (pduObj.cmdName.substring(pduObj.cmdName - 5) === '_resp') {
const err = new Error('Given pduObj is a response, use sendReturn() instead. cmdName: ' + pduObj.cmdName);
log.verbose(logPrefix + err.message);
that.log.verbose(logPrefix + err.message);
return cb(err);
}
// When the return is fetched, call the cb
that.on('incomingPduObj' + pduObj.seqNr, function (incPduObj) {
log.debug(logPrefix + 'this.on(incomingPduObj) - cmdName: ' + incPduObj.cmdName + ' seqNr: ' + incPduObj.seqNr + ' cmdStatus: ' + incPduObj.cmdStatus);
that.log.debug(logPrefix + 'this.on(incomingPduObj) - cmdName: ' + incPduObj.cmdName + ' seqNr: ' + incPduObj.seqNr + ' cmdStatus: ' + incPduObj.cmdStatus);
// Make sure this is the actual response to the sent PDU
if (incPduObj.isResp() && incPduObj.seqNr === pduObj.seqNr) {
@@ -194,7 +196,7 @@ function send(pdu, closeAfterSend, cb) {
}
} else {
const err = new Error('Event triggered but incoming PDU is not a response or seqNr does not match. isResp: ' + incPduObj.isResp().toString() + ' incSeqNr: ' + incPduObj.seqNr + ' expected seqNr: ' + pduObj.seqNr);
log.warn(logPrefix + 'this.on(incomingPduObj) - ' + err.message);
that.log.warn(logPrefix + 'this.on(incomingPduObj) - ' + err.message);
cb(err);
}
});
@@ -219,7 +221,7 @@ function sendReturn(pdu, status, params, closeAfterSend, cb) {
const logPrefix = topLogPrefix + 'sendReturn() - ',
that = this;
log.silly(logPrefix + 'ran');
that.log.silly(logPrefix + 'ran');
if (typeof params === 'function') {
cb = params;
@@ -238,13 +240,13 @@ function sendReturn(pdu, status, params, closeAfterSend, cb) {
utils.pduReturn(pdu, status, params, function (err, retPdu) {
if (err) {
log.error(logPrefix + 'Could not create return PDU: ' + err.message);
that.log.error(logPrefix + 'Could not create return PDU: ' + err.message);
that.closeSocket();
return cb(err);
}
log.silly(logPrefix + 'Sending return PDU: ' + retPdu.toString('hex'));
that.log.silly(logPrefix + 'Sending return PDU: ' + retPdu.toString('hex'));
that.sockWrite(retPdu, closeAfterSend);
cb(null, retPdu);
});
@@ -264,7 +266,8 @@ function sendReturn(pdu, status, params, closeAfterSend, cb) {
*/
function sendSms(smsOptions, cb) {
const logPrefix = topLogPrefix + 'sendSms() - ',
pduObj = {};
pduObj = {},
that = this;
pduObj.cmdName = 'submit_sm';
pduObj.params = {
@@ -276,27 +279,27 @@ function sendSms(smsOptions, cb) {
// Flash messages overrides default data_coding
if (smsOptions.flash) {
log.debug(logPrefix + 'Flash SMS detected, set data_coding to 0x10!');
pduObj.params.data_coding = 0x10;
that.log.debug(logPrefix + 'Flash SMS detected, set data_coding to 0x10!');
pduObj.params.data_coding = 0x10;
}
// Request DLRs!
if (smsOptions.dlr) {
pduObj.params.registered_delivery = 0x01;
pduObj.params.registered_delivery = 0x01;
}
// Check if we must split this message into multiple
if (utils.bitCount(smsOptions.message) > 1120) {
log.debug(logPrefix + 'Message larger than 1120 bits, send it as long message!');
that.log.debug(logPrefix + 'Message larger than 1120 bits, send it as long message!');
this.sendLongSms(smsOptions, cb);
that.sendLongSms(smsOptions, cb);
return;
}
log.debug(logPrefix + 'pduObj: ' + JSON.stringify(pduObj));
that.log.debug(logPrefix + 'pduObj: ' + JSON.stringify(pduObj));
this.send(pduObj, function (err, retPduObj) {
that.send(pduObj, function (err, retPduObj) {
if (typeof cb === 'function') {
cb(err, [retPduObj.params.message_id], [retPduObj]);
}
@@ -320,7 +323,7 @@ function sendLongSms(smsOptions, cb) {
encoding = defs.encodings.detect(smsOptions.message), // Set encoding once for all message parts
smsIds = [],
that = this,
msgs = utils.splitMsg(smsOptions.message);
msgs = utils.splitMsg(smsOptions.message);
function sendPart(i) {
const pduObj = {
@@ -338,19 +341,19 @@ function sendLongSms(smsOptions, cb) {
// Request DLRs!
if (smsOptions.dlr) {
pduObj.params.registered_delivery = 0x01;
pduObj.params.registered_delivery = 0x01;
}
log.debug(logPrefix + 'pduObj: ' + JSON.stringify(pduObj));
that.log.debug(logPrefix + 'pduObj: ' + JSON.stringify(pduObj));
that.send(pduObj, function (err, retPduObj) {
smsIds.push(retPduObj.params.message_id);
retPduObjs.push(retPduObj);
log.silly(logPrefix + 'Got cb from that.send()');
that.log.silly(logPrefix + 'Got cb from that.send()');
if (typeof cb === 'function' && smsIds.length === msgs.length) {
log.silly(logPrefix + 'All cbs returned, run the parent cb.');
that.log.silly(logPrefix + 'All cbs returned, run the parent cb.');
cb(err, smsIds, retPduObjs);
}
});
@@ -420,7 +423,7 @@ function checkLongSmses() {
smsObj = {},
that = this;
log.silly(logPrefix + 'Running');
that.log.silly(logPrefix + 'Running');
// Call when complete SMS is received
function smsReceived() {
@@ -436,7 +439,7 @@ function checkLongSmses() {
// All parts are accounted for! Emit sms event and clear from tmp storage
if (smsGroup.partsCount === smsGroup.pduObjs.length) {
log.debug(logPrefix + 'All parts accounted for in smsGroupId "' + smsGroupId + '", emitting sms event.');
that.log.debug(logPrefix + 'All parts accounted for in smsGroupId "' + smsGroupId + '", emitting sms event.');
// These are needed for references here and there in functions
smsObj.session = that;
@@ -449,6 +452,7 @@ function checkLongSmses() {
smsObj.dlr = Boolean(smsGroup.pduObjs[0].pduObj.params.registered_delivery);
smsObj.sendResp = smsResp;
smsObj.sendDlr = utils.smsDlr;
smsObj.log = that.log;
// Concatenate all the parts messages to one and set references to the session
@@ -464,7 +468,7 @@ function checkLongSmses() {
}
smsReceived();
} else if (moment(new Date()).diff(smsGroup.created, 'hours') > 24) {
log.info(logPrefix + 'smsGroupId "' + smsGroupId + '" is removed from this.longSmses due to being older than 24 hours.');
that.log.info(logPrefix + 'smsGroupId "' + smsGroupId + '" is removed from this.longSmses due to being older than 24 hours.');
delete this.longSmses[smsGroupId];
}
@@ -474,18 +478,25 @@ function checkLongSmses() {
/**
* Generic session function
*
* @param {object} sock - socket object
* @param {object} options - {sock, log}
* @return {object} (returnObj)
*/
function session(sock) {
const logPrefix = topLogPrefix + 'session() - socket address: ' + sock.remoteAddress + ':' + sock.remotePort + ' - ',
function session(options) {
const logPrefix = topLogPrefix + 'session() - socket address: ' + options.sock.remoteAddress + ':' + options.sock.remotePort + ' - ',
returnObj = new events.EventEmitter();
log.silly(logPrefix + 'New session started');
if (! options.log) {
options.log = new lUtils.Log();
}
utils.log = options.log;
returnObj.log = options.log;
returnObj.log.silly(logPrefix + 'New session started');
returnObj.loggedIn = false;
returnObj.ourSeqNr = 1; // Sequence number used for commands initiated from us
returnObj.sock = sock; // Make the socket transparent via the returned emitter
returnObj.sock = options.sock; // Make the socket transparent via the returned emitter
returnObj.incOurSeqNr = incOurSeqNr;
returnObj.closeSocket = closeSocket;
returnObj.sockWrite = sockWrite;
@@ -517,7 +528,7 @@ function session(sock) {
// TLV message_state must exists
if (pduObj.tlvs.message_state === undefined) {
log.info(thisLogPrefix + 'TLV message_state is missing. SeqNr: ' + pduObj.seqNr);
returnObj.log.info(thisLogPrefix + 'TLV message_state is missing. SeqNr: ' + pduObj.seqNr);
returnObj.sendReturn(pduObj, 'ESME_RINVTLVSTREAM');
return;
@@ -525,7 +536,7 @@ function session(sock) {
// TLV message_state needs to be valid
if (defs.constsById.MESSAGE_STATE[pduObj.tlvs.message_state.tagValue] === undefined) {
log.info(thisLogPrefix + 'Invalid TLV message_state: "' + pduObj.tlvs.message_state.tagValue + '". SeqNr: ' + pduObj.seqNr);
returnObj.log.info(thisLogPrefix + 'Invalid TLV message_state: "' + pduObj.tlvs.message_state.tagValue + '". SeqNr: ' + pduObj.seqNr);
returnObj.sendReturn(pduObj, 'ESME_RINVTLVSTREAM');
return;
@@ -533,7 +544,7 @@ function session(sock) {
// TLV receipted_message_id must exist
if (pduObj.tlvs.receipted_message_id === undefined) {
log.info(thisLogPrefix + 'TLV receipted_message_id is missing. SeqNr: ' + pduObj.seqNr);
returnObj.log.info(thisLogPrefix + 'TLV receipted_message_id is missing. SeqNr: ' + pduObj.seqNr);
returnObj.sendReturn(pduObj, 'ESME_RINVTLVSTREAM');
return;
@@ -551,7 +562,7 @@ function session(sock) {
returnObj.handleCmd.enquire_link = function enquire_link(pduObj) {
const thisLogPrefix = logPrefix + 'enquire_link() - ';
log.silly(thisLogPrefix + 'Enquiring link');
returnObj.log.silly(thisLogPrefix + 'Enquiring link');
returnObj.resetEnqLinkTimer();
returnObj.sendReturn(pduObj);
};
@@ -561,13 +572,13 @@ function session(sock) {
const thisLogPrefix = logPrefix + 'submit_sm() - ',
smsObj = {};
log.silly(thisLogPrefix + 'ran');
returnObj.log.silly(thisLogPrefix + 'ran');
// If esm_class is 0x40 it means this is just a part of a larger message
//if (pduObj.params.esm_class === 0x40) {
// Fix: esm_class can be combination of bits. We need to extract 0x40 and then compare
if ((pduObj.params.esm_class & 0x40) === 0x40) {
log.debug(thisLogPrefix + 'long sms detected, esm_class 0x40.');
returnObj.log.debug(thisLogPrefix + 'long sms detected, esm_class 0x40.');
returnObj.longSms(pduObj);
return; // Long messages should not get handled here at all, so cancel execution here
}
@@ -582,12 +593,13 @@ function session(sock) {
smsObj.dlr = Boolean(pduObj.params.registered_delivery);
smsObj.sendResp = smsResp;
smsObj.sendDlr = utils.smsDlr;
smsObj.log = returnObj.log;
if (pduObj.params.data_coding === 0x10) {
smsObj.flash = true;
}
log.silly(thisLogPrefix + 'Emitting sms object');
returnObj.log.silly(thisLogPrefix + 'Emitting sms object');
returnObj.emit('sms', smsObj);
};
@@ -601,18 +613,18 @@ function session(sock) {
returnObj.login = function login() {
const thisLogPrefix = logPrefix + 'login() - ';
log.info(thisLogPrefix + 'Dummy login function ran, this might be a mistake');
returnObj.loggedIn = true;
returnObj.log.info(thisLogPrefix + 'Dummy login function ran, this might be a mistake');
returnObj.loggedIn = true;
};
// Dummy method - should be used by serverSession or clientSession
returnObj.resetEnqLinkTimer = function resetEnqLinkTimer() {
const thisLogPrefix = logPrefix + 'resetEnqLinkTimer() - ';
log.silly(thisLogPrefix + 'Resetting the kill timer');
returnObj.log.silly(thisLogPrefix + 'Resetting the kill timer');
};
// Unbind this session
returnObj.unbind = function () {
returnObj.unbind = function unbind() {
returnObj.send({
'cmdName': 'unbind'
}, true);
@@ -623,7 +635,7 @@ function session(sock) {
returnObj.dataQueue = new Buffer(0);
// Add a 'data' event handler to this instance of socket
sock.on('data', function (data) {
options.sock.on('data', function (data) {
const thisLogPrefix = logPrefix + 'sock.on(data) - ';
// Pass the data along to the returnObj
@@ -632,7 +644,7 @@ function session(sock) {
// Reset the enquire link timer
returnObj.resetEnqLinkTimer();
log.debug(thisLogPrefix + 'Incoming data: ' + data.toString('hex'));
returnObj.log.debug(thisLogPrefix + 'Incoming data: ' + data.toString('hex'));
// Add this data to the dataQueue for processing
returnObj.dataQueue = Buffer.concat([returnObj.dataQueue, data]);
@@ -641,23 +653,23 @@ function session(sock) {
while (returnObj.dataQueue.length > 4) {
const cmdLength = parseInt(returnObj.dataQueue.readUInt32BE(0)); // Get this commands length
let pdu;
// Malformed PDU with command length 0
if (cmdLength <= 0) {
// Since PDU is Malformed we need to discard buffer.
log.silly(thisLogPrefix + 'Malformed PDU with 0 Length. Discarding buffer.');
returnObj.dataQueue = returnObj.dataQueue.slice(0, returnObj.dataQueue.length);
returnObj.log.silly(thisLogPrefix + 'Malformed PDU with 0 Length. Discarding buffer.');
returnObj.dataQueue = returnObj.dataQueue.slice(0, returnObj.dataQueue.length);
// Since PDU is malformed we need to close socket since we cannot trust data from now on.
returnObj.closeSocket();
}
let pdu;
log.silly(thisLogPrefix + 'Processing ' + cmdLength + ' bytes of data');
returnObj.log.silly(thisLogPrefix + 'Processing ' + cmdLength + ' bytes of data');
// If there is at least enough bytes in the dataQueue to fill this PDU, do it!
if (cmdLength <= returnObj.dataQueue.length) {
log.silly(thisLogPrefix + 'Full PDU found in dataQueue, processing ' + cmdLength + ' bytes of queue total ' + returnObj.dataQueue.length + ' bytes');
returnObj.log.silly(thisLogPrefix + 'Full PDU found in dataQueue, processing ' + cmdLength + ' bytes of queue total ' + returnObj.dataQueue.length + ' bytes');
// Slice up the dataQueue buffer to this commands length
pdu = returnObj.dataQueue.slice(0, cmdLength);
@@ -667,19 +679,19 @@ function session(sock) {
returnObj.emit('incomingPdu', pdu);
} else {
log.debug(thisLogPrefix + 'Tried to process ' + cmdLength + ' bytes, but only ' + returnObj.dataQueue.length + ' bytes found. Awaiting more data. Current data in queue: ' + returnObj.dataQueue.toString('hex'));
returnObj.log.debug(thisLogPrefix + 'Tried to process ' + cmdLength + ' bytes, but only ' + returnObj.dataQueue.length + ' bytes found. Awaiting more data. Current data in queue: ' + returnObj.dataQueue.toString('hex'));
break;
}
if (returnObj.dataQueue.length === 0) {
log.silly(thisLogPrefix + 'All queue handled, breaking while loop.');
returnObj.log.silly(thisLogPrefix + 'All queue handled, breaking while loop.');
break;
}
// If the command length is larger than the queue, we need to wait for more data. Stop processing!
if (cmdLength > returnObj.dataQueue) {
log.debug(thisLogPrefix + 'Incomplete PDU found in dataQueue, waiting for more data to continue. Current cmdLength: ' + cmdLength + ' current queue: ' + returnObj.dataQueue.toString('hex'));
returnObj.log.debug(thisLogPrefix + 'Incomplete PDU found in dataQueue, waiting for more data to continue. Current cmdLength: ' + cmdLength + ' current queue: ' + returnObj.dataQueue.toString('hex'));
break;
}
}
@@ -691,11 +703,11 @@ function session(sock) {
utils.pduToObj(pdu, function (err, pduObj) {
if (err) {
log.warn(thisLogPrefix + 'Invalid PDU, closing socket.');
returnObj.log.warn(thisLogPrefix + 'Invalid PDU, closing socket.');
returnObj.closeSocket();
} else {
log.verbose(thisLogPrefix + 'Incoming PDU parsed. Seqnr: ' + pduObj.seqNr + ' cmd: ' + pduObj.cmdName + ' cmdStatus: ' + pduObj.cmdStatus + ' hex: ' + pdu.toString('hex'));
returnObj.log.verbose(thisLogPrefix + 'Incoming PDU parsed. Seqnr: ' + pduObj.seqNr + ' cmd: ' + pduObj.cmdName + ' cmdStatus: ' + pduObj.cmdStatus + ' hex: ' + pdu.toString('hex'));
if (pduObj.isResp()) {
// We do this so we can remove the dynamic event listeners to not have a memory leak
@@ -711,23 +723,23 @@ function session(sock) {
});
// Add a 'close' event handler to this instance of socket
sock.on('close', function () {
options.sock.on('close', function () {
const thisLogPrefix = logPrefix + 'sock.on(close) - ';
returnObj.emit('close');
if (returnObj.enqLinkTimer) {
log.debug(thisLogPrefix + 'enqLinkTimer found, clearing.');
returnObj.log.debug(thisLogPrefix + 'enqLinkTimer found, clearing.');
clearTimeout(returnObj.enqLinkTimer);
}
log.debug(thisLogPrefix + 'socket closed');
returnObj.log.debug(thisLogPrefix + 'socket closed');
});
sock.on('error', function () {
options.sock.on('error', function () {
const thisLogPrefix = logPrefix + 'sock.on(error) - ';
log.warn(thisLogPrefix + 'Socket error detected!');
returnObj.log.warn(thisLogPrefix + 'Socket error detected!');
if (returnObj.enqLinkTimer) {
log.debug(thisLogPrefix + 'enqLinkTimer found, clearing.');
returnObj.log.debug(thisLogPrefix + 'enqLinkTimer found, clearing.');
clearTimeout(returnObj.enqLinkTimer);
}
});