in content/lib/paho-mqtt.js [1255:1442]
ClientImpl.prototype._handleMessage = function(wireMessage) {
this._trace("Client._handleMessage", wireMessage);
try {
switch(wireMessage.type) {
case MESSAGE_TYPE.CONNACK:
this._connectTimeout.cancel();
if (this._reconnectTimeout)
this._reconnectTimeout.cancel();
// If we have started using clean session then clear up the local state.
if (this.connectOptions.cleanSession) {
for (var key in this._sentMessages) {
var sentMessage = this._sentMessages[key];
localStorage.removeItem("Sent:"+this._localKey+sentMessage.messageIdentifier);
}
this._sentMessages = {};
for (var key in this._receivedMessages) {
var receivedMessage = this._receivedMessages[key];
localStorage.removeItem("Received:"+this._localKey+receivedMessage.messageIdentifier);
}
this._receivedMessages = {};
}
// Client connected and ready for business.
if (wireMessage.returnCode === 0) {
this.connected = true;
// Jump to the end of the list of uris and stop looking for a good host.
if (this.connectOptions.uris)
this.hostIndex = this.connectOptions.uris.length;
} else {
this._disconnected(ERROR.CONNACK_RETURNCODE.code , format(ERROR.CONNACK_RETURNCODE, [wireMessage.returnCode, CONNACK_RC[wireMessage.returnCode]]));
break;
}
// Resend messages.
var sequencedMessages = [];
for (var msgId in this._sentMessages) {
if (this._sentMessages.hasOwnProperty(msgId))
sequencedMessages.push(this._sentMessages[msgId]);
}
// Also schedule qos 0 buffered messages if any
if (this._buffered_msg_queue.length > 0) {
var msg = null;
var fifo = this._buffered_msg_queue.reverse();
while ((msg = fifo.pop())) {
sequencedMessages.push(msg);
if (this.onMessageDelivered)
this._notify_msg_sent[msg] = this.onMessageDelivered(msg.payloadMessage);
}
}
// Sort sentMessages into the original sent order.
var sequencedMessages = sequencedMessages.sort(function(a,b) {return a.sequence - b.sequence;} );
for (var i=0, len=sequencedMessages.length; i<len; i++) {
var sentMessage = sequencedMessages[i];
if (sentMessage.type == MESSAGE_TYPE.PUBLISH && sentMessage.pubRecReceived) {
var pubRelMessage = new WireMessage(MESSAGE_TYPE.PUBREL, {messageIdentifier:sentMessage.messageIdentifier});
this._schedule_message(pubRelMessage);
} else {
this._schedule_message(sentMessage);
}
}
// Execute the connectOptions.onSuccess callback if there is one.
// Will also now return if this connection was the result of an automatic
// reconnect and which URI was successfully connected to.
if (this.connectOptions.onSuccess) {
this.connectOptions.onSuccess({invocationContext:this.connectOptions.invocationContext});
}
var reconnected = false;
if (this._reconnecting) {
reconnected = true;
this._reconnectInterval = 1;
this._reconnecting = false;
}
// Execute the onConnected callback if there is one.
this._connected(reconnected, this._wsuri);
// Process all queued messages now that the connection is established.
this._process_queue();
break;
case MESSAGE_TYPE.PUBLISH:
this._receivePublish(wireMessage);
break;
case MESSAGE_TYPE.PUBACK:
var sentMessage = this._sentMessages[wireMessage.messageIdentifier];
// If this is a re flow of a PUBACK after we have restarted receivedMessage will not exist.
if (sentMessage) {
delete this._sentMessages[wireMessage.messageIdentifier];
localStorage.removeItem("Sent:"+this._localKey+wireMessage.messageIdentifier);
if (this.onMessageDelivered)
this.onMessageDelivered(sentMessage.payloadMessage);
}
break;
case MESSAGE_TYPE.PUBREC:
var sentMessage = this._sentMessages[wireMessage.messageIdentifier];
// If this is a re flow of a PUBREC after we have restarted receivedMessage will not exist.
if (sentMessage) {
sentMessage.pubRecReceived = true;
var pubRelMessage = new WireMessage(MESSAGE_TYPE.PUBREL, {messageIdentifier:wireMessage.messageIdentifier});
this.store("Sent:", sentMessage);
this._schedule_message(pubRelMessage);
}
break;
case MESSAGE_TYPE.PUBREL:
var receivedMessage = this._receivedMessages[wireMessage.messageIdentifier];
localStorage.removeItem("Received:"+this._localKey+wireMessage.messageIdentifier);
// If this is a re flow of a PUBREL after we have restarted receivedMessage will not exist.
if (receivedMessage) {
this._receiveMessage(receivedMessage);
delete this._receivedMessages[wireMessage.messageIdentifier];
}
// Always flow PubComp, we may have previously flowed PubComp but the server lost it and restarted.
var pubCompMessage = new WireMessage(MESSAGE_TYPE.PUBCOMP, {messageIdentifier:wireMessage.messageIdentifier});
this._schedule_message(pubCompMessage);
break;
case MESSAGE_TYPE.PUBCOMP:
var sentMessage = this._sentMessages[wireMessage.messageIdentifier];
delete this._sentMessages[wireMessage.messageIdentifier];
localStorage.removeItem("Sent:"+this._localKey+wireMessage.messageIdentifier);
if (this.onMessageDelivered)
this.onMessageDelivered(sentMessage.payloadMessage);
break;
case MESSAGE_TYPE.SUBACK:
var sentMessage = this._sentMessages[wireMessage.messageIdentifier];
if (sentMessage) {
if(sentMessage.timeOut)
sentMessage.timeOut.cancel();
// This will need to be fixed when we add multiple topic support
if (wireMessage.returnCode[0] === 0x80) {
if (sentMessage.onFailure) {
sentMessage.onFailure(wireMessage.returnCode);
}
} else if (sentMessage.onSuccess) {
sentMessage.onSuccess(wireMessage.returnCode);
}
delete this._sentMessages[wireMessage.messageIdentifier];
}
break;
case MESSAGE_TYPE.UNSUBACK:
var sentMessage = this._sentMessages[wireMessage.messageIdentifier];
if (sentMessage) {
if (sentMessage.timeOut)
sentMessage.timeOut.cancel();
if (sentMessage.callback) {
sentMessage.callback();
}
delete this._sentMessages[wireMessage.messageIdentifier];
}
break;
case MESSAGE_TYPE.PINGRESP:
/* The sendPinger or receivePinger may have sent a ping, the receivePinger has already been reset. */
this.sendPinger.reset();
break;
case MESSAGE_TYPE.DISCONNECT:
// Clients do not expect to receive disconnect packets.
this._disconnected(ERROR.INVALID_MQTT_MESSAGE_TYPE.code , format(ERROR.INVALID_MQTT_MESSAGE_TYPE, [wireMessage.type]));
break;
default:
this._disconnected(ERROR.INVALID_MQTT_MESSAGE_TYPE.code , format(ERROR.INVALID_MQTT_MESSAGE_TYPE, [wireMessage.type]));
}
} catch (error) {
var errorStack = ((error.hasOwnProperty('stack') == 'undefined') ? error.stack.toString() : "No Error Stack Available");
this._disconnected(ERROR.INTERNAL_ERROR.code , format(ERROR.INTERNAL_ERROR, [error.message,errorStack]));
return;
}
};