natas_hw
/
paho
mirror of https://github.com/eclipse/paho.mqtt.c
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Correct persisting of pubrel received publish messages in MQTT V5

This commit is contained in:
Ian Craggs 2019-11-26 14:20:17 +00:00
parent a2742412d2
commit ac3ae38069
7 changed files with 37 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/MQTTAsync.c
View File

@ -3871,11 +3871,11 @@ static MQTTPacket* MQTTAsync_cycle(int* sock, unsigned long timeout, int* rc)
(*(command->command.onFailure5))(command->command.context, &data);
}
MQTTAsync_freeCommand(command);
if (mqttversion >= MQTTVERSION_5)
MQTTProperties_free(&msgprops);
break;
}
}
if (mqttversion >= MQTTVERSION_5)
MQTTProperties_free(&msgprops);
}
}
else if (pack->header.bits.type == PUBREL)

28
src/MQTTPacket.c
View File

@ -91,7 +91,7 @@ pf new_packets[] =
static char* readUTFlen(char** pptr, char* enddata, int* len);
static int MQTTPacket_send_ack(int type, int msgid, int dup, networkHandles *net);
static int MQTTPacket_send_ack(int MQTTVersion, int type, int msgid, int dup, networkHandles *net);
/**
* Reads one MQTT packet from a socket.
@ -594,13 +594,14 @@ void MQTTPacket_freeAck(Ack* pack)
/**
* Send an MQTT acknowledgement packet down a socket.
* @param MQTTVersion the version of MQTT being used
* @param type the MQTT packet type e.g. SUBACK
* @param msgid the MQTT message id to use
* @param dup boolean - whether to set the MQTT DUP flag
* @param net the network handle to send the data to
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
static int MQTTPacket_send_ack(int type, int msgid, int dup, networkHandles *net)
static int MQTTPacket_send_ack(int MQTTVersion, int type, int msgid, int dup, networkHandles *net)
{
Header header;
int rc;
@ -614,7 +615,7 @@ static int MQTTPacket_send_ack(int type, int msgid, int dup, networkHandles *net
if (type == PUBREL)
header.bits.qos = 1;
writeInt(&ptr, msgid);
if ((rc = MQTTPacket_send(net, header, buf, 2, 1, MQTTVERSION_3_1_1)) != TCPSOCKET_INTERRUPTED)
if ((rc = MQTTPacket_send(net, header, buf, 2, 1, MQTTVersion)) != TCPSOCKET_INTERRUPTED)
free(buf);
FUNC_EXIT_RC(rc);
return rc;
@ -623,17 +624,18 @@ static int MQTTPacket_send_ack(int type, int msgid, int dup, networkHandles *net
/**
* Send an MQTT PUBACK packet down a socket.
* @param MQTTVersion the version of MQTT being used
* @param msgid the MQTT message id to use
* @param socket the open socket to send the data to
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_puback(int msgid, networkHandles* net, const char* clientID)
int MQTTPacket_send_puback(int MQTTVersion, int msgid, networkHandles* net, const char* clientID)
{
int rc = 0;
FUNC_ENTRY;
rc = MQTTPacket_send_ack(PUBACK, msgid, 0, net);
rc = MQTTPacket_send_ack(MQTTVersion, PUBACK, msgid, 0, net);
Log(LOG_PROTOCOL, 12, NULL, net->socket, clientID, msgid, rc);
FUNC_EXIT_RC(rc);
return rc;
@ -676,17 +678,18 @@ void MQTTPacket_freeUnsuback(Unsuback* pack)
/**
* Send an MQTT PUBREC packet down a socket.
* @param MQTTVersion the version of MQTT being used
* @param msgid the MQTT message id to use
* @param socket the open socket to send the data to
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_pubrec(int msgid, networkHandles* net, const char* clientID)
int MQTTPacket_send_pubrec(int MQTTVersion, int msgid, networkHandles* net, const char* clientID)
{
int rc = 0;
FUNC_ENTRY;
rc = MQTTPacket_send_ack(PUBREC, msgid, 0, net);
rc = MQTTPacket_send_ack(MQTTVersion, PUBREC, msgid, 0, net);
Log(LOG_PROTOCOL, 13, NULL, net->socket, clientID, msgid, rc);
FUNC_EXIT_RC(rc);
return rc;
@ -695,18 +698,19 @@ int MQTTPacket_send_pubrec(int msgid, networkHandles* net, const char* clientID)
/**
* Send an MQTT PUBREL packet down a socket.
* @param MQTTVersion the version of MQTT being used
* @param msgid the MQTT message id to use
* @param dup boolean - whether to set the MQTT DUP flag
* @param socket the open socket to send the data to
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_pubrel(int msgid, int dup, networkHandles* net, const char* clientID)
int MQTTPacket_send_pubrel(int MQTTVersion, int msgid, int dup, networkHandles* net, const char* clientID)
{
int rc = 0;
FUNC_ENTRY;
rc = MQTTPacket_send_ack(PUBREL, msgid, dup, net);
rc = MQTTPacket_send_ack(MQTTVersion, PUBREL, msgid, dup, net);
Log(LOG_PROTOCOL, 16, NULL, net->socket, clientID, msgid, rc);
FUNC_EXIT_RC(rc);
return rc;
@ -715,17 +719,18 @@ int MQTTPacket_send_pubrel(int msgid, int dup, networkHandles* net, const char*
/**
* Send an MQTT PUBCOMP packet down a socket.
* @param MQTTVersion the version of MQTT being used
* @param msgid the MQTT message id to use
* @param socket the open socket to send the data to
* @param clientID the string client identifier, only used for tracing
* @return the completion code (e.g. TCPSOCKET_COMPLETE)
*/
int MQTTPacket_send_pubcomp(int msgid, networkHandles* net, const char* clientID)
int MQTTPacket_send_pubcomp(int MQTTVersion, int msgid, networkHandles* net, const char* clientID)
{
int rc = 0;
FUNC_ENTRY;
rc = MQTTPacket_send_ack(PUBCOMP, msgid, 0, net);
rc = MQTTPacket_send_ack(MQTTVersion, PUBCOMP, msgid, 0, net);
Log(LOG_PROTOCOL, 18, NULL, net->socket, clientID, msgid, rc);
FUNC_EXIT_RC(rc);
return rc;
@ -734,6 +739,7 @@ int MQTTPacket_send_pubcomp(int msgid, networkHandles* net, const char* clientID
/**
* Function used in the new packets table to create acknowledgement packets.
* @param MQTTVersion the version of MQTT being used
* @param aHeader the MQTT header byte
* @param data the rest of the packet
* @param datalen the length of the rest of the packet

8
src/MQTTPacket.h
View File

@ -246,15 +246,15 @@ int MQTTPacket_send_disconnect(Clients* client, enum MQTTReasonCodes reason, MQT
void* MQTTPacket_publish(int MQTTVersion, unsigned char aHeader, char* data, size_t datalen);
void MQTTPacket_freePublish(Publish* pack);
int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, networkHandles* net, const char* clientID);
int MQTTPacket_send_puback(int msgid, networkHandles* net, const char* clientID);
int MQTTPacket_send_puback(int MQTTVersion, int msgid, networkHandles* net, const char* clientID);
void* MQTTPacket_ack(int MQTTVersion, unsigned char aHeader, char* data, size_t datalen);
void MQTTPacket_freeAck(Ack* pack);
void MQTTPacket_freeSuback(Suback* pack);
void MQTTPacket_freeUnsuback(Unsuback* pack);
int MQTTPacket_send_pubrec(int msgid, networkHandles* net, const char* clientID);
int MQTTPacket_send_pubrel(int msgid, int dup, networkHandles* net, const char* clientID);
int MQTTPacket_send_pubcomp(int msgid, networkHandles* net, const char* clientID);
int MQTTPacket_send_pubrec(int MQTTVersion, int msgid, networkHandles* net, const char* clientID);
int MQTTPacket_send_pubrel(int MQTTVersion, int msgid, int dup, networkHandles* net, const char* clientID);
int MQTTPacket_send_pubcomp(int MQTTVersion, int msgid, networkHandles* net, const char* clientID);
void MQTTPacket_free_packet(MQTTPacket* pack);

14
src/MQTTProtocolClient.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2018 IBM Corp.
* Copyright (c) 2009, 2019 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
@ -288,7 +288,7 @@ int MQTTProtocol_handlePublishes(void* pack, int sock)
else if (publish->header.bits.qos == 1)
{
/* send puback before processing the publications because a lot of return publications could fill up the socket buffer */
rc = MQTTPacket_send_puback(publish->msgId, &client->net, client->clientID);
rc = MQTTPacket_send_puback(publish->MQTTVersion, publish->msgId, &client->net, client->clientID);
/* if we get a socket error from sending the puback, should we ignore the publication? */
Protocol_processPublication(publish, client);
}
@ -320,7 +320,7 @@ int MQTTProtocol_handlePublishes(void* pack, int sock)
already_received = 1;
} else
ListAppend(client->inboundMsgs, m, sizeof(Messages) + len);
rc = MQTTPacket_send_pubrec(publish->msgId, &client->net, client->clientID);
rc = MQTTPacket_send_pubrec(publish->MQTTVersion, publish->msgId, &client->net, client->clientID);
if (m->MQTTVersion >= MQTTVERSION_5 && already_received == 0)
{
Publish publish1;
@ -447,7 +447,7 @@ int MQTTProtocol_handlePubrecs(void* pack, int sock)
}
else
{
rc = MQTTPacket_send_pubrel(pubrec->msgId, 0, &client->net, client->clientID);
rc = MQTTPacket_send_pubrel(pubrec->MQTTVersion, pubrec->msgId, 0, &client->net, client->clientID);
m->nextMessageType = PUBCOMP;
time(&(m->lastTouch));
}
@ -486,7 +486,7 @@ int MQTTProtocol_handlePubrels(void* pack, int sock)
rc = SOCKET_ERROR; /* queue acks? */
else
/* Apparently this is "normal" behaviour, so we don't need to issue a warning */
rc = MQTTPacket_send_pubcomp(pubrel->msgId, &client->net, client->clientID);
rc = MQTTPacket_send_pubcomp(pubrel->MQTTVersion, pubrel->msgId, &client->net, client->clientID);
}
else
{
@ -503,7 +503,7 @@ int MQTTProtocol_handlePubrels(void* pack, int sock)
memset(&publish, '\0', sizeof(publish));
/* send pubcomp before processing the publications because a lot of return publications could fill up the socket buffer */
rc = MQTTPacket_send_pubcomp(pubrel->msgId, &client->net, client->clientID);
rc = MQTTPacket_send_pubcomp(pubrel->MQTTVersion, pubrel->msgId, &client->net, client->clientID);
publish.header.bits.qos = m->qos;
publish.header.bits.retain = m->retain;
publish.msgId = m->msgid;
@ -698,7 +698,7 @@ static void MQTTProtocol_retries(time_t now, Clients* client, int regardless)
else if (m->qos && m->nextMessageType == PUBCOMP)
{
Log(TRACE_MIN, 7, NULL, "PUBREL", client->clientID, client->net.socket, m->msgid);
if (MQTTPacket_send_pubrel(m->msgid, 0, &client->net, client->clientID) != TCPSOCKET_COMPLETE)
if (MQTTPacket_send_pubrel(m->MQTTVersion, m->msgid, 0, &client->net, client->clientID) != TCPSOCKET_COMPLETE)
{
client->good = 0;
Log(TRACE_PROTOCOL, 29, NULL, client->clientID, client->net.socket,

4
test/test2.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2018 IBM Corp.
* Copyright (c) 2009, 2019 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
@ -675,7 +675,7 @@ int main(int argc, char** argv)
fprintf(xml, "<testsuite name=\"test1\" tests=\"%d\">\n", (int)(ARRAY_SIZE(tests) - 1));
setenv("MQTT_C_CLIENT_TRACE", "ON", 1);
setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 0);
setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 1);
getopts(argc, argv);

4
test/test3.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2012, 2018 IBM Corp.
* Copyright (c) 2012, 2019 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v1.0
@ -1616,7 +1616,7 @@ int main(int argc, char** argv)
fprintf(xml, "<testsuite name=\"test3\" tests=\"%d\">\n", (int)(ARRAY_SIZE(tests) - 1));
setenv("MQTT_C_CLIENT_TRACE", "ON", 1);
setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 0);
setenv("MQTT_C_CLIENT_TRACE_LEVEL", "ERROR", 1);
getopts(argc, argv);
if (options.test_no == 0)
{ /* run all the tests */

5
test/test45.c
View File

@ -1492,7 +1492,7 @@ int test7_run(int qos, int start_mqtt_version, int restore_mqtt_version)
test_finished = 0;
createOpts.MQTTVersion = start_mqtt_version;
MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_PROTOCOL);
MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
rc = MQTTAsync_createWithOptions(&c, options.connection, "async_test7",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
assert("good rc from create", rc == MQTTASYNC_SUCCESS, "rc was %d\n", rc);
@ -1623,7 +1623,7 @@ int test7_run(int qos, int start_mqtt_version, int restore_mqtt_version)
MQTTAsync_destroy(&c); /* force re-reading persistence on create */
MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_PROTOCOL);
MQTTAsync_setTraceLevel(MQTTASYNC_TRACE_ERROR);
createOpts.MQTTVersion = restore_mqtt_version;
rc = MQTTAsync_createWithOptions(&c, options.connection, "async_test7",
MQTTCLIENT_PERSISTENCE_DEFAULT, NULL, &createOpts);
@ -1732,6 +1732,7 @@ int test7(struct Options options)
fprintf(xml, "<testcase classname=\"test7\" name=\"persistence\"");
global_start_time = start_clock();
rc = test7_run(1, MQTTVERSION_5, MQTTVERSION_5) +
test7_run(2, MQTTVERSION_3_1_1, MQTTVERSION_3_1_1) +
test7_run(2, MQTTVERSION_5, MQTTVERSION_5) +
test7_run(2, MQTTVERSION_3_1_1, MQTTVERSION_5) +
test7_run(2, MQTTVERSION_5, MQTTVERSION_3_1_1);