Handling processing of data chunks that is not arriving in order

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') {

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) {

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() {

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)) {