diff --git a/lib/client.js b/lib/client.js index 4a39cbb..5abe11d 100644 --- a/lib/client.js +++ b/lib/client.js @@ -59,7 +59,7 @@ function clientSession(sock, options) { returnObj.login(); returnObj.resetEnqLinkTimer(); - returnObj.on('incomingPdu', function(pduObj) { + returnObj.on('incomingPduObj', function(pduObj) { if (pduObj.cmdName === 'deliver_sm') { returnObj.deliverSm(pduObj); } else if (pduObj.cmdName === 'enquire_link') { diff --git a/lib/server.js b/lib/server.js index 9f8943f..334ebde 100644 --- a/lib/server.js +++ b/lib/server.js @@ -93,7 +93,7 @@ function serverSession(sock, options) { returnObj.resetEnqLinkTimer(); - returnObj.on('incomingPdu', function(pduObj) { + returnObj.on('incomingPduObj', function(pduObj) { if (pduObj.cmdName === 'unbind') { returnObj.sendReturn(pduObj, 'ESME_ROK', undefined, true); } else if (returnObj.loggedIn === false) { diff --git a/lib/session.js b/lib/session.js index f7a8dfa..af55a4a 100644 --- a/lib/session.js +++ b/lib/session.js @@ -131,9 +131,9 @@ function session(sock) { } // When the return is fetched, call the callback - returnObj.on('incomingPdu' + pduObj.seqNr, function(incPduObj) { + returnObj.on('incomingPduObj' + pduObj.seqNr, function(incPduObj) { - log.debug('larvitsmpp: lib/session.js: session() - returnObj.send() - returnObj.on(incomingPdu) - cmdName: ' + incPduObj.cmdName + ' seqNr: ' + incPduObj.seqNr + ' cmdStatus: ' + incPduObj.cmdStatus); + log.debug('larvitsmpp: lib/session.js: session() - returnObj.send() - returnObj.on(incomingPduObj) - cmdName: ' + incPduObj.cmdName + ' seqNr: ' + incPduObj.seqNr + ' cmdStatus: ' + incPduObj.cmdStatus); // Clean up by removing this listener or else it will lurk along forever returnObj.removeAllListeners('incomingPdu' + pduObj.seqNr); @@ -146,7 +146,7 @@ function session(sock) { returnObj.closeSocket(); } } else { - err = new Error('larvitsmpp: lib/session.js: session() - returnObj.send() - returnObj.on(incomingPdu) - Event triggered but incoming PDU is not a response or seqNr does not match. isResponse: ' + incPduObj.isResponse().toString() + ' incSeqNr: ' + incPduObj.seqNr + ' expected seqNr: ' + pduObj.seqNr); + err = new Error('larvitsmpp: lib/session.js: session() - returnObj.send() - returnObj.on(incomingPduObj) - Event triggered but incoming PDU is not a response or seqNr does not match. isResponse: ' + incPduObj.isResponse().toString() + ' incSeqNr: ' + incPduObj.seqNr + ' expected seqNr: ' + pduObj.seqNr); log.warn(err.message); callback(err); } @@ -463,33 +463,73 @@ function session(sock) { }, true); }; + // Setup a data queue in case we only get partial data on the socket + // This way we can concatenate them later on + returnObj.dataQueue = new Buffer(0); + // Add a 'data' event handler to this instance of socket - sock.on('data', function(pduBuf) { + sock.on('data', function(data) { + var cmdLength, + pdu; + // Pass the data along to the returnObj - returnObj.emit('data', pduBuf); + returnObj.emit('data', data); // Reset the enquire link timer returnObj.resetEnqLinkTimer(); - log.silly('larvitsmpp: lib/session.js: session() - sock.on(data) - Incoming PDU: ' + pduBuf.toString('hex')); + log.silly('larvitsmpp: lib/session.js: session() - sock.on(data) - Incoming data: ' + data.toString('hex')); - utils.pduToObj(pduBuf, function(err, pduObj) { + // Add this data to the dataQueue for processing + returnObj.dataQueue = Buffer.concat([returnObj.dataQueue, data]); + + // Process queue + while (returnObj.dataQueue.length !== 0) { + // Get this commands length + cmdLength = parseInt(data.readUInt32BE(0)); + log.silly('larvitsmpp: lib/session.js: session() - sock.on(data) - 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('larvitsmpp: lib/session.js: session() - sock.on(data) - 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); + + // Slice off the command from the dataQueue + returnObj.dataQueue = returnObj.dataQueue.slice(cmdLength, returnObj.dataQueue.length); + + returnObj.emit('incomingPdu', pdu); + } + + // If the command length is larger than the queue, we need to wait for more data. Stop processing! + if (cmdLength > returnObj.dataQueue) { + log.debug('larvitsmpp: lib/session.js: session() - sock.on(data) - Incomplete PDU found in dataQueue, waiting for more data to continue.'); + + break; + } + } + }); + + returnObj.on('incomingPdu', function(pdu) { + utils.pduToObj(pdu, function(err, pduObj) { if (err) { - log.warn('larvitsmpp: lib/session.js: session() - Invalid PDU. ' + err.message); + log.warn('larvitsmpp: lib/session.js: session() - returnObj.on(incomingPdu) - Invalid PDU, closing socket.'); returnObj.closeSocket(); } else { - log.verbose('larvitsmpp: lib/session.js: session() - sock.on(data) - Incoming PDU. Seqnr: ' + pduObj.seqNr + ' cmd: ' + pduObj.cmdName + ' cmdStatus: ' + pduObj.cmdStatus + ' hex: ' + pduBuf.toString('hex')); + log.verbose('larvitsmpp: lib/session.js: session() - returnObj.on(incomingPdu) - Incoming PDU parsed. Seqnr: ' + pduObj.seqNr + ' cmd: ' + pduObj.cmdName + ' cmdStatus: ' + pduObj.cmdStatus + ' hex: ' + pdu.toString('hex')); if (pduObj.isResponse()) { // We do this so we can remove the dynamic event listeners to not have a memory leak - returnObj.emit('incomingPdu' + pduObj.seqNr, pduObj); + returnObj.emit('incomingPduObj' + pduObj.seqNr, pduObj); } else { - returnObj.emit('incomingPdu', pduObj); + returnObj.emit('incomingPduObj', pduObj); } } }); - }); + }); + // Add a 'close' event handler to this instance of socket sock.on('close', function() { diff --git a/lib/utils.js b/lib/utils.js index a34d625..4c36128 100644 --- a/lib/utils.js +++ b/lib/utils.js @@ -230,7 +230,7 @@ function pduToObj(pdu, stupidNullByte, callback) { // Lookup the command id in the definitions if (defs.cmdsById[retObj.cmdId] === undefined) { - err = new Error('Unknown PDU command id: ' + retObj.cmdId); + err = new Error('Unknown PDU command id: ' + retObj.cmdId + ' PDU buff in hex: ' + pdu.toString('hex')); } if (isNaN(retObj.seqNr)) {