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

Merge branch 'develop'

This commit is contained in:
Ian Craggs 2026-02-20 15:08:43 +00:00
parent eae145401f 3cbacd05ec
commit a3dd61ffbb
No known key found for this signature in database
GPG Key ID: A7AE1A8F2CCAB186
32 changed files with 1145 additions and 962 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

79
.clang-format Normal file
View File

@ -0,0 +1,79 @@
---
Language: C
# Indentation: tabs, one tab per indent level
# The Emacs footer in source files specifies indent-tabs-mode: t
UseTab: ForIndentation
TabWidth: 4
IndentWidth: 4
ContinuationIndentWidth: 8
IndentCaseLabels: true
IndentCaseBlocks: false
IndentGotoLabels: false
IndentPPDirectives: None
IndentWrappedFunctionNames: false
# Braces: Allman style - opening brace on its own line for everything
BreakBeforeBraces: Allman
# Alignment
AlignAfterOpenBracket: DontAlign
AlignConsecutiveAssignments: false
AlignConsecutiveDeclarations: false
AlignConsecutiveMacros: false
AlignEscapedNewlines: Left
AlignOperands: true
AlignTrailingComments: true
# Pointer style: mixed in codebase (char* ptr and char *ptr both appear),
# derive per file with Left as default
PointerAlignment: Left
DerivePointerAlignment: true
# Spaces
SpaceAfterCStyleCast: false
SpaceAfterLogicalNot: false
SpaceBeforeAssignmentOperators: true
SpaceBeforeParens: ControlStatements
SpaceInEmptyBlock: false
SpaceInEmptyParentheses: false
SpacesInCStyleCastParentheses: false
SpacesInParentheses: false
SpacesInSquareBrackets: false
SpacesBeforeTrailingComments: 1
# Line breaking
ColumnLimit: 120
AllowAllArgumentsOnNextLine: true
AllowAllParametersOfDeclarationOnNextLine: true
AllowShortBlocksOnASingleLine: Never
AllowShortCaseLabelsOnASingleLine: false
AllowShortEnumsOnASingleLine: false
AllowShortFunctionsOnASingleLine: None
AllowShortIfStatementsOnASingleLine: Never
AllowShortLoopsOnASingleLine: false
AlwaysBreakAfterReturnType: None
AlwaysBreakBeforeMultilineStrings: false
BinPackArguments: true
BinPackParameters: true
BreakBeforeBinaryOperators: None
BreakBeforeTernaryOperators: true
BreakStringLiterals: true
# Includes: preserve existing order, do not sort
SortIncludes: Never
IncludeBlocks: Preserve
# Blank lines
MaxEmptyLinesToKeep: 2
# Comments: do not reflow to preserve manual formatting
ReflowComments: false
# Penalties for line breaking decisions
PenaltyBreakBeforeFirstCallParameter: 19
PenaltyBreakComment: 300
PenaltyBreakFirstLessLess: 120
PenaltyBreakString: 1000
PenaltyExcessCharacter: 1000000
PenaltyReturnTypeOnItsOwnLine: 60

4
.github/workflows/artifact_linux.yml vendored
View File

@ -26,5 +26,5 @@ jobs:
- name: Upload artifact
uses: actions/upload-artifact@v4
with:
name: Eclipse-Paho-MQTT-C-1.3.15-Linux.tar.gz
path: /home/runner/work/paho.mqtt.c/paho.mqtt.c/build.paho/Eclipse-Paho-MQTT-C-1.3.15-Linux.tar.gz
name: Eclipse-Paho-MQTT-C-1.3.16-Linux.tar.gz
path: /home/runner/work/paho.mqtt.c/paho.mqtt.c/build.paho/Eclipse-Paho-MQTT-C-1.3.16-Linux.tar.gz

4
.github/workflows/artifact_macos.yml vendored
View File

@ -26,5 +26,5 @@ jobs:
- name: Upload artifact
uses: actions/upload-artifact@v4
with:
name: Eclipse-Paho-MQTT-C-1.3.15-Darwin.tar.gz
path: /Users/runner/work/paho.mqtt.c/paho.mqtt.c/build.paho/Eclipse-Paho-MQTT-C-1.3.15-Darwin.tar.gz
name: Eclipse-Paho-MQTT-C-1.3.16-Darwin.tar.gz
path: /Users/runner/work/paho.mqtt.c/paho.mqtt.c/build.paho/Eclipse-Paho-MQTT-C-1.3.16-Darwin.tar.gz

9
CMakeLists.txt
View File

@ -1,5 +1,5 @@
#*******************************************************************************
# Copyright (c) 2015, 2025 logi.cals GmbH, Frank Pagliughi <fpagliughi@mindspring.com> and others
# Copyright (c) 2015, 2026 logi.cals GmbH, Frank Pagliughi <fpagliughi@mindspring.com> and others
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Eclipse Public License v2.0
@ -18,7 +18,7 @@
cmake_minimum_required(VERSION 3.12)
project("Eclipse Paho C"
VERSION 1.3.15
VERSION 1.3.16
LANGUAGES C
)
@ -64,6 +64,7 @@ option(PAHO_ENABLE_TESTING "Build tests and run" TRUE)
option(PAHO_ENABLE_CPACK "Enable CPack" TRUE)
option(PAHO_HIGH_PERFORMANCE "Disable tracing and heap tracking" FALSE)
option(PAHO_USE_SELECT "Revert to select system call instead of poll" FALSE)
option(PAHO_NO_TCP_NODELAY "Don't disable Nagle's algorithm on TCP sockets" FALSE)
if(NOT WIN32)
option(PAHO_WITH_UNIX_SOCKETS "Flag that defines whether to enable Unix-domain sockets" FALSE)
@ -85,6 +86,10 @@ if(PAHO_WITH_LIBUUID)
add_definitions(-DUSE_LIBUUID=1)
endif()
if(PAHO_NO_TCP_NODELAY)
add_definitions(-DNO_TCP_NODELAY=1)
endif()
if(NOT PAHO_BUILD_SHARED AND NOT PAHO_BUILD_STATIC)
message(FATAL_ERROR "You must set either PAHO_BUILD_SHARED, PAHO_BUILD_STATIC, or both")
endif()

2
appveyor.yml
View File

@ -22,7 +22,7 @@ environment:
PAHO_BUILD_STATIC: FALSE
PAHO_BUILD_SHARED: TRUE
PAHO_HIGH_PERFORMANCE: TRUE
PYTHON_VERSION: Python312
PYTHON_VERSION: Python313
configuration: Debug
install:

2
doc/man/man1/paho_c_pub.1
View File

@ -110,7 +110,7 @@ Do not print error messages.
.TP
.PD
.B \-\-trace
Print library internal trace. Valid levels are \fBmin\fR, \fBmax\fR, \fBerror\fR and \fprotocol\fR.
Print library internal trace. Valid levels are \fBmin\fR, \fBmax\fR, \fBerror\fR and \fBprotocol\fR.
.TP
.PD 0
.BI \-r

2
doc/man/man1/paho_c_sub.1
View File

@ -102,7 +102,7 @@ Do not print error messages.
.TP
.PD
.B \-\-trace
Print library internal trace. Valid levels are \fBmin\fR, \fBmax\fR, \fBerror\fR and \fprotocol\fR.
Print library internal trace. Valid levels are \fBmin\fR, \fBmax\fR, \fBerror\fR and \fBprotocol\fR.
.TP
.PD 0
.BI \-R

2
doc/man/man1/paho_cs_pub.1
View File

@ -110,7 +110,7 @@ Do not print error messages.
.TP
.PD
.B \-\-trace
Print library internal trace. Valid levels are \fBmin\fR, \fBmax\fR, \fBerror\fR and \fprotocol\fR.
Print library internal trace. Valid levels are \fBmin\fR, \fBmax\fR, \fBerror\fR and \fBprotocol\fR.
.TP
.PD 0
.BI \-r

2
doc/man/man1/paho_cs_sub.1
View File

@ -102,7 +102,7 @@ Do not print error messages.
.TP
.PD
.B \-\-trace
Print library internal trace. Valid levels are \fBmin\fR, \fBmax\fR, \fBerror\fR and \fprotocol\fR.
Print library internal trace. Valid levels are \fBmin\fR, \fBmax\fR, \fBerror\fR and \fBprotocol\fR.
.TP
.PD 0
.BI \-R

7
fmt.sh Executable file
View File

@ -0,0 +1,7 @@
#!/bin/bash
#
# Runs clang format over the whole project tree, excluding the 'build/' directory.
#
find . -type d \( -path './build' \) -prune -iname '*.h' -o -iname '*.c' | xargs clang-format -i

59
src/Base64.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2018, 2019 Wind River Systems, Inc. All Rights Reserved.
* Copyright (c) 2018, 2026 Wind River Systems, Inc., and others
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -48,40 +48,45 @@ static b64_size_t Base64_encodeDecode(
b64_size_t ret = 0u;
if ( in_len > 0u )
{
int rv;
BIO *bio, *b64, *b_in, *b_out;
b64 = BIO_new(BIO_f_base64());
bio = BIO_new(BIO_s_mem());
b64 = BIO_push(b64, bio);
BIO_set_flags(b64, BIO_FLAGS_BASE64_NO_NL); /* ignore new-lines */
if ( encode )
{
b_in = bio;
b_out = b64;
/* For encoding: output length is 4 * (input_length / 3) rounded up to nearest multiple of 4 */
b64_size_t required_len = ((in_len + 2) / 3) * 4;
if (out_len >= required_len + 1) /* +1 for null terminator */
{
int encoded_len = EVP_EncodeBlock((unsigned char*)out, (const unsigned char*)in, (int)in_len);
if (encoded_len > 0)
{
ret = (b64_size_t)encoded_len;
if (out_len > ret)
out[ret] = '\0';
}
}
}
else
{
b_in = b64;
b_out = bio;
}
rv = BIO_write(b_out, in, (int)in_len);
BIO_flush(b_out); /* indicate end of encoding */
if ( rv > 0 )
{
rv = BIO_read(b_in, out, (int)out_len);
if ( rv > 0 )
/* For decoding: output length is at most 3 * (input_length / 4) */
b64_size_t max_out_len = (in_len / 4) * 3;
if (out_len >= max_out_len + 1) /* +1 for null terminator */
{
ret = (b64_size_t)rv;
if ( out_len > ret )
out[ret] = '\0';
EVP_ENCODE_CTX *ctx = EVP_ENCODE_CTX_new();
if (ctx)
{
int decoded_len = 0;
int final_len = 0;
unsigned char *temp_out = (unsigned char*)out;
EVP_DecodeInit(ctx);
EVP_DecodeUpdate(ctx, temp_out, &decoded_len, (const unsigned char*)in, (int)in_len);
EVP_DecodeFinal(ctx, temp_out + decoded_len, &final_len);
EVP_ENCODE_CTX_free(ctx);
ret = (b64_size_t)(decoded_len + final_len);
if (out_len > ret)
out[ret] = '\0';
}
}
}
BIO_free_all(b64); /* free all used memory */
}
return ret;
}

4
src/CMakeLists.txt
View File

@ -1,5 +1,5 @@
#*******************************************************************************
# Copyright (c) 2015, 2024 logi.cals GmbH, Frank Pagliughi <fpagliughi@mindspring.com> and others
# Copyright (c) 2015, 2026 logi.cals GmbH, Frank Pagliughi <fpagliughi@mindspring.com> and others
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Eclipse Public License v2.0
@ -79,7 +79,7 @@ elseif(UNIX)
elseif(CMAKE_SYSTEM_NAME MATCHES "FreeBSD")
set(LIBS_SYSTEM compat pthread)
elseif(CMAKE_SYSTEM_NAME MATCHES "QNX")
set(LIBS_SYSTEM c)
set(LIBS_SYSTEM c socket)
else()
set(LIBS_SYSTEM c pthread)
endif()

104
src/MQTTAsync.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2025 IBM Corp., Ian Craggs and others
* Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -129,7 +129,7 @@ void MQTTAsync_init_rand(void)
mutex_type mqttasync_mutex = NULL;
mutex_type socket_mutex = NULL;
mutex_type mqttcommand_mutex = NULL;
sem_type send_sem = NULL;
evt_type send_evt = NULL;
#if !defined(NO_HEAP_TRACKING)
extern mutex_type stack_mutex;
extern mutex_type heap_mutex;
@ -142,52 +142,47 @@ int MQTTAsync_init(void)
if (mqttasync_mutex == NULL)
{
if ((mqttasync_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
mqttasync_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("mqttasync_mutex error %d\n", rc);
goto exit;
}
if ((mqttcommand_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
mqttcommand_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("mqttcommand_mutex error %d\n", rc);
goto exit;
}
if ((send_sem = CreateEvent(
NULL, /* default security attributes */
FALSE, /* manual-reset event? */
FALSE, /* initial state is nonsignaled */
NULL /* object name */
)) == NULL)
send_evt = Thread_create_evt(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("send_sem error %d\n", rc);
printf("send_evt error %d\n", rc);
goto exit;
}
#if !defined(NO_HEAP_TRACKING)
if ((stack_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
stack_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("stack_mutex error %d\n", rc);
goto exit;
}
if ((heap_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
heap_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("heap_mutex error %d\n", rc);
goto exit;
}
#endif
if ((log_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
log_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("log_mutex error %d\n", rc);
goto exit;
}
if ((socket_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
socket_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("socket_mutex error %d\n", rc);
goto exit;
}
@ -202,20 +197,20 @@ exit:
void MQTTAsync_cleanup(void)
{
if (send_sem)
CloseHandle(send_sem);
if (send_evt)
Thread_destroy_evt(send_evt);
#if !defined(NO_HEAP_TRACKING)
if (stack_mutex)
CloseHandle(stack_mutex);
Paho_thread_destroy_mutex(stack_mutex);
if (heap_mutex)
CloseHandle(heap_mutex);
Paho_thread_destroy_mutex(heap_mutex);
#endif
if (log_mutex)
CloseHandle(log_mutex);
Paho_thread_destroy_mutex(log_mutex);
if (socket_mutex)
CloseHandle(socket_mutex);
Paho_thread_destroy_mutex(socket_mutex);
if (mqttasync_mutex)
CloseHandle(mqttasync_mutex);
Paho_thread_destroy_mutex(mqttasync_mutex);
}
#if defined(PAHO_MQTT_STATIC)
@ -264,8 +259,8 @@ mutex_type socket_mutex = &socket_mutex_store;
static pthread_mutex_t mqttcommand_mutex_store = PTHREAD_MUTEX_INITIALIZER;
mutex_type mqttcommand_mutex = &mqttcommand_mutex_store;
static cond_type_struct send_cond_store = { PTHREAD_COND_INITIALIZER, PTHREAD_MUTEX_INITIALIZER };
cond_type send_cond = &send_cond_store;
static evt_type_struct send_evt_store = { PTHREAD_COND_INITIALIZER, PTHREAD_MUTEX_INITIALIZER, 0 };
evt_type send_evt = &send_evt_store;
int MQTTAsync_init(void)
{
@ -284,10 +279,10 @@ int MQTTAsync_init(void)
printf("MQTTAsync: error %d initializing command_mutex\n", rc);
else if ((rc = pthread_mutex_init(socket_mutex, &attr)) != 0)
printf("MQTTClient: error %d initializing socket_mutex\n", rc);
else if ((rc = pthread_cond_init(&send_cond->cond, NULL)) != 0)
printf("MQTTAsync: error %d initializing send_cond cond\n", rc);
else if ((rc = pthread_mutex_init(&send_cond->mutex, &attr)) != 0)
printf("MQTTAsync: error %d initializing send_cond mutex\n", rc);
else if ((rc = pthread_cond_init(&send_evt->cond, NULL)) != 0)
printf("MQTTAsync: error %d initializing send_evt cond\n", rc);
else if ((rc = pthread_mutex_init(&send_evt->mutex, &attr)) != 0)
printf("MQTTAsync: error %d initializing send_evt mutex\n", rc);
return rc;
}
@ -1022,7 +1017,7 @@ int MQTTAsync_inCallback()
int MQTTAsync_subscribeMany(MQTTAsync handle, int count, char* const* topic, const int* qos, MQTTAsync_responseOptions* response)
{
MQTTAsyncs* m = handle;
int i = 0;
int i = 0, j = 0;
int rc = MQTTASYNC_SUCCESS;
MQTTAsync_queuedCommand* sub;
int msgid = 0;
@ -1096,6 +1091,10 @@ int MQTTAsync_subscribeMany(MQTTAsync handle, int count, char* const* topic, con
if ((sub->command.details.sub.optlist = malloc(sizeof(MQTTSubscribe_options) * count)) == NULL)
{
rc = PAHO_MEMORY_ERROR;
if(sub)
{
free(sub);
}
goto exit;
}
if (response->subscribeOptionsCount == 0)
@ -1123,6 +1122,21 @@ int MQTTAsync_subscribeMany(MQTTAsync handle, int count, char* const* topic, con
if ((sub->command.details.sub.topics[i] = MQTTStrdup(topic[i])) == NULL)
{
rc = PAHO_MEMORY_ERROR;
for(j = 0; j < i; ++j)
{
if(sub->command.details.sub.topics[j])
{
free(sub->command.details.sub.topics[j]);
}
}
if(sub->command.details.sub.optlist)
{
free(sub->command.details.sub.optlist);
}
free(sub->command.details.sub.topics);
free(sub->command.details.sub.qoss);
free(sub);
goto exit;
}
sub->command.details.sub.qoss[i] = qos[i];
@ -1130,7 +1144,25 @@ int MQTTAsync_subscribeMany(MQTTAsync handle, int count, char* const* topic, con
rc = MQTTAsync_addCommand(sub, sizeof(sub));
}
else
{
rc = PAHO_MEMORY_ERROR;
if(sub->command.details.sub.optlist)
{
free(sub->command.details.sub.optlist);
}
if(sub->command.details.sub.topics)
{
free(sub->command.details.sub.topics);
}
if(sub->command.details.sub.qoss)
{
free(sub->command.details.sub.qoss);
}
if(sub)
{
free(sub);
}
}
exit:
if (!MQTTAsync_inCallback())

14
src/MQTTAsync.h
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2025 IBM Corp., Ian Craggs and others
* Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -29,11 +29,11 @@
* @cond MQTTAsync_main
* @mainpage Asynchronous MQTT client library for C (MQTTAsync)
*
* &copy; Copyright 2009, 2025 IBM Corp., Ian Craggs and others.
* &copy; Copyright 2009, 2026 IBM Corp., Ian Craggs and others.
*
* @brief An Asynchronous MQTT client library for C.
*
* Version 1.3.15
* Version 1.3.16
*
* An MQTT client application connects to MQTT-capable servers.
* A typical client is responsible for collecting information from a telemetry
@ -918,13 +918,13 @@ LIBMQTT_API int MQTTAsync_reconnect(MQTTAsync handle);
* <br>
* @em tcp:// or @em mqtt:// - Insecure TCP
* <br>
* @em ssl:// or @em mqtts:// - Encrypted SSL/TLS
* @em ssl:// or @em tls:// or @em mqtts:// - Encrypted SSL/TLS
* <br>
* @em ws:// - Insecure websockets
* <br>
* @em wss:// - Secure web sockets
* <br>
* The TLS enabled prefixes (ssl, mqtts, wss) are only valid if a TLS
* The TLS enabled prefixes (ssl, tls, mqtts, wss) are only valid if a TLS
* version of the library is linked with.
* For <i>host</i>, you can specify either an IP address or a host name. For
* instance, to connect to a server running on the local machines with the
@ -1311,8 +1311,8 @@ typedef struct
/**
* An array of null-terminated strings specifying the servers to
* which the client will connect. Each string takes the form <i>protocol://host:port</i>.
* <i>protocol</i> must be <i>tcp</i>, <i>ssl</i>, <i>ws</i> or <i>wss</i>.
* The TLS enabled prefixes (ssl, wss) are only valid if a TLS version of the library
* <i>protocol</i> must be <i>tcp</i>, <i>ssl</i>, <i>tls</i>, <i>ws</i> or <i>wss</i>.
* The TLS enabled prefixes (ssl, tls, wss) are only valid if a TLS version of the library
* is linked with.
* For <i>host</i>, you can
* specify either an IP address or a domain name. For instance, to connect to

80
src/MQTTAsyncUtils.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2025 IBM Corp., Ian Craggs and others
* Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -88,7 +88,6 @@ extern int MQTTAsync_tostop;
extern mutex_type mqttasync_mutex;
extern mutex_type socket_mutex;
extern mutex_type mqttcommand_mutex;
extern sem_type send_sem;
#if !defined(NO_HEAP_TRACKING)
extern mutex_type stack_mutex;
extern mutex_type heap_mutex;
@ -98,9 +97,10 @@ extern mutex_type log_mutex;
extern mutex_type mqttasync_mutex;
extern mutex_type socket_mutex;
extern mutex_type mqttcommand_mutex;
extern cond_type send_cond;
#endif
extern evt_type send_evt;
#if !defined(min)
#define min(a, b) (((a) < (b)) ? (a) : (b))
#endif
@ -936,13 +936,8 @@ int MQTTAsync_addCommand(MQTTAsync_queuedCommand* command, int command_size)
}
exit:
MQTTAsync_unlock_mutex(mqttcommand_mutex);
#if !defined(_WIN32)
if ((rc1 = Thread_signal_cond(send_cond)) != 0)
Log(LOG_ERROR, 0, "Error %d from signal cond", rc1);
#else
if ((rc1 = Thread_post_sem(send_sem)) != 0)
Log(LOG_ERROR, 0, "Error %d from signal cond", rc1);
#endif
if ((rc1 = Thread_signal_evt(send_evt)) != 0)
Log(LOG_ERROR, 0, "Error %d from signal event", rc1);
FUNC_EXIT_RC(rc);
return rc;
}
@ -1032,7 +1027,7 @@ static void MQTTProtocol_checkPendingWrites(void)
ListElement* le = state.pending_writes.first;
while (le)
{
if (Socket_noPendingWrites(((pending_write*)(le->content))->socket))
if (MQTTAsync_Socket_noPendingWrites(((pending_write*)(le->content))->socket))
{
MQTTProtocol_removePublication(((pending_write*)(le->content))->p);
state.pending_writes.current = le;
@ -1860,13 +1855,8 @@ thread_return_type WINAPI MQTTAsync_sendThread(void* n)
command_count = MQTTAsync_commands->count;
MQTTAsync_unlock_mutex(mqttcommand_mutex);
}
#if !defined(_WIN32)
if ((rc = Thread_wait_cond(send_cond, timeout)) != 0 && rc != ETIMEDOUT)
Log(LOG_ERROR, -1, "Error %d waiting for condition variable", rc);
#else
if ((rc = Thread_wait_sem(send_sem, timeout)) != 0 && rc != ETIMEDOUT)
Log(LOG_ERROR, -1, "Error %d waiting for semaphore", rc);
#endif
if ((rc = Thread_wait_evt(send_evt, timeout)) != 0 && rc != ETIMEDOUT)
Log(LOG_ERROR, -1, "Error %d waiting for send event", rc);
timeout = 1000; /* 1 second for follow on waits */
MQTTAsync_checkTimeouts();
}
@ -1877,7 +1867,7 @@ thread_return_type WINAPI MQTTAsync_sendThread(void* n)
MQTTAsync_unlock_mutex(mqttasync_mutex);
#if defined(OPENSSL)
#if ((OPENSSL_VERSION_NUMBER < 0x1010000fL) || defined(LIBRESSL_VERSION_NUMBER))
#if ((OPENSSL_VERSION_NUMBER < 0x1010000fL) || defined(LIBRESSL_VERSION_NUMBER)) || defined(OPENSSL_IS_BORINGSSL)
ERR_remove_state(0);
#else
OPENSSL_thread_stop();
@ -2071,11 +2061,7 @@ static int MQTTAsync_completeConnection(MQTTAsyncs* m, Connack* connack)
}
}
m->pack = NULL;
#if !defined(_WIN32)
Thread_signal_cond(send_cond);
#else
Thread_post_sem(send_sem);
#endif
Thread_signal_evt(send_evt);
}
FUNC_EXIT_RC(rc);
return rc;
@ -2100,7 +2086,9 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
MQTTPacket* pack = NULL;
MQTTAsync_unlock_mutex(mqttasync_mutex);
Log(TRACE_MINIMUM, -1, "MQTTAsync_cycle call timeout %lu", timeout);
pack = MQTTAsync_cycle(&sock, timeout, &rc);
Log(TRACE_MINIMUM, -1, "MQTTAsync_cycle returns pack %p, sock %d, timeout %lu, rc %d", pack, sock, timeout, rc);
MQTTAsync_lock_mutex(mqttasync_mutex);
if (MQTTAsync_tostop)
break;
@ -2401,16 +2389,11 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
receiveThread_state = STOPPED;
receiveThread_id = 0;
MQTTAsync_unlock_mutex(mqttasync_mutex);
#if !defined(_WIN32)
if (sendThread_state != STOPPED)
Thread_signal_cond(send_cond);
#else
if (sendThread_state != STOPPED)
Thread_post_sem(send_sem);
#endif
Thread_signal_evt(send_evt);
#if defined(OPENSSL)
#if ((OPENSSL_VERSION_NUMBER < 0x1010000fL) || defined(LIBRESSL_VERSION_NUMBER))
#if ((OPENSSL_VERSION_NUMBER < 0x1010000fL) || defined(LIBRESSL_VERSION_NUMBER)) || defined(OPENSSL_IS_BORINGSSL)
ERR_remove_state(0);
#else
OPENSSL_thread_stop();
@ -2479,7 +2462,7 @@ static void MQTTAsync_closeOnly(Clients* client, enum MQTTReasonCodes reasonCode
if (client->net.socket > 0)
{
MQTTProtocol_checkPendingWrites();
if (client->connected && Socket_noPendingWrites(client->net.socket))
if (client->connected && MQTTAsync_Socket_noPendingWrites(client->net.socket))
MQTTPacket_send_disconnect(client, reasonCode, props);
MQTTAsync_lock_mutex(socket_mutex);
WebSocket_close(&client->net, WebSocket_CLOSE_NORMAL, NULL);
@ -2949,7 +2932,7 @@ static int MQTTAsync_connecting(MQTTAsyncs* m)
setSocketForSSLrc = SSLSocket_setSocketForSSL(&m->c->net, m->c->sslopts,
serverURI, hostname_len);
if (setSocketForSSLrc != MQTTASYNC_SUCCESS)
if (setSocketForSSLrc == 1)
{
if (m->c->session != NULL)
if ((rc = SSL_set_session(m->c->net.ssl, m->c->session)) != 1)
@ -3084,16 +3067,21 @@ static MQTTPacket* MQTTAsync_cycle(SOCKET* sock, unsigned long timeout, int* rc)
if ((*sock = SSLSocket_getPendingRead()) == -1)
{
#endif
int should_stop = 0;
/*int should_stop = 0;*/
int interrupted = 0;
/* 0 from getReadySocket indicates no work to do, rc -1 == error */
*sock = Socket_getReadySocket(0, (int)timeout, socket_mutex, &rc1);
*sock = Socket_getReadySocket(0, (int)timeout, socket_mutex, &rc1, &interrupted);
*rc = rc1;
MQTTAsync_lock_mutex(mqttasync_mutex);
/*MQTTAsync_lock_mutex(mqttasync_mutex);
should_stop = MQTTAsync_tostop;
MQTTAsync_unlock_mutex(mqttasync_mutex);
if (!should_stop && *sock == 0 && (timeout > 0L))
MQTTAsync_sleep(100L);
MQTTAsync_unlock_mutex(mqttasync_mutex);*/
/* Frank deleted this small delay on the suggestion of Claude.
I want to preserve it as a comment in case we need it in the future.
The preceeding 3 lines of code are only needed if the following line
is reinstated - Ian
if (!should_stop && *sock == 0 && (timeout > 0L) && (interrupted == 0))
MQTTAsync_sleep(50L);*/
#if defined(OPENSSL)
}
#endif
@ -3152,11 +3140,7 @@ static MQTTPacket* MQTTAsync_cycle(SOCKET* sock, unsigned long timeout, int* rc)
{
*rc = MQTTProtocol_handlePubcomps(pack, *sock, &pubToRemove);
if (sendThread_state != STOPPED)
#if !defined(_WIN32)
Thread_signal_cond(send_cond);
#else
Thread_post_sem(send_sem);
#endif
Thread_signal_evt(send_evt);
}
else if (msgtype == PUBREC)
*rc = MQTTProtocol_handlePubrecs(pack, *sock, &pubToRemove);
@ -3164,15 +3148,11 @@ static MQTTPacket* MQTTAsync_cycle(SOCKET* sock, unsigned long timeout, int* rc)
{
*rc = MQTTProtocol_handlePubacks(pack, *sock, &pubToRemove);
if (sendThread_state != STOPPED)
#if !defined(_WIN32)
Thread_signal_cond(send_cond);
#else
Thread_post_sem(send_sem);
#endif
Thread_signal_evt(send_evt);
}
if (!m)
Log(LOG_ERROR, -1, "PUBCOMP, PUBACK or PUBREC received for no client, msgid %d", msgid);
if (m && (msgtype != PUBREC || ackrc >= MQTTREASONCODE_UNSPECIFIED_ERROR))
if (m && *rc == 0 && (msgtype != PUBREC || ackrc >= MQTTREASONCODE_UNSPECIFIED_ERROR))
{
ListElement* current = NULL;

126
src/MQTTClient.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2024 IBM Corp., Ian Craggs and others
* Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -91,7 +91,7 @@ const char *client_timestamp_eye = "MQTTClientV3_Timestamp " BUILD_TIMESTAMP;
const char *client_version_eye = "MQTTClientV3_Version " CLIENT_VERSION;
struct conlost_sync_data {
sem_type sem;
evt_type evt;
void *m;
};
@ -128,51 +128,51 @@ extern mutex_type log_mutex;
int MQTTClient_init(void)
{
DWORD rc = 0;
int rc = 0;
if (mqttclient_mutex == NULL)
{
if ((mqttclient_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
mqttclient_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("mqttclient_mutex error %d\n", rc);
goto exit;
}
if ((subscribe_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
subscribe_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("subscribe_mutex error %d\n", rc);
goto exit;
}
if ((connect_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
connect_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("connect_mutex error %d\n", rc);
goto exit;
}
#if !defined(NO_HEAP_TRACKING)
if ((stack_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
stack_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("stack_mutex error %d\n", rc);
goto exit;
}
if ((heap_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
heap_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("heap_mutex error %d\n", rc);
goto exit;
}
#endif
if ((log_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
log_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("log_mutex error %d\n", rc);
goto exit;
}
if ((socket_mutex = CreateMutex(NULL, 0, NULL)) == NULL)
socket_mutex = Paho_thread_create_mutex(&rc);
if (rc != 0)
{
rc = GetLastError();
printf("socket_mutex error %d\n", rc);
goto exit;
}
@ -184,21 +184,21 @@ exit:
void MQTTClient_cleanup(void)
{
if (connect_mutex)
CloseHandle(connect_mutex);
Paho_thread_destroy_mutex(connect_mutex);
if (subscribe_mutex)
CloseHandle(subscribe_mutex);
Paho_thread_destroy_mutex(subscribe_mutex);
#if !defined(NO_HEAP_TRACKING)
if (stack_mutex)
CloseHandle(stack_mutex);
Paho_thread_destroy_mutex(stack_mutex);
if (heap_mutex)
CloseHandle(heap_mutex);
Paho_thread_destroy_mutex(heap_mutex);
#endif
if (log_mutex)
CloseHandle(log_mutex);
Paho_thread_destroy_mutex(log_mutex);
if (socket_mutex)
CloseHandle(socket_mutex);
Paho_thread_destroy_mutex(socket_mutex);
if (mqttclient_mutex)
CloseHandle(mqttclient_mutex);
Paho_thread_destroy_mutex(mqttclient_mutex);
}
#if defined(PAHO_MQTT_STATIC)
@ -318,11 +318,11 @@ typedef struct
void* auth_handle_context; /* the context to be associated with the authHandle callback*/
#endif
sem_type connect_sem;
int rc; /* getsockopt return code in connect */
sem_type connack_sem;
sem_type suback_sem;
sem_type unsuback_sem;
evt_type connect_evt;
evt_type connack_evt;
evt_type suback_evt;
evt_type unsuback_evt;
MQTTPacket* pack;
unsigned long commandTimeout;
@ -524,10 +524,11 @@ int MQTTClient_createWithOptions(MQTTClient* handle, const char* serverURI, cons
m->c->messageQueue = ListInitialize();
m->c->outboundQueue = ListInitialize();
m->c->clientID = MQTTStrdup(clientId);
m->connect_sem = Thread_create_sem(&rc);
m->connack_sem = Thread_create_sem(&rc);
m->suback_sem = Thread_create_sem(&rc);
m->unsuback_sem = Thread_create_sem(&rc);
// TODO (fmp): None of these events are being checked for failure
m->connect_evt = Thread_create_evt(&rc);
m->connack_evt = Thread_create_evt(&rc);
m->suback_evt = Thread_create_evt(&rc);
m->unsuback_evt = Thread_create_evt(&rc);
#if !defined(NO_PERSISTENCE)
rc = MQTTPersistence_create(&(m->c->persistence), persistence_type, persistence_context);
@ -625,10 +626,10 @@ void MQTTClient_destroy(MQTTClient* handle)
}
if (m->serverURI)
free(m->serverURI);
Thread_destroy_sem(m->connect_sem);
Thread_destroy_sem(m->connack_sem);
Thread_destroy_sem(m->suback_sem);
Thread_destroy_sem(m->unsuback_sem);
Thread_destroy_evt(m->connect_evt);
Thread_destroy_evt(m->connack_evt);
Thread_destroy_evt(m->suback_evt);
Thread_destroy_evt(m->unsuback_evt);
if (!ListRemove(handles, m))
Log(LOG_ERROR, -1, "free error");
*handle = NULL;
@ -734,7 +735,7 @@ static thread_return_type WINAPI connectionLost_call(void* context)
(*(m->cl))(m->context, NULL);
Thread_post_sem(data->sem);
Thread_signal_evt(data->evt);
return 0;
}
@ -889,15 +890,15 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
{
if (m->c->connect_state == SSL_IN_PROGRESS)
{
Log(TRACE_MIN, -1, "Posting connect semaphore for client %s", m->c->clientID);
Log(TRACE_MIN, -1, "Signaling connect event for client %s", m->c->clientID);
m->c->connect_state = NOT_IN_PROGRESS;
Thread_post_sem(m->connect_sem);
Thread_signal_evt(m->connect_evt);
}
if (m->c->connect_state == WAIT_FOR_CONNACK)
{
Log(TRACE_MIN, -1, "Posting connack semaphore for client %s", m->c->clientID);
Log(TRACE_MIN, -1, "Signaling connack event for client %s", m->c->clientID);
m->c->connect_state = NOT_IN_PROGRESS;
Thread_post_sem(m->connack_sem);
Thread_signal_evt(m->connack_evt);
}
}
}
@ -936,21 +937,21 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
{
if (pack->header.bits.type == CONNACK)
{
Log(TRACE_MIN, -1, "Posting connack semaphore for client %s", m->c->clientID);
Log(TRACE_MIN, -1, "Signaling connack event for client %s", m->c->clientID);
m->pack = pack;
Thread_post_sem(m->connack_sem);
Thread_signal_evt(m->connack_evt);
}
else if (pack->header.bits.type == SUBACK)
{
Log(TRACE_MIN, -1, "Posting suback semaphore for client %s", m->c->clientID);
Log(TRACE_MIN, -1, "Signaling suback event for client %s", m->c->clientID);
m->pack = pack;
Thread_post_sem(m->suback_sem);
Thread_signal_evt(m->suback_evt);
}
else if (pack->header.bits.type == UNSUBACK)
{
Log(TRACE_MIN, -1, "Posting unsuback semaphore for client %s", m->c->clientID);
Log(TRACE_MIN, -1, "Signaling unsuback event for client %s", m->c->clientID);
m->pack = pack;
Thread_post_sem(m->unsuback_sem);
Thread_signal_evt(m->unsuback_evt);
}
else if (m->c->MQTTVersion >= MQTTVERSION_5)
{
@ -1000,9 +1001,9 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
if ((m->rc = getsockopt(m->c->net.socket, SOL_SOCKET, SO_ERROR, (char*)&error, &len)) == 0)
m->rc = error;
Log(TRACE_MIN, -1, "Posting connect semaphore for client %s rc %d", m->c->clientID, m->rc);
Log(TRACE_MIN, -1, "Signaling connect event for client %s rc %d", m->c->clientID, m->rc);
m->c->connect_state = NOT_IN_PROGRESS;
Thread_post_sem(m->connect_sem);
Thread_signal_evt(m->connect_evt);
}
#if defined(OPENSSL)
else if (m->c->connect_state == SSL_IN_PROGRESS)
@ -1017,9 +1018,9 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
if (rc == 1 && (m->c->cleansession == 0 && m->c->cleanstart == 0) && m->c->session == NULL)
m->c->session = SSL_get1_session(m->c->net.ssl);
m->rc = rc;
Log(TRACE_MIN, -1, "Posting connect semaphore for SSL client %s rc %d", m->c->clientID, m->rc);
Log(TRACE_MIN, -1, "Signaling connect event for SSL client %s rc %d", m->c->clientID, m->rc);
m->c->connect_state = NOT_IN_PROGRESS;
Thread_post_sem(m->connect_sem);
Thread_signal_evt(m->connect_evt);
}
}
#endif
@ -1027,9 +1028,9 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
{
if (rc != TCPSOCKET_INTERRUPTED)
{
Log(TRACE_MIN, -1, "Posting websocket handshake for client %s rc %d", m->c->clientID, m->rc);
Log(TRACE_MIN, -1, "Signaling websocket handshake for client %s rc %d", m->c->clientID, m->rc);
m->c->connect_state = WAIT_FOR_CONNACK;
Thread_post_sem(m->connect_sem);
Thread_signal_evt(m->connect_evt);
}
}
}
@ -1309,7 +1310,7 @@ static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_c
setSocketForSSLrc = SSLSocket_setSocketForSSL(&m->c->net, m->c->sslopts,
serverURI, hostname_len);
if (setSocketForSSLrc != MQTTCLIENT_SUCCESS)
if (setSocketForSSLrc == 1)
{
if (m->c->session != NULL)
if ((rc = SSL_set_session(m->c->net.ssl, m->c->session)) != 1)
@ -2008,11 +2009,11 @@ exit:
MQTTClient_stop();
if (call_connection_lost && m->cl && was_connected)
{
sync.sem = Thread_create_sem(&rc);
sync.evt = Thread_create_evt(&rc);
Log(TRACE_MIN, -1, "Calling connectionLost for client %s", m->c->clientID);
Paho_thread_start(connectionLost_call, &sync);
Thread_wait_sem(sync.sem, 5000);
Thread_destroy_sem(sync.sem);
Thread_wait_evt(sync.evt, 5000);
Thread_destroy_evt(sync.evt);
}
FUNC_EXIT_RC(rc);
return rc;
@ -2615,6 +2616,7 @@ static MQTTPacket* MQTTClient_cycle(SOCKET* sock, ELAPSED_TIME_TYPE timeout, int
static Ack ack;
MQTTPacket* pack = NULL;
int rc1 = 0;
int interrupted = 0;
START_TIME_TYPE start;
FUNC_ENTRY;
@ -2624,7 +2626,7 @@ static MQTTPacket* MQTTClient_cycle(SOCKET* sock, ELAPSED_TIME_TYPE timeout, int
/* 0 from getReadySocket indicates no work to do, rc -1 == error */
#endif
start = MQTTTime_start_clock();
*sock = Socket_getReadySocket(0, (int)timeout, socket_mutex, rc);
*sock = Socket_getReadySocket(0, (int)timeout, socket_mutex, rc, &interrupted);
*rc = rc1;
if (*sock == 0 && timeout >= 100L && MQTTTime_elapsed(start) < (int64_t)10)
MQTTTime_sleep(100L);
@ -2728,15 +2730,15 @@ static MQTTPacket* MQTTClient_waitfor(MQTTClient handle, int packet_type, int* r
{
if (packet_type == CONNECT)
{
if ((*rc = Thread_wait_sem(m->connect_sem, (int)timeout)) == 0)
if ((*rc = Thread_wait_evt(m->connect_evt, (int)timeout)) == 0)
*rc = m->rc;
}
else if (packet_type == CONNACK)
*rc = Thread_wait_sem(m->connack_sem, (int)timeout);
*rc = Thread_wait_evt(m->connack_evt, (int)timeout);
else if (packet_type == SUBACK)
*rc = Thread_wait_sem(m->suback_sem, (int)timeout);
*rc = Thread_wait_evt(m->suback_evt, (int)timeout);
else if (packet_type == UNSUBACK)
*rc = Thread_wait_sem(m->unsuback_sem, (int)timeout);
*rc = Thread_wait_evt(m->unsuback_evt, (int)timeout);
if (*rc == 0 && packet_type != CONNECT && m->pack == NULL)
Log(LOG_ERROR, -1, "waitfor unexpectedly is NULL for client %s, packet_type %d, timeout %ld", m->c->clientID, packet_type, timeout);
pack = m->pack;

12
src/MQTTClient.h
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2025 IBM Corp., Ian Craggs and others
* Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -41,11 +41,11 @@
* @endcond
* @cond MQTTClient_main
* @mainpage MQTT Client library for C (MQTTClient)
* &copy; Copyright 2009, 2025 IBM Corp., Ian Craggs and others
* &copy; Copyright 2009, 2026 IBM Corp., Ian Craggs and others
*
* @brief An MQTT client library in C.
*
* Version 1.3.15
* Version 1.3.16
*
* These pages describe the original more synchronous API which might be
* considered easier to use. Some of the calls will block. For the new
@ -508,13 +508,13 @@ LIBMQTT_API int MQTTClient_setPublished(MQTTClient handle, void* context, MQTTCl
* <br>
* @em tcp:// or @em mqtt:// - Insecure TCP
* <br>
* @em ssl:// or @em mqtts:// - Encrypted SSL/TLS
* @em ssl:// or @em tls:// or @em mqtts:// - Encrypted SSL/TLS
* <br>
* @em ws:// - Insecure websockets
* <br>
* @em wss:// - Secure web sockets
* <br>
* The TLS enabled prefixes (ssl, mqtts, wss) are only valid if a TLS
* The TLS enabled prefixes (ssl, tls, mqtts, wss) are only valid if a TLS
* version of the library is linked with.
* For <i>host</i>, you can specify either an IP address or a host name. For
* instance, to connect to a server running on the local machines with the
@ -927,7 +927,7 @@ typedef struct
* An optional array of null-terminated strings specifying the servers to
* which the client will connect. Each string takes the form <i>protocol://host:port</i>.
* <i>protocol</i> must be <i>tcp</i>, <i>ssl</i>, <i>ws</i> or <i>wss</i>.
* The TLS enabled prefixes (ssl, wss) are only valid if a TLS version of the library
* The TLS enabled prefixes (ssl, tls, wss) are only valid if a TLS version of the library
* is linked with.
* For <i>host</i>, you can
* specify either an IP address or a host name. For instance, to connect to

11
src/MQTTPacket.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2025 IBM Corp. and Ian Craggs
* Copyright (c) 2009, 2026 IBM Corp. and Ian Craggs
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -869,13 +869,14 @@ int MQTTPacket_formatPayload(int buflen, char* buf, int payloadlen, char* payloa
else
{
static char *hexdigit = "0123456789ABCDEF";
unsigned char curchar = (unsigned char)payload[i];
if (pos >= buflen - 3)
break;
buf[pos++] = '\\';
buf[pos++] = 'x';
buf[pos++] = hexdigit[payload[i] & 0xF0];
buf[pos++] = hexdigit[payload[i] & 0x0F];
buf[pos++] = hexdigit[(curchar & 0xF0) >> 4];
buf[pos++] = hexdigit[curchar & 0x0F];
}
}
@ -945,11 +946,7 @@ int MQTTPacket_send_publish(Publish* pack, int dup, int qos, int retained, netwo
memcpy(pack->mask, packetbufs.mask, sizeof(pack->mask));
}
{
#if defined(_WIN32)
#define buflen 30
#else
const int buflen = 30;
#endif
char buf[buflen];
int len = 0;

11
src/MQTTPersistence.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2023 IBM Corp.
* Copyright (c) 2009, 2026 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -655,14 +655,10 @@ void MQTTPersistence_wrapMsgID(Clients *client)
#if !defined(NO_PERSISTENCE)
#define KEYSIZE PERSISTENCE_MAX_KEY_LENGTH + 1
int MQTTPersistence_unpersistQueueEntry(Clients* client, MQTTPersistence_qEntry* qe)
{
int rc = 0;
#if defined(_WIN32)
#define KEYSIZE PERSISTENCE_MAX_KEY_LENGTH + 1
#else
const size_t KEYSIZE = PERSISTENCE_MAX_KEY_LENGTH + 1;
#endif
char key[KEYSIZE];
int chars = 0;
@ -688,9 +684,6 @@ int MQTTPersistence_persistQueueEntry(Clients* aclient, MQTTPersistence_qEntry*
{
int rc = 0;
int bufindex = 0;
#if !defined(_WIN32)
const size_t KEYSIZE = PERSISTENCE_MAX_KEY_LENGTH + 1;
#endif
char key[KEYSIZE];
int chars = 0;
int lens[MAX_NO_OF_BUFFERS];

11
src/MQTTProtocolClient.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2024 IBM Corp. and Ian Craggs
* Copyright (c) 2009, 2026 IBM Corp. and Ian Craggs
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -334,11 +334,7 @@ int MQTTProtocol_handlePublishes(void* pack, SOCKET sock)
/* Format and print publish data to trace */
{
#if defined(_WIN32)
#define buflen 30
#else
const int buflen = 30;
#endif
char buf[buflen];
int len = 0;
@ -463,7 +459,10 @@ int MQTTProtocol_handlePubacks(void* pack, SOCKET sock, Publications** pubToRemo
{
Messages* m = (Messages*)(client->outboundMsgs->current->content);
if (m->qos != 1)
Log(TRACE_MIN, 4, NULL, "PUBACK", client->clientID, puback->msgId, m->qos);
{
Log(LOG_ERROR, 4, NULL, "PUBACK", client->clientID, puback->msgId, m->qos);
rc = SOCKET_ERROR;
}
else
{
Log(TRACE_MIN, 6, NULL, "PUBACK", client->clientID, puback->msgId);

8
src/MQTTProtocolOut.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2025 IBM Corp., Ian Craggs and others
* Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -296,11 +296,15 @@ int MQTTProtocol_connect(const char* address, Clients* aClient, int unixsock, in
}
if (rc == 0 && SSLSocket_setSocketForSSL(&aClient->net, aClient->sslopts, address, addr_len) == 1)
{
rc = aClient->sslopts->struct_version >= 3 ?
int sslrc = aClient->sslopts->struct_version >= 3 ?
SSLSocket_connect(aClient->net.ssl, aClient->net.socket, address,
aClient->sslopts->verify, aClient->sslopts->ssl_error_cb, aClient->sslopts->ssl_error_context) :
SSLSocket_connect(aClient->net.ssl, aClient->net.socket, address,
aClient->sslopts->verify, NULL, NULL);
if (sslrc == 1) /* success */
rc = 0;
else if (sslrc < 0)
rc = sslrc;
if (rc == TCPSOCKET_INTERRUPTED)
aClient->connect_state = SSL_IN_PROGRESS; /* SSL connect called - wait for completion */
}

113
src/SSLSocket.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2023 IBM Corp., Ian Craggs
* Copyright (c) 2009, 2026 IBM Corp., Ian Craggs
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -27,6 +27,7 @@
*/
#if defined(OPENSSL)
#include "Heap.h"
#include "SocketBuffer.h"
#include "MQTTClient.h"
@ -724,7 +725,15 @@ int SSLSocket_setSocketForSSL(networkHandles* net, MQTTClient_SSLOptions* opts,
if (opts->enableServerCertAuth)
SSL_CTX_set_verify(net->ctx, SSL_VERIFY_PEER, NULL);
net->ssl = SSL_new(net->ctx);
if ((net->ssl = SSL_new(net->ctx)) == NULL)
{
if (opts->struct_version >= 3)
SSLSocket_error("SSL_new", net->ssl, net->socket, rc, opts->ssl_error_cb, opts->ssl_error_context);
else
SSLSocket_error("SSL_new", net->ssl, net->socket, rc, NULL, NULL);
rc = PAHO_MEMORY_ERROR;
goto exit;
}
/* Log all ciphers available to the SSL sessions (loaded in ctx) */
for (i = 0; ;i++)
@ -755,7 +764,7 @@ int SSLSocket_setSocketForSSL(networkHandles* net, MQTTClient_SSLOptions* opts,
else
rc = PAHO_MEMORY_ERROR;
}
exit:
FUNC_EXIT_RC(rc);
return rc;
}
@ -777,53 +786,72 @@ int SSLSocket_connect(SSL* ssl, SOCKET sock, const char* hostname, int verify, i
error = SSLSocket_error("SSL_connect", ssl, sock, rc, cb, u);
if (error == SSL_FATAL)
rc = error;
if (error == SSL_ERROR_WANT_READ || error == SSL_ERROR_WANT_WRITE)
rc = TCPSOCKET_INTERRUPTED;
}
#if (OPENSSL_VERSION_NUMBER >= 0x010002000) /* 1.0.2 and later */
else if (verify)
{
char* peername = NULL;
int port;
size_t hostname_len;
X509* cert = SSL_get_peer_certificate(ssl);
hostname_len = MQTTProtocol_addressPort(hostname, &port, NULL, MQTT_DEFAULT_PORT);
rc = X509_check_host(cert, hostname, hostname_len, 0, &peername);
if (rc == 1)
Log(TRACE_PROTOCOL, -1, "peername from X509_check_host is %s", peername);
else
Log(TRACE_PROTOCOL, -1, "X509_check_host for hostname %.*s failed, rc %d",
(int)hostname_len, hostname, rc);
if (peername != NULL)
OPENSSL_free(peername);
/* 0 == fail, -1 == SSL internal error, -2 == malformed input */
if (rc == 0 || rc == -1 || rc == -2)
{
char* ip_addr = malloc(hostname_len + 1);
/* cannot use = strndup(hostname, hostname_len); here because of custom Heap */
if (ip_addr)
switch (error)
{
strncpy(ip_addr, hostname, hostname_len);
ip_addr[hostname_len] = '\0';
case SSL_ERROR_WANT_READ:
rc = TCPSOCKET_INTERRUPTED;
Socket_clearPendingWrite(sock);
break;
case SSL_ERROR_WANT_WRITE:
rc = TCPSOCKET_INTERRUPTED;
Socket_addPendingWrite(sock);
break;
default:
Socket_clearPendingWrite(sock);
break;
}
}
}
else
{
Socket_clearPendingWrite(sock);
#if (OPENSSL_VERSION_NUMBER >= 0x010002000) /* 1.0.2 and later */
if (verify)
{
char* peername = NULL;
int port;
size_t hostname_len;
rc = X509_check_ip_asc(cert, ip_addr, 0);
Log(TRACE_MIN, -1, "rc from X509_check_ip_asc is %d", rc);
X509* cert = SSL_get_peer_certificate(ssl);
hostname_len = MQTTProtocol_addressPort(hostname, &port, NULL, MQTT_DEFAULT_PORT);
free(ip_addr);
rc = X509_check_host(cert, hostname, hostname_len, 0, &peername);
if (rc == 1)
Log(TRACE_PROTOCOL, -1, "peername from X509_check_host is %s", peername);
else
Log(TRACE_PROTOCOL, -1, "X509_check_host for hostname %.*s failed, rc %d",
(int)hostname_len, hostname, rc);
if (peername != NULL)
OPENSSL_free(peername);
/* 0 == fail, -1 == SSL internal error, -2 == malformed input */
if (rc == 0 || rc == -1 || rc == -2)
{
char* ip_addr = malloc(hostname_len + 1);
/* cannot use = strndup(hostname, hostname_len); here because of custom Heap */
if (ip_addr)
{
strncpy(ip_addr, hostname, hostname_len);
ip_addr[hostname_len] = '\0';
rc = X509_check_ip_asc(cert, ip_addr, 0);
Log(TRACE_MIN, -1, "rc from X509_check_ip_asc is %d", rc);
free(ip_addr);
}
if (rc == 0 || rc == -1 || rc == -2)
rc = SSL_FATAL;
}
if (rc == 0 || rc == -1 || rc == -2)
rc = SSL_FATAL;
if (cert)
X509_free(cert);
}
if (cert)
X509_free(cert);
}
#endif
}
FUNC_EXIT_RC(rc);
return rc;
@ -888,7 +916,8 @@ char *SSLSocket_getdata(SSL* ssl, SOCKET socket, size_t bytes, size_t* actual_le
goto exit;
}
buf = SocketBuffer_getQueuedData(socket, bytes, actual_len);
if ((buf = SocketBuffer_getQueuedData(socket, bytes, actual_len)) == NULL)
goto exit;
if (*actual_len != bytes)
{

237
src/Socket.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2025 IBM Corp., Ian Craggs and others
* Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -25,6 +25,10 @@
* \brief Socket related functions
*
* Some other related functions are in the SocketBuffer module
*
* Some portions of this file were generated by Claude:
* - the Windows implementation of socketpair()
*
*/
@ -124,7 +128,7 @@ int Socket_error(char* aString, SOCKET sock)
if (err != EINTR && err != EAGAIN && err != EINPROGRESS && err != EWOULDBLOCK)
{
if (strcmp(aString, "shutdown") != 0 || (err != ENOTCONN && err != ECONNRESET))
Log(TRACE_MINIMUM, -1, "Socket error %s(%d) in %s for socket %d", strerror(err), err, aString, sock);
Log(LOG_PROTOCOL, -1, "Socket error %s(%d) in %s for socket %d", strerror(err), err, aString, sock);
}
return err;
}
@ -146,6 +150,155 @@ void SIGPIPE_ignore()
}
#endif
/*
* socketpair() emulation for Windows
* Creates a pair of connected sockets using TCP loopback
*
* This function was generated by Claude
*/
#if defined(_WIN32)
#ifndef AF_LOCAL
#define AF_LOCAL AF_INET
#endif
int socketpair(int domain, int type, int protocol, SOCKET socks[2])
{
SOCKET listener = INVALID_SOCKET;
SOCKET client = INVALID_SOCKET;
SOCKET server = INVALID_SOCKET;
struct sockaddr_in addr;
int addr_len = sizeof(addr);
int reuse = 1;
/* Only support AF_INET (or AF_UNIX which we map to AF_INET) */
if (domain != AF_INET && domain != AF_UNIX) {
errno = EAFNOSUPPORT;
return -1;
}
/* Only support SOCK_STREAM */
if (type != SOCK_STREAM) {
errno = EPROTONOSUPPORT;
return -1;
}
socks[0] = socks[1] = INVALID_SOCKET;
/* Create listener socket on loopback */
listener = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (listener == INVALID_SOCKET) {
goto error;
}
/* Allow rapid reuse of the address */
setsockopt(listener, SOL_SOCKET, SO_REUSEADDR,
(const char*)&reuse, sizeof(reuse));
/* Bind to loopback with OS-assigned port */
memset(&addr, 0, sizeof(addr));
addr.sin_family = AF_INET;
addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
addr.sin_port = 0; /* Let OS choose port */
if (bind(listener, (struct sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR) {
goto error;
}
/* Get the actual address (with port assigned by OS) */
if (getsockname(listener, (struct sockaddr*)&addr, &addr_len) == SOCKET_ERROR) {
goto error;
}
/* Listen for connection */
if (listen(listener, 1) == SOCKET_ERROR) {
goto error;
}
/* Create client socket */
client = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
if (client == INVALID_SOCKET) {
goto error;
}
/* Connect to listener */
if (connect(client, (struct sockaddr*)&addr, sizeof(addr)) == SOCKET_ERROR) {
goto error;
}
/* Accept connection */
server = accept(listener, NULL, NULL);
if (server == INVALID_SOCKET) {
goto error;
}
/* Close listener - no longer needed */
closesocket(listener);
/* Return the pair */
socks[0] = client;
socks[1] = server;
return 0;
error:
{
int saved_errno = WSAGetLastError();
if (listener != INVALID_SOCKET) {
closesocket(listener);
}
if (client != INVALID_SOCKET) {
closesocket(client);
}
if (server != INVALID_SOCKET) {
closesocket(server);
}
/* Map Winsock error to errno */
switch (saved_errno) {
case WSAEAFNOSUPPORT:
errno = EAFNOSUPPORT;
break;
case WSAEPROTONOSUPPORT:
errno = EPROTONOSUPPORT;
break;
case WSAEMFILE:
case WSAENOBUFS:
errno = EMFILE;
break;
default:
errno = EIO;
break;
}
return -1;
}
}
#endif /* _WIN32 */
static SOCKET sockfd[2];
int Socket_pair()
{
int rc = 0;
FUNC_ENTRY;
if ((rc = socketpair(AF_LOCAL, SOCK_STREAM, 0, sockfd)) == 0) /* 0 if successful */
rc = Socket_addSocket(sockfd[0]);
FUNC_EXIT_RC(rc);
return rc;
}
int Socket_interrupt()
{
FUNC_ENTRY;
int rc = send(sockfd[1], "\0", 1, 0);
FUNC_EXIT_RC(rc);
return rc;
}
/**
* Initialize the socket module
*/
@ -183,6 +336,7 @@ void Socket_outInitialize(void)
mod_s.saved.fds_read = NULL;
mod_s.saved.nfds = 0;
#endif
Socket_pair();
FUNC_EXIT;
}
@ -284,6 +438,7 @@ static int cmpsockfds(const void *p1, const void *p2)
/**
* Add a socket to the list of socket to check with select
* @param newSd the new socket to add
* @return 0 if successful
*/
int Socket_addSocket(SOCKET newSd)
{
@ -394,7 +549,9 @@ int isReady(int index)
FUNC_ENTRY;
if ((mod_s.saved.fds_read[index].revents & POLLHUP) || (mod_s.saved.fds_read[index].revents & POLLNVAL))
if (*socket == sockfd[0]) /* don't return the interrupting socket */
rc = 0;
else if ((mod_s.saved.fds_read[index].revents & POLLHUP) || (mod_s.saved.fds_read[index].revents & POLLNVAL))
; /* signal work to be done if there is an error on the socket */
else if (ListFindItem(mod_s.connect_pending, socket, intcompare) &&
(mod_s.saved.fds_write[index].revents & POLLOUT))
@ -409,6 +566,27 @@ int isReady(int index)
}
#endif
/**
* Check whether the socket indicated by the index is in the
* TCP connect pending state.
* @param index
* @return 1 if the socket has a TCP connect pending, otherwise 0
*/
int isConnectPending(int index)
{
int rc = 0;
SOCKET* socket = &mod_s.saved.fds_write[index].fd;
FUNC_ENTRY;
if (ListFindItem(mod_s.connect_pending, socket, intcompare))
Log(TRACE_PROTOCOL, -1, "TCP connect is pending for socket %d, POLLOUT %d",
*socket, (mod_s.saved.fds_write[index].revents & POLLOUT));
if (ListFindItem(mod_s.connect_pending, socket, intcompare))
rc = 1;
FUNC_EXIT_RC(rc);
return rc;
}
#if defined(USE_SELECT)
/**
@ -528,7 +706,7 @@ exit:
* @param rc a value other than 0 indicates an error of the returned socket
* @return the socket next ready, or 0 if none is ready
*/
SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int* rc)
SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int* rc, int* interrupted)
{
SOCKET sock = 0;
*rc = 0;
@ -628,7 +806,29 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
*rc = SOCKET_ERROR;
goto exit;
}
/* if there is a writeable socket in the connecting state then set the timeout for the
* next poll to 0 or a low value, so we can handle the connect immediately */
mod_s.saved.cur_fd = 0;
while (mod_s.saved.cur_fd != -1)
{
if (isConnectPending(mod_s.saved.cur_fd))
{
Log(TRACE_MINIMUM, -1, "Socket %d is connecting, reducing timeout from %d, no of sockets %d\n",
mod_s.saved.cur_fd, timeout_ms, mod_s.saved.nfds);
/* Will this result in excessive CPU use if the TCP connect is taking a long time,
* such as on a satellite link? We could use a backoff algorithm if it turns out to be true */
timeout_ms = 0;
break;
}
mod_s.saved.cur_fd = (mod_s.saved.cur_fd == mod_s.saved.nfds - 1) ? -1 : mod_s.saved.cur_fd + 1;
}
/* clear interrupt socket - make sure it is unset */
{
static char dbuf[1];
/*int irc = */recv(sockfd[0], dbuf, 1, 0);
}
/* Prevent performance issue by unlocking the socket_mutex while waiting for a ready socket. */
Paho_thread_unlock_mutex(mutex);
*rc = poll(mod_s.saved.fds_read, mod_s.saved.nfds, timeout_ms);
@ -640,6 +840,17 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
}
Log(TRACE_MAX, -1, "Return code %d from poll", *rc);
/* check interrupt socket to see if we were interrupted */
{
static char dbuf[1];
int irc = recv(sockfd[0], dbuf, 1, 0);
if (irc > 0)
{
*interrupted = 1;
Log(TRACE_MINIMUM, -1, "Poll was interrupted, rc %d", irc);
}
}
if (rc1 == 0 && *rc == 0)
{
sock = 0;
@ -955,6 +1166,7 @@ int Socket_close_only(SOCKET socket)
int rc;
FUNC_ENTRY;
Socket_interrupt();
#if defined(_WIN32)
if (shutdown(socket, SD_BOTH) == SOCKET_ERROR)
Socket_error("shutdown", socket);
@ -1245,10 +1457,20 @@ int Socket_new(const char* addr, size_t addr_len, int port, SOCKET* sock)
else
{
#if defined(NOSIGPIPE)
int opt = 1;
{
int opt = 1;
if (setsockopt(*sock, SOL_SOCKET, SO_NOSIGPIPE, (void*)&opt, sizeof(opt)) != 0)
Log(LOG_ERROR, -1, "Could not set SO_NOSIGPIPE for socket %d", *sock);
if (setsockopt(*sock, SOL_SOCKET, SO_NOSIGPIPE, (void*)&opt, sizeof(opt)) != 0)
Log(LOG_ERROR, -1, "Could not set SO_NOSIGPIPE for socket %d", *sock);
}
#endif
#if !defined(NO_TCP_NODELAY)
{
int opt = 1;
if (setsockopt(*sock, IPPROTO_TCP, TCP_NODELAY, (void*)&opt, sizeof(opt)) != 0)
Log(LOG_ERROR, -1, "Could not set TCP_NODELAY for socket %d", *sock);
}
#endif
/*#define SMALL_TCP_BUFFER_TESTING
This section sets the TCP send buffer to a small amount to provoke TCPSOCKET_INTERRUPTED
@ -1298,6 +1520,7 @@ int Socket_new(const char* addr, size_t addr_len, int port, SOCKET* sock)
goto exit;
}
Log(TRACE_MIN, 15, "Connect pending");
Socket_interrupt();
}
}
/* Prevent socket leak by closing unusable sockets,

11
src/Socket.h
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2022 IBM Corp., Ian Craggs and others
* Copyright (c) 2009, 2026 IBM Corp., Ian Craggs and others
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -61,6 +61,9 @@
#include <netinet/tcp.h>
#include <arpa/inet.h>
#include <netdb.h>
#ifdef __QNXNTO__
#define AI_ADDRCONFIG 0
#endif
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
@ -70,7 +73,7 @@
#define SOCKET int
#endif
#include "mutex_type.h" /* Needed for mutex_type */
#include "Thread.h" /* Needed for mutex_type */
/** socket operation completed successfully */
#define TCPSOCKET_COMPLETE 0
@ -138,7 +141,7 @@ typedef struct
void Socket_outInitialize(void);
void Socket_outTerminate(void);
SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int* rc);
SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int* rc, int* interrupted);
int Socket_getch(SOCKET socket, char* c);
char *Socket_getdata(SOCKET socket, size_t bytes, size_t* actual_len, int* rc);
int Socket_putdatas(SOCKET socket, char* buf0, size_t buf0len, PacketBuffers bufs);
@ -166,4 +169,6 @@ void Socket_setWriteCompleteCallback(Socket_writeComplete*);
typedef void Socket_writeAvailable(SOCKET socket);
void Socket_setWriteAvailableCallback(Socket_writeAvailable*);
int Socket_interrupt();
#endif /* SOCKET_H */

633
src/Thread.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2025 IBM Corp. and Ian Craggs
* Copyright (c) 2009, 2026 IBM Corp. and Ian Craggs
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -17,6 +17,7 @@
* Ian Craggs - fix for bug #420851
* Ian Craggs - change MacOS semaphore implementation
* Ian Craggs - fix for clock #284
* Frank Pagliughi - Consolidated semaphores and conditions into "events"
*******************************************************************************/
/**
@ -28,9 +29,6 @@
#include "Thread.h"
#if defined(THREAD_UNIT_TESTS)
#define NOSTACKTRACE
#endif
#include "Log.h"
#include "StackTrace.h"
@ -87,10 +85,12 @@ void Paho_thread_start(thread_fn fn, void* parameter)
int Thread_set_name(const char* thread_name)
{
int rc = 0;
/*
#if defined(_WIN32)
#define MAX_THREAD_NAME_LENGTH 30
wchar_t wchars[MAX_THREAD_NAME_LENGTH];
#endif
*/
FUNC_ENTRY;
#if defined(_WIN32)
@ -116,6 +116,54 @@ int Thread_set_name(const char* thread_name)
return rc;
}
/**
* Get the thread id of the thread from which this function is called
* @return thread id, type varying according to OS
*/
thread_id_type Paho_thread_getid(void)
{
#if defined(_WIN32)
return GetCurrentThreadId();
#else
return pthread_self();
#endif
}
#if !defined(_WIN32)
struct timespec Thread_time_from_now(int ms)
{
struct timespec from_now;
struct timespec interval;
interval.tv_sec = ms / 1000;
interval.tv_nsec = (ms % 1000) * 1000000L;
/*
* Note that MacOS v10.11 was fully deprecated in Aug 2018.
* So this __APPLE__ path only applies to very old, deprecated systems.
*/
#if defined(__APPLE__) && __MAC_OS_X_VERSION_MIN_REQUIRED < 101200
struct timeval cur_time;
gettimeofday(&cur_time, NULL);
from_now.tv_sec = cur_time.tv_sec;
from_now.tv_nsec = cur_time.tv_usec * 1000;
#else
clock_gettime(CLOCK_REALTIME, &from_now);
#endif
from_now.tv_sec += interval.tv_sec;
from_now.tv_nsec += interval.tv_sec;
while (from_now.tv_nsec >= NSEC_PER_SEC)
{
from_now.tv_sec++;
from_now.tv_nsec -= NSEC_PER_SEC;
}
return from_now;
}
#endif
/* Mutex Functions */
/**
* Create a new mutex
@ -205,495 +253,118 @@ int Paho_thread_destroy_mutex(mutex_type mutex)
}
/**
* Get the thread id of the thread from which this function is called
* @return thread id, type varying according to OS
*/
thread_id_type Paho_thread_getid(void)
{
#if defined(_WIN32)
return GetCurrentThreadId();
#else
return pthread_self();
#endif
}
/* Event functions */
/**
* Create a new semaphore
* @param rc return code: 0 for success, negative otherwise
* @return the new condition variable
* Create a new event
* @return the event struct
*/
sem_type Thread_create_sem(int *rc)
evt_type Thread_create_evt(int *rc)
{
sem_type sem = NULL;
FUNC_ENTRY;
*rc = -1;
#if defined(_WIN32)
sem = CreateEvent(
NULL, /* default security attributes */
FALSE, /* manual-reset event? */
FALSE, /* initial state is nonsignaled */
NULL /* object name */
);
*rc = (sem == NULL) ? GetLastError() : 0;
#elif defined(OSX)
sem = dispatch_semaphore_create(0L);
*rc = (sem == NULL) ? -1 : 0;
#else
sem = malloc(sizeof(sem_t));
if (sem)
*rc = sem_init(sem, 0, 0);
#endif
FUNC_EXIT_RC(*rc);
return sem;
}
/**
* Wait for a semaphore to be posted, or timeout.
* @param sem the semaphore
* @param timeout the maximum time to wait, in milliseconds
* @return completion code
*/
int Thread_wait_sem(sem_type sem, int timeout)
{
/* sem_timedwait is the obvious call to use, but seemed not to work on the Viper,
* so I've used trywait in a loop instead. Ian Craggs 23/7/2010
*/
int rc = -1;
#if !defined(_WIN32) && !defined(OSX)
#define USE_TRYWAIT
#if defined(USE_TRYWAIT)
int i = 0;
useconds_t interval = 10000; /* 10000 microseconds: 10 milliseconds */
int count = (1000 * timeout) / interval; /* how many intervals in timeout period */
#else
struct timespec ts;
#endif
#endif
FUNC_ENTRY;
#if defined(_WIN32)
/* returns 0 (WAIT_OBJECT_0) on success, non-zero (WAIT_TIMEOUT) if timeout occurred */
rc = WaitForSingleObject(sem, timeout < 0 ? 0 : timeout);
if (rc == WAIT_TIMEOUT)
rc = ETIMEDOUT;
#elif defined(OSX)
/* returns 0 on success, non-zero if timeout occurred */
rc = (int)dispatch_semaphore_wait(sem, dispatch_time(DISPATCH_TIME_NOW, (int64_t)timeout*1000000L));
if (rc != 0)
rc = ETIMEDOUT;
#elif defined(USE_TRYWAIT)
while (++i < count && (rc = sem_trywait(sem)) != 0)
{
if (rc == -1 && ((rc = errno) != EAGAIN))
{
rc = 0;
break;
}
usleep(interval); /* microseconds - .1 of a second */
}
#else
/* We have to use CLOCK_REALTIME rather than MONOTONIC for sem_timedwait interval.
* Does this make it susceptible to system clock changes?
* The intervals are small enough, and repeated, that I think it's not an issue.
*/
if (clock_gettime(CLOCK_REALTIME, &ts) != -1)
{
ts.tv_sec += timeout;
rc = sem_timedwait(sem, &ts);
}
#endif
FUNC_EXIT_RC(rc);
return rc;
}
/**
* Check to see if a semaphore has been posted, without waiting
* The semaphore will be unchanged, if the return value is false.
* The semaphore will have been decremented, if the return value is true.
* @param sem the semaphore
* @return 0 (false) or 1 (true)
*/
int Thread_check_sem(sem_type sem)
{
#if defined(_WIN32)
/* if the return value is not 0, the semaphore will not have been decremented */
return WaitForSingleObject(sem, 0) == WAIT_OBJECT_0;
#elif defined(OSX)
/* if the return value is not 0, the semaphore will not have been decremented */
return dispatch_semaphore_wait(sem, DISPATCH_TIME_NOW) == 0;
#else
/* If the call was unsuccessful, the state of the semaphore shall be unchanged,
* and the function shall return a value of -1 */
return sem_trywait(sem) == 0;
#endif
}
/**
* Post a semaphore
* @param sem the semaphore
* @return 0 on success
*/
int Thread_post_sem(sem_type sem)
{
int rc = 0;
FUNC_ENTRY;
#if defined(_WIN32)
if (SetEvent(sem) == 0)
rc = GetLastError();
#elif defined(OSX)
rc = (int)dispatch_semaphore_signal(sem);
#else
int val;
int rc1 = sem_getvalue(sem, &val);
if (rc1 != 0)
rc = errno;
else if (val == 0 && sem_post(sem) == -1)
rc = errno;
#endif
FUNC_EXIT_RC(rc);
return rc;
}
/**
* Destroy a semaphore which has already been created
* @param sem the semaphore
*/
int Thread_destroy_sem(sem_type sem)
{
int rc = 0;
FUNC_ENTRY;
#if defined(_WIN32)
rc = CloseHandle(sem);
#elif defined(OSX)
dispatch_release(sem);
#else
rc = sem_destroy(sem);
free(sem);
#endif
FUNC_EXIT_RC(rc);
return rc;
}
evt_type evt = NULL;
#if !defined(_WIN32)
/**
* Create a new condition variable
* @return the condition variable struct
*/
cond_type Thread_create_cond(int *rc)
{
cond_type condvar = NULL;
pthread_condattr_t attr;
#endif
FUNC_ENTRY;
*rc = -1;
pthread_condattr_init(&attr);
#if 0
/* in theory, a monotonic clock should be able to be used. However on at least
* one system reported, even though setclock() reported success, it didn't work.
*/
if ((rc = pthread_condattr_setclock(&attr, CLOCK_MONOTONIC)) == 0)
use_clock_monotonic = 1;
else
Log(LOG_ERROR, -1, "Error %d calling pthread_condattr_setclock(CLOCK_MONOTONIC)", rc);
#endif
condvar = malloc(sizeof(cond_type_struct));
if (condvar)
{
*rc = pthread_cond_init(&condvar->cond, &attr);
*rc = pthread_mutex_init(&condvar->mutex, NULL);
}
FUNC_EXIT_RC(*rc);
return condvar;
}
/**
* Signal a condition variable
* @return completion code
*/
int Thread_signal_cond(cond_type condvar)
{
int rc = 0;
FUNC_ENTRY;
pthread_mutex_lock(&condvar->mutex);
rc = pthread_cond_signal(&condvar->cond);
pthread_mutex_unlock(&condvar->mutex);
FUNC_EXIT_RC(rc);
return rc;
}
/**
* Wait with a timeout (ms) for condition variable
* @return 0 for success, ETIMEDOUT otherwise
*/
int Thread_wait_cond(cond_type condvar, int timeout_ms)
{
int rc = 0;
struct timespec cond_timeout;
FUNC_ENTRY;
#if defined(__APPLE__) && __MAC_OS_X_VERSION_MIN_REQUIRED < 101200 /* for older versions of MacOS */
struct timeval cur_time;
gettimeofday(&cur_time, NULL);
cond_timeout.tv_sec = cur_time.tv_sec;
cond_timeout.tv_nsec = cur_time.tv_usec * 1000;
#else
clock_gettime(CLOCK_REALTIME, &cond_timeout);
#endif
if (timeout_ms > 0) {
struct timespec interval;
interval.tv_sec = timeout_ms / 1000;
interval.tv_nsec = (timeout_ms % 1000) * 1000000L;
cond_timeout.tv_sec += interval.tv_sec;
cond_timeout.tv_nsec += interval.tv_nsec;
while (cond_timeout.tv_nsec >= NSEC_PER_SEC)
{
cond_timeout.tv_sec++;
cond_timeout.tv_nsec -= NSEC_PER_SEC;
}
}
pthread_mutex_lock(&condvar->mutex);
rc = pthread_cond_timedwait(&condvar->cond, &condvar->mutex, &cond_timeout);
pthread_mutex_unlock(&condvar->mutex);
FUNC_EXIT_RC(rc);
return rc;
}
/**
* Destroy a condition variable
* @return completion code
*/
int Thread_destroy_cond(cond_type condvar)
{
int rc = 0;
rc = pthread_mutex_destroy(&condvar->mutex);
rc = pthread_cond_destroy(&condvar->cond);
free(condvar);
return rc;
}
#endif
#if defined(THREAD_UNIT_TESTS)
#if defined(_WIN32) || defined(_WINDOWS)
#define mqsleep(A) Sleep(1000*A)
#define START_TIME_TYPE DWORD
static DWORD start_time = 0;
START_TIME_TYPE start_clock(void)
{
return GetTickCount();
}
#elif defined(AIX)
#define mqsleep sleep
#define START_TIME_TYPE struct timespec
START_TIME_TYPE start_clock(void)
{
static struct timespec start;
clock_gettime(CLOCK_REALTIME, &start);
return start;
}
#else
#define mqsleep sleep
#define START_TIME_TYPE struct timeval
/* TODO - unused - remove? static struct timeval start_time; */
START_TIME_TYPE start_clock(void)
{
struct timeval start_time;
gettimeofday(&start_time, NULL);
return start_time;
}
#endif
#if defined(_WIN32)
long elapsed(START_TIME_TYPE start_time)
{
return GetTickCount() - start_time;
}
#elif defined(AIX)
#define assert(a)
long elapsed(struct timespec start)
{
struct timespec now, res;
clock_gettime(CLOCK_REALTIME, &now);
ntimersub(now, start, res);
return (res.tv_sec)*1000L + (res.tv_nsec)/1000000L;
}
evt = CreateEvent(
NULL, /* default security attributes */
FALSE, /* manual-reset event? */
FALSE, /* initial state is nonsignaled */
NULL /* object name */
);
if (rc)
*rc = (evt == NULL) ? GetLastError() : 0;
#else
long elapsed(START_TIME_TYPE start_time)
{
struct timeval now, res;
pthread_condattr_init(&attr);
gettimeofday(&now, NULL);
timersub(&now, &start_time, &res);
return (res.tv_sec)*1000 + (res.tv_usec)/1000;
}
#endif
int tests = 0, failures = 0;
void myassert(char* filename, int lineno, char* description, int value, char* format, ...)
{
++tests;
if (!value)
evt = malloc(sizeof(evt_type_struct));
if (evt)
{
va_list args;
++failures;
printf("Assertion failed, file %s, line %d, description: %s\n", filename, lineno, description);
va_start(args, format);
vprintf(format, args);
va_end(args);
//cur_output += sprintf(cur_output, "<failure type=\"%s\">file %s, line %d </failure>\n",
// description, filename, lineno);
*rc = pthread_cond_init(&evt->cond, &attr);
*rc = pthread_mutex_init(&evt->mutex, NULL);
evt->val = 0;
}
else
printf("Assertion succeeded, file %s, line %d, description: %s\n", filename, lineno, description);
}
#define assert(a, b, c, d) myassert(__FILE__, __LINE__, a, b, c, d)
#define assert1(a, b, c, d, e) myassert(__FILE__, __LINE__, a, b, c, d, e)
#include <stdio.h>
thread_return_type cond_secondary(void* n)
{
int rc = 0;
cond_type cond = n;
printf("This should return immediately as it was posted already\n");
rc = Thread_wait_cond(cond, 99999);
assert("rc 1 from wait_cond", rc == 1, "rc was %d", rc);
printf("This should hang around a few seconds\n");
rc = Thread_wait_cond(cond, 99999);
assert("rc 1 from wait_cond", rc == 1, "rc was %d", rc);
printf("Secondary cond thread ending\n");
return 0;
}
int cond_test()
{
int rc = 0;
cond_type cond = Thread_create_cond();
printf("Post secondary so it should return immediately\n");
rc = Thread_signal_cond(cond);
assert("rc 0 from signal cond", rc == 0, "rc was %d", rc);
printf("Starting secondary thread\n");
Thread_start(cond_secondary, (void*)cond);
sleep(3);
printf("post secondary\n");
rc = Thread_signal_cond(cond);
assert("rc 1 from signal cond", rc == 1, "rc was %d", rc);
sleep(3);
printf("Main thread ending\n");
return failures;
}
thread_return_type sem_secondary(void* n)
{
int rc = 0;
sem_type sem = n;
printf("Secondary semaphore pointer %p\n", sem);
rc = Thread_check_sem(sem);
assert("rc 1 from check_sem", rc == 1, "rc was %d", rc);
printf("Secondary thread about to wait\n");
rc = Thread_wait_sem(sem, 99999);
printf("Secondary thread returned from wait %d\n", rc);
printf("Secondary thread about to wait\n");
rc = Thread_wait_sem(sem, 99999);
printf("Secondary thread returned from wait %d\n", rc);
printf("Secondary check sem %d\n", Thread_check_sem(sem));
printf("Secondary thread ending\n");
return 0;
}
int sem_test()
{
int rc = 0;
sem_type sem = Thread_create_sem();
printf("Primary semaphore pointer %p\n", sem);
rc = Thread_check_sem(sem);
assert("rc 0 from check_sem", rc == 0, "rc was %d\n", rc);
printf("post secondary so then check should be 1\n");
rc = Thread_post_sem(sem);
assert("rc 0 from post_sem", rc == 0, "rc was %d\n", rc);
rc = Thread_check_sem(sem);
assert("rc 1 from check_sem", rc == 1, "rc was %d", rc);
printf("Starting secondary thread\n");
Thread_start(sem_secondary, (void*)sem);
sleep(3);
rc = Thread_check_sem(sem);
assert("rc 1 from check_sem", rc == 1, "rc was %d", rc);
printf("post secondary\n");
rc = Thread_post_sem(sem);
assert("rc 1 from post_sem", rc == 1, "rc was %d", rc);
sleep(3);
printf("Main thread ending\n");
return failures;
}
int main(int argc, char *argv[])
{
sem_test();
//cond_test();
}
#endif
FUNC_EXIT_RC(*rc);
return evt;
}
/**
* Signal an event
* @return completion code, 0 is success
*/
int Thread_signal_evt(evt_type evt)
{
int rc = 0;
FUNC_ENTRY;
#if defined(_WIN32)
if (SetEvent(evt) == 0)
rc = GetLastError();
#else
pthread_mutex_lock(&evt->mutex);
evt->val = 1;
// TODO: Determine if this could be more efficient with pthread_cond_signal()
rc = pthread_cond_broadcast(&evt->cond);
pthread_mutex_unlock(&evt->mutex);
#endif
FUNC_EXIT_RC(rc);
return rc;
}
/**
* Wait with a timeout (ms) for the event to become signaled.
* @return The completion code: 0 for success, ETIMEDOUT otherwise
*/
int Thread_wait_evt(evt_type evt, int timeout_ms)
{
int rc = 0;
FUNC_ENTRY;
#if defined(_WIN32)
/* returns 0 (WAIT_OBJECT_0) on success, non-zero (WAIT_TIMEOUT) if timeout occurred */
rc = WaitForSingleObject(evt, timeout_ms < 0 ? 0 : timeout_ms);
if (rc == WAIT_TIMEOUT)
rc = ETIMEDOUT;
#else
struct timespec evt_timeout;
evt_timeout = Thread_time_from_now(timeout_ms);
pthread_mutex_lock(&evt->mutex);
while (evt->val == 0 &&
(rc = pthread_cond_timedwait(&evt->cond, &evt->mutex, &evt_timeout)) == 0)
;
if (rc == 0) {
evt->val = 0;
}
pthread_mutex_unlock(&evt->mutex);
#endif
FUNC_EXIT_RC(rc);
return rc;
}
/**
* Destroy an event object
* @return completion code, 0 for success
*/
int Thread_destroy_evt(evt_type evt)
{
int rc = 0;
FUNC_ENTRY;
#if defined(_WIN32)
rc = CloseHandle(evt);
#else
rc = pthread_mutex_destroy(&evt->mutex);
int rcc = pthread_cond_destroy(&evt->cond);
if (rcc != 0)
rc = rcc;
free(evt);
#endif
FUNC_EXIT_RC(rc);
return rc;
}

43
src/Thread.h
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2023 IBM Corp.
* Copyright (c) 2009, 2026 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -15,6 +15,7 @@
* Ian Craggs, Allan Stockdill-Mander - async client updates
* Ian Craggs - fix for bug #420851
* Ian Craggs - change MacOS semaphore implementation
* Frank Pagliughi - Consolidated semaphores and conditions into "events"
*******************************************************************************/
#if !defined(THREAD_H)
@ -33,56 +34,44 @@
#include "MQTTClient.h"
#include "mutex_type.h" /* Needed for mutex_type */
#if defined(_WIN32)
#include <windows.h>
#define mutex_type HANDLE
#define thread_type HANDLE
#define thread_id_type DWORD
#define thread_return_type DWORD
#define thread_fn LPTHREAD_START_ROUTINE
#define cond_type HANDLE
#define sem_type HANDLE
#define evt_type HANDLE
#undef ETIMEDOUT
#define ETIMEDOUT WSAETIMEDOUT
#else
#include <pthread.h>
#define mutex_type pthread_mutex_t*
#define thread_type pthread_t
#define thread_id_type pthread_t
#define thread_return_type void*
typedef thread_return_type (*thread_fn)(void*);
typedef struct { pthread_cond_t cond; pthread_mutex_t mutex; } cond_type_struct;
typedef cond_type_struct *cond_type;
#if defined(OSX)
#include <dispatch/dispatch.h>
typedef dispatch_semaphore_t sem_type;
#else
#include <semaphore.h>
typedef sem_t *sem_type;
#endif
cond_type Thread_create_cond(int*);
int Thread_signal_cond(cond_type);
int Thread_wait_cond(cond_type condvar, int timeout);
int Thread_destroy_cond(cond_type);
typedef struct { pthread_cond_t cond; pthread_mutex_t mutex; int val; } evt_type_struct;
typedef evt_type_struct *evt_type;
#endif
/* Thread functions */
LIBMQTT_API void Paho_thread_start(thread_fn, void*);
int Thread_set_name(const char* thread_name);
LIBMQTT_API thread_id_type Paho_thread_getid();
/* Mutex functions */
LIBMQTT_API mutex_type Paho_thread_create_mutex(int*);
LIBMQTT_API int Paho_thread_lock_mutex(mutex_type);
LIBMQTT_API int Paho_thread_unlock_mutex(mutex_type);
int Paho_thread_destroy_mutex(mutex_type);
LIBMQTT_API thread_id_type Paho_thread_getid();
sem_type Thread_create_sem(int*);
int Thread_wait_sem(sem_type sem, int timeout);
int Thread_check_sem(sem_type sem);
int Thread_post_sem(sem_type sem);
int Thread_destroy_sem(sem_type sem);
/* Event Functions */
evt_type Thread_create_evt(int*);
int Thread_signal_evt(evt_type);
int Thread_wait_evt(evt_type condvar, int timeout);
int Thread_destroy_evt(evt_type);
#endif

25
src/mutex_type.h
View File

@ -1,25 +0,0 @@
/*******************************************************************************
* Copyright (c) 2009, 2018 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
* and Eclipse Distribution License v1.0 which accompany this distribution.
*
* The Eclipse Public License is available at
* https://www.eclipse.org/legal/epl-2.0/
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
*******************************************************************************/
#if !defined(_MUTEX_TYPE_H_)
#define _MUTEX_TYPE_H_
#if defined(_WIN32)
#include <windows.h>
#define mutex_type HANDLE
#else
#include <pthread.h>
#define mutex_type pthread_mutex_t*
#endif
#endif /* _MUTEX_TYPE_H_ */

49
test/CMakeLists.txt
View File

@ -1207,8 +1207,53 @@ if(PAHO_WITH_SSL OR PAHO_WITH_LIBRESSL)
PROPERTIES TIMEOUT 540
)
endif()
endif()
ENDIF()
# SSL WANT_WRITE handshake test - validates proper handling of SSL_ERROR_WANT_WRITE/WANT_READ
IF (PAHO_BUILD_STATIC)
ADD_EXECUTABLE(
test_ssl_want_write-static
test_ssl_want_write.c
)
TARGET_LINK_LIBRARIES(
test_ssl_want_write-static
paho-mqtt3cs-static
)
ADD_TEST(
NAME test-ssl-want-write-handshake-static
COMMAND test_ssl_want_write-static "--hostname" ${MQTT_SSL_HOSTNAME} "--client_key" "${CERTDIR}/client.pem" "--server_key" "${CERTDIR}/test-root-ca.crt"
)
SET_TESTS_PROPERTIES(
test-ssl-want-write-handshake-static
PROPERTIES TIMEOUT 60
)
ENDIF()
IF (PAHO_BUILD_SHARED)
ADD_EXECUTABLE(
test_ssl_want_write
test_ssl_want_write.c
)
TARGET_LINK_LIBRARIES(
test_ssl_want_write
paho-mqtt3cs
)
ADD_TEST(
NAME test-ssl-want-write-handshake
COMMAND test_ssl_want_write "--hostname" ${MQTT_SSL_HOSTNAME} "--client_key" "${CERTDIR}/client.pem" "--server_key" "${CERTDIR}/test-root-ca.crt"
)
SET_TESTS_PROPERTIES(
test-ssl-want-write-handshake
PROPERTIES TIMEOUT 60
)
ENDIF()
ENDIF()
if(PAHO_BUILD_STATIC)

4
test/test_connect_destroy.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2024 IBM Corp. and others
* Copyright (c) 2009, 2026 IBM Corp. and others
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -389,7 +389,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);

252
test/test_ssl_want_write.c Normal file
View File

@ -0,0 +1,252 @@
/*******************************************************************************
* Copyright (c) 2024, 2026 Ian Craggs and others
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
* and Eclipse Distribution License v1.0 which accompany this distribution.
*
* The Eclipse Public License is available at
* https://www.eclipse.org/legal/epl-2.0/
* and the Eclipse Distribution License is available at
* http://www.eclipse.org/org/documents/edl-v10.php.
*
* Contributors:
* Unit test for SSL_ERROR_WANT_WRITE/WANT_READ handling during SSL handshake
*******************************************************************************/
/**
* @file
* Test for SSL_ERROR_WANT_WRITE and SSL_ERROR_WANT_READ handling in SSLSocket_connect()
*
* This test verifies that SSL connections complete successfully even when
* SSL_ERROR_WANT_WRITE or SSL_ERROR_WANT_READ occur during the handshake.
* The fix ensures proper socket polling by calling Socket_addPendingWrite()
* when WANT_WRITE occurs.
*/
#include "MQTTClient.h"
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <stdarg.h>
#include <time.h>
#if !defined(_WINDOWS)
#include <sys/time.h>
#include <unistd.h>
#endif
/* Test configuration */
struct Options
{
char* connection;
char* client_key_file;
char* server_key_file;
int verbose;
} options =
{
"ssl://localhost:18883",
NULL,
NULL,
1
};
void usage(void)
{
printf("SSL WANT_WRITE Handshake Test\n");
printf("Options:\n");
printf("\t--hostname <hostname> - SSL server hostname (default: ssl://localhost:18883)\n");
printf("\t--client_key <file> - Client certificate file\n");
printf("\t--server_key <file> - Server CA certificate file\n");
printf("\t--verbose - Enable verbose output\n");
printf("\t--help - This help\n");
exit(EXIT_FAILURE);
}
void getopts(int argc, char** argv)
{
int count = 1;
while (count < argc)
{
if (strcmp(argv[count], "--hostname") == 0)
{
if (++count < argc)
options.connection = argv[count];
else
usage();
}
else if (strcmp(argv[count], "--client_key") == 0)
{
if (++count < argc)
options.client_key_file = argv[count];
else
usage();
}
else if (strcmp(argv[count], "--server_key") == 0)
{
if (++count < argc)
options.server_key_file = argv[count];
else
usage();
}
else if (strcmp(argv[count], "--verbose") == 0)
{
options.verbose = 1;
}
else if (strcmp(argv[count], "--help") == 0)
{
usage();
}
else
{
printf("Unknown option: %s\n", argv[count]);
usage();
}
count++;
}
}
/* Logging */
#define LOGA_DEBUG 0
#define LOGA_INFO 1
#include <stdarg.h>
#include <time.h>
#include <sys/timeb.h>
void MyLog(int LOGA_level, char* format, ...)
{
static char msg_buf[256];
va_list args;
#if defined(_WIN32) || defined(_WINDOWS)
struct timeb ts;
#else
struct timeval ts;
#endif
struct tm timeinfo;
if (LOGA_level == LOGA_DEBUG && options.verbose == 0)
return;
#if defined(_WIN32) || defined(_WINDOWS)
ftime(&ts);
localtime_s(&timeinfo, &ts.time);
#else
gettimeofday(&ts, NULL);
localtime_r(&ts.tv_sec, &timeinfo);
#endif
strftime(msg_buf, 80, "%Y%m%d %H%M%S", &timeinfo);
#if defined(_WIN32) || defined(_WINDOWS)
sprintf(&msg_buf[strlen(msg_buf)], ".%.3hu ", ts.millitm);
#else
sprintf(&msg_buf[strlen(msg_buf)], ".%.3lu ", ts.tv_usec / 1000L);
#endif
va_start(args, format);
vsnprintf(&msg_buf[strlen(msg_buf)], sizeof(msg_buf) - strlen(msg_buf), format, args);
va_end(args);
printf("%s\n", msg_buf);
fflush(stdout);
}
/* Test tracking */
int tests = 0;
int failures = 0;
#if !defined(__func__)
#define __func__ __FUNCTION__
#endif
#define ASSERT(condition, message) \
do { \
tests++; \
if (!(condition)) { \
failures++; \
MyLog(1, "FAIL: %s (line %d): %s", __func__, __LINE__, message); \
return -1; \
} \
} while(0)
/**
* Test: Basic SSL connection
*
* Verifies that SSL connections complete successfully.
* This validates that the SSL_ERROR_WANT_WRITE/WANT_READ handling
* in SSLSocket_connect() works correctly.
*/
int test_ssl_connect_basic(void)
{
MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;
int rc;
MyLog(1, "Test: Basic SSL connection");
rc = MQTTClient_create(&client, options.connection, "test_ssl_want_write",
MQTTCLIENT_PERSISTENCE_NONE, NULL);
ASSERT(rc == MQTTCLIENT_SUCCESS, "MQTTClient_create failed");
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
if (options.server_key_file)
{
ssl_opts.trustStore = options.server_key_file;
conn_opts.ssl = &ssl_opts;
}
if (options.client_key_file)
{
ssl_opts.keyStore = options.client_key_file;
}
MyLog(1, "Connecting to %s", options.connection);
rc = MQTTClient_connect(client, &conn_opts);
if (rc != MQTTCLIENT_SUCCESS)
{
MyLog(1, "Connection failed: rc=%d", rc);
MyLog(1, "Note: This is expected if no SSL broker is running on %s", options.connection);
/* Not a test failure - just means no broker available */
}
else
{
MyLog(1, "Connection succeeded");
MQTTClient_disconnect(client, 1000);
}
MQTTClient_destroy(&client);
return 0;
}
/**
* Main test runner
*/
int main(int argc, char** argv)
{
int rc = 0;
getopts(argc, argv);
printf("SSL Connection Test\n");
printf("===================\n");
printf("Connection: %s\n", options.connection);
if (options.client_key_file)
printf("Client Key: %s\n", options.client_key_file);
if (options.server_key_file)
printf("Server Key: %s\n", options.server_key_file);
printf("\n");
rc = test_ssl_connect_basic();
printf("\nTest Results: %d tests, %d failures\n", tests, failures);
if (failures == 0)
printf("SUCCESS\n");
else
printf("FAILURE: %d test(s) failed\n", failures);
return (failures == 0) ? 0 : 1;
}

185
test/thread.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2025 IBM Corp.
* Copyright (c) 2009, 2026 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -226,159 +226,53 @@ void myassert(char* filename, int lineno, char* description, int value, char* fo
MyLog(LOGA_DEBUG, "Assertion succeeded, file %s, line %d, description: %s", filename, lineno, description);
}
static thread_return_type WINAPI sem_secondary(void* n)
thread_return_type WINAPI evt_secondary(void* n)
{
int rc = 0;
sem_type sem = n;
START_TIME_TYPE start;
long duration;
MyLog(LOGA_DEBUG, "Secondary semaphore pointer %p", sem);
rc = Thread_check_sem(sem);
assert("rc 0 from check_sem", rc == 0, "rc was %d", rc);
MyLog(LOGA_DEBUG, "Secondary thread about to wait");
start = start_clock();
rc = Thread_wait_sem(sem, 99999);
duration = elapsed(start);
assert("rc 0 from lock mutex", rc == 0, "rc was %d", rc);
MyLog(LOGA_INFO, "Lock duration was %ld", duration);
assert("duration is 2s", duration >= 1999L && duration < 2050L, "duration was %ld", duration);
MyLog(LOGA_DEBUG, "Secondary thread ending");
return 0;
}
int test_sem(struct Options options)
{
char* testname = "test_sem";
int rc = 0, i = 0;
START_TIME_TYPE start;
long duration;
sem_type sem = Thread_create_sem(&rc);
MyLog(LOGA_INFO, "Starting semaphore test");
fprintf(xml, "<testcase classname=\"test\" name=\"%s\"", testname);
global_start_time = start_clock();
MyLog(LOGA_DEBUG, "Primary semaphore pointer %p\n", sem);
/* The semaphore should be created non-signaled */
rc = Thread_check_sem(sem);
assert("rc 0 from check_sem", rc == 0, "rc was %d\n", rc);
MyLog(LOGA_DEBUG, "Post semaphore so then check should be 1\n");
rc = Thread_post_sem(sem);
assert("rc 0 from post_sem", rc == 0, "rc was %d\n", rc);
/* should be 1, and then reset to 0 */
rc = Thread_check_sem(sem);
assert("rc 1 from check_sem", rc == 1, "rc was %d", rc);
/* so now it'll be 0 */
rc = Thread_check_sem(sem);
assert("rc 0 from check_sem", rc == 0, "rc was %d", rc);
/* multiple posts */
for (i = 0; i < 10; ++i)
{
rc = Thread_post_sem(sem);
assert("rc 0 from post_sem", rc == 0, "rc was %d\n", rc);
}
rc = Thread_check_sem(sem);
assert("rc 1 from check_sem", rc == 1, "rc was %d", rc);
// Binary sem, so additional checks should be zero
for (i = 0; i < 10; ++i)
{
rc = Thread_check_sem(sem);
assert("rc 0 from check_sem", rc == 0, "rc was %d", rc);
}
rc = Thread_check_sem(sem);
assert("rc 0 from check_sem", rc == 0, "rc was %d", rc);
MyLog(LOGA_DEBUG, "Check timeout");
start = start_clock();
rc = Thread_wait_sem(sem, 1500);
duration = elapsed(start);
#if defined(EAGAIN)
assert("rc ETIMEDOUT from lock mutex", rc == ETIMEDOUT || rc == EAGAIN, "rc was %d", rc);
#else
assert("rc ETIMEDOUT from lock mutex", rc == ETIMEDOUT, "rc was %d", rc);
#endif
MyLog(LOGA_INFO, "Lock duration was %ld", duration);
assert("duration is 2s", duration >= 1500L, "duration was %ld", duration);
MyLog(LOGA_DEBUG, "Starting secondary thread");
Paho_thread_start(sem_secondary, (void*)sem);
mysleep(2);
MyLog(LOGA_DEBUG, "post secondary");
rc = Thread_post_sem(sem);
assert("rc 0 from post_sem", rc == 0, "rc was %d", rc);
mysleep(1);
MyLog(LOGA_DEBUG, "Main thread ending");
/*exit: */ MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.",
(failures == 0) ? "passed" : "failed", testname, tests, failures);
write_test_result();
return failures;
}
#if !defined(_WIN32)
thread_return_type cond_secondary(void* n)
{
int rc = 0;
cond_type cond = n;
evt_type evt = n;
START_TIME_TYPE start;
long duration;
MyLog(LOGA_DEBUG, "This will time out");
start = start_clock();
rc = Thread_wait_cond(cond, 1000);
rc = Thread_wait_evt(evt, 1000);
duration = elapsed(start);
MyLog(LOGA_INFO, "Lock duration was %ld", duration);
assert("duration is about 1s", duration >= 999L && duration <= 1050L, "duration was %ld", duration);
assert("rc non 0 from wait_cond", rc == ETIMEDOUT, "rc was %d", rc);
assert("rc non 0 from wait_evt", rc == ETIMEDOUT, "rc was %d", rc);
MyLog(LOGA_DEBUG, "This should hang around a few seconds");
start = start_clock();
rc = Thread_wait_cond(cond, 99999);
rc = Thread_wait_evt(evt, 99999);
duration = elapsed(start);
MyLog(LOGA_INFO, "Lock duration was %ld", duration);
assert("duration is around 1s", duration >= 990L && duration <= 1010L, "duration was %ld", duration);
assert("rc 9 from wait_cond", rc == 0, "rc was %d", rc);
assert("rc 9 from wait_evt", rc == 0, "rc was %d", rc);
MyLog(LOGA_DEBUG, "Secondary cond thread ending");
MyLog(LOGA_DEBUG, "Secondary evt thread ending");
return 0;
}
int test_cond(struct Options options)
int test_evt(struct Options options)
{
char* testname = "test_cond";
char* testname = "test_evt";
int rc = 0, i = 0;
START_TIME_TYPE start;
long duration;
cond_type cond = Thread_create_cond(&rc);
evt_type evt = Thread_create_evt(&rc);
MyLog(LOGA_INFO, "Starting condition variable test");
fprintf(xml, "<testcase classname=\"cond\" name=\"%s\"", testname);
MyLog(LOGA_INFO, "Starting event test");
fprintf(xml, "<testcase classname=\"evt\" name=\"%s\"", testname);
global_start_time = start_clock();
/* The semaphore should be created non-signaled */
rc = Thread_wait_cond(cond, 0);
assert("rc 0 from wait_cond", rc == ETIMEDOUT, "rc was %d", rc);
rc = Thread_wait_evt(evt, 0);
assert("rc 0 from wait_evt", rc == ETIMEDOUT, "rc was %d", rc);
MyLog(LOGA_DEBUG, "Check timeout");
start = start_clock();
rc = Thread_wait_cond(cond, 2000);
rc = Thread_wait_evt(evt, 2000);
duration = elapsed(start);
assert("rc ETIMEDOUT from lock mutex", rc == ETIMEDOUT, "rc was %d", rc);
MyLog(LOGA_INFO, "Lock duration was %ld", duration);
@ -387,42 +281,44 @@ int test_cond(struct Options options)
/* multiple posts */
for (i = 0; i < 10; ++i)
{
rc = Thread_signal_cond(cond);
assert("rc 0 from signal cond", rc == 0, "rc was %d\n", rc);
rc = Thread_signal_evt(evt);
assert("rc 0 from signal evt", rc == 0, "rc was %d\n", rc);
}
/* the signals are not stored */
for (i = 0; i < 10; ++i)
rc = Thread_wait_evt(evt, 0);
assert("rc zero from wait_evt", rc == 0, "rc was %d", rc);
/* the signals are not accumulated */
for (i = 0; i < 9; ++i)
{
rc = Thread_wait_cond(cond, 0);
assert("rc non-zero from wait_cond", rc == ETIMEDOUT, "rc was %d", rc);
rc = Thread_wait_evt(evt, 0);
assert("rc non-zero from wait_evt", rc == ETIMEDOUT, "rc was %d", rc);
}
MyLog(LOGA_DEBUG, "Post secondary but it will time out");
rc = Thread_signal_cond(cond);
assert("rc 0 from signal cond", rc == 0, "rc was %d", rc);
// MyLog(LOGA_DEBUG, "Post secondary but it will time out");
// rc = Thread_signal_evt(evt);
// assert("rc 0 from signal evt", rc == 0, "rc was %d", rc);
MyLog(LOGA_DEBUG, "Starting secondary thread");
Paho_thread_start(cond_secondary, (void*)cond);
Paho_thread_start(evt_secondary, (void*)evt);
MyLog(LOGA_DEBUG, "wait for secondary thread to enter second wait");
mysleep(2);
MyLog(LOGA_DEBUG, "post secondary");
rc = Thread_signal_cond(cond);
assert("rc 0 from signal cond", rc == 0, "rc was %d", rc);
rc = Thread_signal_evt(evt);
assert("rc 0 from signal evt", rc == 0, "rc was %d", rc);
mysleep(1);
MyLog(LOGA_DEBUG, "Main thread ending");
exit: MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.",
(failures == 0) ? "passed" : "failed", testname, tests, failures);
MyLog(LOGA_INFO, "%s: test %s. %d tests run, %d failures.",
(failures == 0) ? "passed" : "failed", testname, tests, failures);
write_test_result();
return failures;
}
#endif
static thread_return_type WINAPI mutex_secondary(void* n)
@ -458,6 +354,7 @@ int test_mutex(struct Options options)
MyLog(LOGA_INFO, "Starting mutex test");
fprintf(xml, "<testcase classname=\"test\" name=\"%s\"", testname);
global_start_time = start_clock();
assert("rc 0 from create mutex", rc == 0, "rc was %d", rc);
/* this should happen immediately, as there is no other lock held */
start = start_clock();
@ -497,12 +394,10 @@ int test_mutex(struct Options options)
int main(int argc, char** argv)
{
int rc = -1;
int (*tests[])(struct Options) = {NULL,
int (*tests[])(struct Options) = {
NULL,
test_mutex,
test_sem,
#if !defined(_WIN32)
test_cond
#endif
test_evt
}; /* indexed starting from 1 */
int i;
@ -532,11 +427,7 @@ int main(int argc, char** argv)
}
}
if (rc == 0)
MyLog(LOGA_INFO, "verdict pass");
else
MyLog(LOGA_INFO, "verdict fail");
MyLog(LOGA_INFO, (rc == 0) ? "verdict pass" : "verdict fail");
fprintf(xml, "</testsuite>\n");
fclose(xml);

2
version.patch
View File

@ -1 +1 @@
15
16