From 46daa7d47dbd14b787471b8c0a3ec7c62a8eba45 Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Fri, 4 Feb 2022 19:08:15 +0000 Subject: [PATCH 1/2] Fix clash of commits that resulted in some QoS2 packets being sent twice --- src/MQTTProtocolClient.c | 24 +++++++++--------------- 1 file changed, 9 insertions(+), 15 deletions(-) diff --git a/src/MQTTProtocolClient.c b/src/MQTTProtocolClient.c index b467d391..9800fb76 100644 --- a/src/MQTTProtocolClient.c +++ b/src/MQTTProtocolClient.c @@ -338,13 +338,11 @@ int MQTTProtocol_handlePublishes(void* pack, int sock) if (publish->header.bits.qos == 1) { - /* if we get a socket error from sending the puback, should we ignore the publication? */ Protocol_processPublication(publish, client, 1); if (socketHasPendingWrites) rc = MQTTProtocol_queueAck(client, PUBACK, publish->msgId); else - /* send puback before processing the publications because a lot of return publications could fill up the socket buffer */ rc = MQTTPacket_send_puback(publish->MQTTVersion, publish->msgId, &client->net, client->clientID); } else if (publish->header.bits.qos == 2) @@ -382,11 +380,6 @@ int MQTTProtocol_handlePublishes(void* pack, int sock) } else ListAppend(client->inboundMsgs, m, sizeof(Messages) + len); - if (socketHasPendingWrites) - rc = MQTTProtocol_queueAck(client, PUBREC, publish->msgId); - else - rc = MQTTPacket_send_pubrec(publish->MQTTVersion, publish->msgId, &client->net, client->clientID); - if (m->MQTTVersion >= MQTTVERSION_5 && already_received == 0) { Publish publish1; @@ -416,7 +409,10 @@ int MQTTProtocol_handlePublishes(void* pack, int sock) } memcpy(m->publish->payload, temp, m->publish->payloadlen); } - rc = MQTTPacket_send_pubrec(publish->MQTTVersion, publish->msgId, &client->net, client->clientID); + if (socketHasPendingWrites) + rc = MQTTProtocol_queueAck(client, PUBREC, publish->msgId); + else + rc = MQTTPacket_send_pubrec(publish->MQTTVersion, publish->msgId, &client->net, client->clientID); publish->topic = NULL; } exit: @@ -580,12 +576,6 @@ 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 */ - if (!Socket_noPendingWrites(sock)) - rc = MQTTProtocol_queueAck(client, PUBCOMP, pubrel->msgId); - else - 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; @@ -612,7 +602,11 @@ int MQTTProtocol_handlePubrels(void* pack, int sock) ListRemove(&(state.publications), m->publish); ListRemove(client->inboundMsgs, m); ++(state.msgs_received); - rc = MQTTPacket_send_pubcomp(pubrel->MQTTVersion, pubrel->msgId, &client->net, client->clientID); + + if (!Socket_noPendingWrites(sock)) + rc = MQTTProtocol_queueAck(client, PUBCOMP, pubrel->msgId); + else + rc = MQTTPacket_send_pubcomp(pubrel->MQTTVersion, pubrel->msgId, &client->net, client->clientID); } } if (pubrel->MQTTVersion >= MQTTVERSION_5) From 9500f4b7e2e5578a2661ccd3c4dca633289b5faf Mon Sep 17 00:00:00 2001 From: Ian Craggs Date: Mon, 7 Feb 2022 10:29:09 +0000 Subject: [PATCH 2/2] Fix MacOS compile warnings --- src/Log.c | 6 +++--- src/Log.h | 12 ++++++++++-- src/StackTrace.c | 8 ++++---- 3 files changed, 17 insertions(+), 9 deletions(-) diff --git a/src/Log.c b/src/Log.c index 6ef09248..92054db0 100644 --- a/src/Log.c +++ b/src/Log.c @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2020 IBM Corp. + * Copyright (c) 2009, 2022 IBM Corp. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 @@ -86,7 +86,7 @@ typedef struct #endif int sametime_count; int number; - int thread_id; + thread_id_type thread_id; int depth; char name[MAX_FUNCTION_NAME_LENGTH + 1]; int line; @@ -453,7 +453,7 @@ void Log(enum LOG_LEVELS log_level, int msgno, const char *format, ...) * @param aFormat the printf format string to be used if the message id does not exist * @param ... the printf inserts */ -void Log_stackTrace(enum LOG_LEVELS log_level, int msgno, int thread_id, int current_depth, const char* name, int line, int* rc) +void Log_stackTrace(enum LOG_LEVELS log_level, int msgno, thread_id_type thread_id, int current_depth, const char* name, int line, int* rc) { traceEntry *cur_entry = NULL; diff --git a/src/Log.h b/src/Log.h index 102e31df..ae9bb9ef 100644 --- a/src/Log.h +++ b/src/Log.h @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2013 IBM Corp. + * Copyright (c) 2009, 2022 IBM Corp. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 @@ -18,6 +18,14 @@ #if !defined(LOG_H) #define LOG_H +#if defined(_WIN32) || defined(_WIN64) + #include + #define thread_id_type DWORD +#else + #include + #define thread_id_type pthread_t +#endif + /*BE map LOG_LEVELS { @@ -76,7 +84,7 @@ int Log_initialize(Log_nameValue*); void Log_terminate(void); void Log(enum LOG_LEVELS, int, const char *, ...); -void Log_stackTrace(enum LOG_LEVELS, int, int, int, const char*, int, int*); +void Log_stackTrace(enum LOG_LEVELS, int, thread_id_type, int, const char*, int, int*); typedef void Log_traceCallback(enum LOG_LEVELS level, const char *message); void Log_setTraceCallback(Log_traceCallback* callback); diff --git a/src/StackTrace.c b/src/StackTrace.c index d618b1fe..0fd61e31 100644 --- a/src/StackTrace.c +++ b/src/StackTrace.c @@ -1,5 +1,5 @@ /******************************************************************************* - * Copyright (c) 2009, 2020 IBM Corp. + * Copyright (c) 2009, 2022 IBM Corp. * * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v2.0 @@ -109,7 +109,7 @@ void StackTrace_entry(const char* name, int line, enum LOG_LEVELS trace_level) if (!setStack(1)) goto exit; if (trace_level != -1) - Log_stackTrace(trace_level, 9, (int)my_thread->id, my_thread->current_depth, name, line, NULL); + Log_stackTrace(trace_level, 9, my_thread->id, my_thread->current_depth, name, line, NULL); strncpy(my_thread->callstack[my_thread->current_depth].name, name, sizeof(my_thread->callstack[0].name)-1); my_thread->callstack[(my_thread->current_depth)++].line = line; if (my_thread->current_depth > my_thread->maxdepth) @@ -133,9 +133,9 @@ void StackTrace_exit(const char* name, int line, void* rc, enum LOG_LEVELS trace if (trace_level != -1) { if (rc == NULL) - Log_stackTrace(trace_level, 10, (int)my_thread->id, my_thread->current_depth, name, line, NULL); + Log_stackTrace(trace_level, 10, my_thread->id, my_thread->current_depth, name, line, NULL); else - Log_stackTrace(trace_level, 11, (int)my_thread->id, my_thread->current_depth, name, line, (int*)rc); + Log_stackTrace(trace_level, 11, my_thread->id, my_thread->current_depth, name, line, (int*)rc); } exit: Thread_unlock_mutex(stack_mutex);