Ian Craggs
Better timing
Diff: MQTTClient.h
- Revision:
- 46:e335fcc1a663
- Parent:
- 44:c299463ae853
- Child:
- 53:15b5a280d22d
--- a/MQTTClient.h Mon Aug 03 12:40:57 2015 +0000
+++ b/MQTTClient.h Tue Aug 18 09:57:19 2015 +0000
@@ -1,5 +1,5 @@
/*******************************************************************************
- * Copyright (c) 2014 IBM Corp.
+ * Copyright (c) 2014, 2015 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
@@ -12,6 +12,10 @@
*
* Contributors:
* Ian Craggs - initial API and implementation and/or initial documentation
+ * Ian Craggs - fix for bug 458512 - QoS 2 messages
+ * Ian Craggs - fix for bug 460389 - send loop uses wrong length
+ * Ian Craggs - fix for bug 464169 - clearing subscriptions
+ * Ian Craggs - fix for bug 464551 - enums and ints can be different size
*******************************************************************************/
#if !defined(MQTTCLIENT_H)
@@ -118,7 +122,7 @@
*/
int connect();
- /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack
+ /** MQTT Connect - send an MQTT connect packet down the network and wait for a Connack
* The nework object must be connected to the network endpoint before calling this
* @param options - connect options
* @return success code -
@@ -190,6 +194,7 @@
private:
+ void cleanSession();
int cycle(Timer& timer);
int waitfor(int packet_type, Timer& timer);
int keepalive();
@@ -239,6 +244,7 @@
unsigned short incomingQoS2messages[MAX_INCOMING_QOS2_MESSAGES];
bool isQoS2msgidFree(unsigned short id);
bool useQoS2msgid(unsigned short id);
+ void freeQoS2msgid(unsigned short id);
#endif
};
@@ -247,28 +253,35 @@
template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS>
+void MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::cleanSession()
+{
+ ping_outstanding = false;
+ for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
+ messageHandlers[i].topicFilter = 0;
+ isconnected = false;
+
+#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
+ inflightMsgid = 0;
+ inflightQoS = QOS0;
+#endif
+
+#if MQTTCLIENT_QOS2
+ pubrel = false;
+ for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
+ incomingQoS2messages[i] = 0;
+#endif
+}
+
+
+template<class Network, class Timer, int a, int MAX_MESSAGE_HANDLERS>
MQTT::Client<Network, Timer, a, MAX_MESSAGE_HANDLERS>::Client(Network& network, unsigned int command_timeout_ms) : ipstack(network), packetid()
{
last_sent = Timer();
last_received = Timer();
- ping_outstanding = false;
- for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
- messageHandlers[i].topicFilter = 0;
this->command_timeout_ms = command_timeout_ms;
- isconnected = false;
-
-#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
- inflightMsgid = 0;
- inflightQoS = QOS0;
-#endif
+ cleanSession();
+}
-
-#if MQTTCLIENT_QOS2
- pubrel = false;
- for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
- incomingQoS2messages[i] = 0;
-#endif
-}
#if MQTTCLIENT_QOS2
template<class Network, class Timer, int a, int b>
@@ -296,6 +309,20 @@
}
return false;
}
+
+
+template<class Network, class Timer, int a, int b>
+void MQTT::Client<Network, Timer, a, b>::freeQoS2msgid(unsigned short id)
+{
+ for (int i = 0; i < MAX_INCOMING_QOS2_MESSAGES; ++i)
+ {
+ if (incomingQoS2messages[i] == id)
+ {
+ incomingQoS2messages[i] = 0;
+ return;
+ }
+ }
+}
#endif
@@ -307,7 +334,7 @@
while (sent < length && !timer.expired())
{
- rc = ipstack.write(&sendbuf[sent], length, timer.left_ms());
+ rc = ipstack.write(&sendbuf[sent], length - sent, timer.left_ms());
if (rc < 0) // there was an error writing the data
break;
sent += rc;
@@ -322,8 +349,8 @@
rc = FAILURE;
#if defined(MQTT_DEBUG)
- char printbuf[50];
- DEBUG("Rc %d from sending packet %s\n", rc, MQTTPacket_toString(printbuf, sizeof(printbuf), sendbuf, length));
+ char printbuf[150];
+ DEBUG("Rc %d from sending packet %s\n", rc, MQTTFormat_toServerString(printbuf, sizeof(printbuf), sendbuf, length));
#endif
return rc;
}
@@ -364,8 +391,8 @@
* @param timeout the max time to wait for the packet read to complete, in milliseconds
* @return the MQTT packet type, or -1 if none
*/
-template<class Network, class Timer, int a, int b>
-int MQTT::Client<Network, Timer, a, b>::readPacket(Timer& timer)
+template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
+int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::readPacket(Timer& timer)
{
int rc = FAILURE;
MQTTHeader header = {0};
@@ -381,6 +408,12 @@
decodePacket(&rem_len, timer.left_ms());
len += MQTTPacket_encode(readbuf + 1, rem_len); /* put the original remaining length into the buffer */
+ if (rem_len > (MAX_MQTT_PACKET_SIZE - len))
+ {
+ rc = BUFFER_OVERFLOW;
+ goto exit;
+ }
+
/* 3. read the rest of the buffer using a callback to supply the rest of the data */
if (rem_len > 0 && (ipstack.read(readbuf + len, rem_len, timer.left_ms()) != rem_len))
goto exit;
@@ -392,8 +425,11 @@
exit:
#if defined(MQTT_DEBUG)
- char printbuf[50];
- DEBUG("Rc %d from receiving packet %s\n", rc, MQTTPacket_toString(printbuf, sizeof(printbuf), readbuf, len));
+ if (rc >= 0)
+ {
+ char printbuf[50];
+ DEBUG("Rc %d from receiving packet %s\n", rc, MQTTFormat_toClientString(printbuf, sizeof(printbuf), readbuf, len));
+ }
#endif
return rc;
}
@@ -473,7 +509,7 @@
timer.countdown_ms(timeout_ms);
while (!timer.expired())
{
- if (cycle(timer) == FAILURE)
+ if (cycle(timer) < 0)
{
rc = FAILURE;
break;
@@ -490,23 +526,30 @@
/* get one piece of work off the wire and one pass through */
// read the socket, see what work is due
- unsigned short packet_type = readPacket(timer);
+ int packet_type = readPacket(timer);
int len = 0,
rc = SUCCESS;
switch (packet_type)
{
+ case FAILURE:
+ case BUFFER_OVERFLOW:
+ rc = packet_type;
+ break;
case CONNACK:
case PUBACK:
case SUBACK:
break;
case PUBLISH:
- MQTTString topicName;
+ {
+ MQTTString topicName = MQTTString_initializer;
Message msg;
- if (MQTTDeserialize_publish((unsigned char*)&msg.dup, (int*)&msg.qos, (unsigned char*)&msg.retained, (unsigned short*)&msg.id, &topicName,
+ int intQoS;
+ if (MQTTDeserialize_publish((unsigned char*)&msg.dup, &intQoS, (unsigned char*)&msg.retained, (unsigned short*)&msg.id, &topicName,
(unsigned char**)&msg.payload, (int*)&msg.payloadlen, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
goto exit;
+ msg.qos = (enum QoS)intQoS;
#if MQTTCLIENT_QOS2
if (msg.qos != QOS2)
#endif
@@ -518,7 +561,7 @@
deliverMessage(topicName, msg);
else
WARN("Maximum number of incoming QoS2 messages exceeded");
- }
+ }
#endif
#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
if (msg.qos != QOS0)
@@ -536,19 +579,25 @@
}
break;
#endif
+ }
#if MQTTCLIENT_QOS2
case PUBREC:
+ case PUBREL:
unsigned short mypacketid;
unsigned char dup, type;
if (MQTTDeserialize_ack(&type, &dup, &mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) != 1)
rc = FAILURE;
- else if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREL, 0, mypacketid)) <= 0)
+ else if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE,
+ (packet_type == PUBREC) ? PUBREL : PUBCOMP, 0, mypacketid)) <= 0)
rc = FAILURE;
else if ((rc = sendPacket(len, timer)) != SUCCESS) // send the PUBREL packet
rc = FAILURE; // there was a problem
if (rc == FAILURE)
goto exit; // there was a problem
+ if (packet_type == PUBREL)
+ freeQoS2msgid(mypacketid);
break;
+
case PUBCOMP:
break;
#endif
@@ -579,7 +628,7 @@
{
if (!ping_outstanding)
{
- Timer timer = Timer(1000);
+ Timer timer(1000);
int len = MQTTSerialize_pingreq(sendbuf, MAX_MQTT_PACKET_SIZE);
if (len > 0 && (rc = sendPacket(len, timer)) == SUCCESS) // send the ping packet
ping_outstanding = true;
@@ -611,7 +660,7 @@
template<class Network, class Timer, int MAX_MQTT_PACKET_SIZE, int b>
int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::connect(MQTTPacket_connectData& options)
{
- Timer connect_timer = Timer(command_timeout_ms);
+ Timer connect_timer(command_timeout_ms);
int rc = FAILURE;
int len = 0;
@@ -639,10 +688,10 @@
}
else
rc = FAILURE;
-
+
#if MQTTCLIENT_QOS2
- // resend an inflight publish
- if (inflightMsgid >0 && inflightQoS == QOS2 && pubrel)
+ // resend any inflight publish
+ if (inflightMsgid > 0 && inflightQoS == QOS2 && pubrel)
{
if ((len = MQTTSerialize_ack(sendbuf, MAX_MQTT_PACKET_SIZE, PUBREL, 0, inflightMsgid)) <= 0)
rc = FAILURE;
@@ -678,9 +727,9 @@
int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::subscribe(const char* topicFilter, enum QoS qos, messageHandler messageHandler)
{
int rc = FAILURE;
- Timer timer = Timer(command_timeout_ms);
+ Timer timer(command_timeout_ms);
int len = 0;
- MQTTString topic = {(char*)topicFilter, 0, 0};
+ MQTTString topic = {(char*)topicFilter, {0, 0}};
if (!isconnected)
goto exit;
@@ -716,7 +765,7 @@
exit:
if (rc != SUCCESS)
- isconnected = false;
+ cleanSession();
return rc;
}
@@ -725,8 +774,8 @@
int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, MAX_MESSAGE_HANDLERS>::unsubscribe(const char* topicFilter)
{
int rc = FAILURE;
- Timer timer = Timer(command_timeout_ms);
- MQTTString topic = {(char*)topicFilter, 0, 0};
+ Timer timer(command_timeout_ms);
+ MQTTString topic = {(char*)topicFilter, {0, 0}};
int len = 0;
if (!isconnected)
@@ -741,14 +790,26 @@
{
unsigned short mypacketid; // should be the same as the packetid above
if (MQTTDeserialize_unsuback(&mypacketid, readbuf, MAX_MQTT_PACKET_SIZE) == 1)
+ {
rc = 0;
+
+ // remove the subscription message handler associated with this topic, if there is one
+ for (int i = 0; i < MAX_MESSAGE_HANDLERS; ++i)
+ {
+ if (messageHandlers[i].topicFilter && strcmp(messageHandlers[i].topicFilter, topicFilter) == 0)
+ {
+ messageHandlers[i].topicFilter = 0;
+ break;
+ }
+ }
+ }
}
else
rc = FAILURE;
exit:
if (rc != SUCCESS)
- isconnected = false;
+ cleanSession();
return rc;
}
@@ -757,11 +818,11 @@
int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(int len, Timer& timer, enum QoS qos)
{
int rc;
-
+
if ((rc = sendPacket(len, timer)) != SUCCESS) // send the publish packet
goto exit; // there was a problem
-#if MQTTCLIENT_QOS1
+#if MQTTCLIENT_QOS1
if (qos == QOS1)
{
if (waitfor(PUBACK, timer) == PUBACK)
@@ -795,7 +856,7 @@
exit:
if (rc != SUCCESS)
- isconnected = false;
+ cleanSession();
return rc;
}
@@ -805,13 +866,13 @@
int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::publish(const char* topicName, void* payload, size_t payloadlen, unsigned short& id, enum QoS qos, bool retained)
{
int rc = FAILURE;
- Timer timer = Timer(command_timeout_ms);
+ Timer timer(command_timeout_ms);
MQTTString topicString = MQTTString_initializer;
int len = 0;
if (!isconnected)
goto exit;
-
+
topicString.cstring = (char*)topicName;
#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
@@ -823,7 +884,7 @@
topicString, (unsigned char*)payload, payloadlen);
if (len <= 0)
goto exit;
-
+
#if MQTTCLIENT_QOS1 || MQTTCLIENT_QOS2
if (!cleansession)
{
@@ -836,7 +897,7 @@
#endif
}
#endif
-
+
rc = publish(len, timer, qos);
exit:
return rc;
@@ -862,14 +923,17 @@
int MQTT::Client<Network, Timer, MAX_MQTT_PACKET_SIZE, b>::disconnect()
{
int rc = FAILURE;
- Timer timer = Timer(command_timeout_ms); // we might wait for incomplete incoming publishes to complete
+ Timer timer(command_timeout_ms); // we might wait for incomplete incoming publishes to complete
int len = MQTTSerialize_disconnect(sendbuf, MAX_MQTT_PACKET_SIZE);
if (len > 0)
rc = sendPacket(len, timer); // send the disconnect packet
- isconnected = false;
+ if (cleansession)
+ cleanSession();
+ else
+ isconnected = false;
return rc;
}
-#endif
+#endif
\ No newline at end of file