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

Rename public Threading functions to avoid clashes #1313

This commit is contained in:
Ian Craggs 2023-10-10 12:20:56 +01:00
parent fa7b8b2d1d
commit afb5e871ba
12 changed files with 171 additions and 171 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
src/Heap.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2020 IBM Corp.
* Copyright (c) 2009, 2023 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -164,7 +164,7 @@ void* mymalloc(char* file, int line, size_t size)
size_t filenamelen = strlen(file)+1;
void* rc = NULL;
Thread_lock_mutex(heap_mutex);
Paho_thread_lock_mutex(heap_mutex);
size = Heap_roundup(size);
if ((s = malloc(sizeof(storageElement))) == NULL)
{
@ -194,7 +194,7 @@ void* mymalloc(char* file, int line, size_t size)
goto exit;
}
memset(s->stack, 0, sizeof(filenamelen));
StackTrace_get(Thread_getid(), s->stack, STACK_LEN);
StackTrace_get(Paho_thread_getid(), s->stack, STACK_LEN);
#endif
s->line = line;
/* Add space for eyecatcher at each end */
@ -216,7 +216,7 @@ void* mymalloc(char* file, int line, size_t size)
state.max_size = state.current_size;
rc = ((eyecatcherType*)(s->ptr)) + 1; /* skip start eyecatcher */
exit:
Thread_unlock_mutex(heap_mutex);
Paho_thread_unlock_mutex(heap_mutex);
return rc;
}
@ -280,10 +280,10 @@ void myfree(char* file, int line, void* p)
{
if (p) /* it is legal und usual to call free(NULL) */
{
Thread_lock_mutex(heap_mutex);
Paho_thread_lock_mutex(heap_mutex);
if (Internal_heap_unlink(file, line, p))
free(((eyecatcherType*)p)-1);
Thread_unlock_mutex(heap_mutex);
Paho_thread_unlock_mutex(heap_mutex);
}
else
{
@ -301,9 +301,9 @@ void myfree(char* file, int line, void* p)
*/
void Heap_unlink(char* file, int line, void* p)
{
Thread_lock_mutex(heap_mutex);
Paho_thread_lock_mutex(heap_mutex);
Internal_heap_unlink(file, line, p);
Thread_unlock_mutex(heap_mutex);
Paho_thread_unlock_mutex(heap_mutex);
}
@ -324,7 +324,7 @@ void *myrealloc(char* file, int line, void* p, size_t size)
void* rc = NULL;
storageElement* s = NULL;
Thread_lock_mutex(heap_mutex);
Paho_thread_lock_mutex(heap_mutex);
s = TreeRemoveKey(&heap, ((eyecatcherType*)p)-1);
if (s == NULL)
Log(LOG_ERROR, 13, "Failed to reallocate heap item at file %s line %d", file, line);
@ -356,7 +356,7 @@ void *myrealloc(char* file, int line, void* p, size_t size)
TreeAdd(&heap, s, space);
}
exit:
Thread_unlock_mutex(heap_mutex);
Paho_thread_unlock_mutex(heap_mutex);
return (rc == NULL) ? NULL : ((eyecatcherType*)(rc)) + 1; /* skip start eyecatcher */
}
@ -371,9 +371,9 @@ void* Heap_findItem(void* p)
{
Node* e = NULL;
Thread_lock_mutex(heap_mutex);
Paho_thread_lock_mutex(heap_mutex);
e = TreeFind(&heap, ((eyecatcherType*)p)-1);
Thread_unlock_mutex(heap_mutex);
Paho_thread_unlock_mutex(heap_mutex);
return (e == NULL) ? NULL : e->content;
}
@ -386,7 +386,7 @@ static void HeapScan(enum LOG_LEVELS log_level)
{
Node* current = NULL;
Thread_lock_mutex(heap_mutex);
Paho_thread_lock_mutex(heap_mutex);
Log(log_level, -1, "Heap scan start, total %d bytes", (int)state.current_size);
while ((current = TreeNextElement(&heap, current)) != NULL)
{
@ -398,7 +398,7 @@ static void HeapScan(enum LOG_LEVELS log_level)
#endif
}
Log(log_level, -1, "Heap scan end");
Thread_unlock_mutex(heap_mutex);
Paho_thread_unlock_mutex(heap_mutex);
}

10
src/Log.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2022 IBM Corp.
* Copyright (c) 2009, 2023 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -430,7 +430,7 @@ void Log(enum LOG_LEVELS log_level, int msgno, const char *format, ...)
va_list args;
/* we're using a static character buffer, so we need to make sure only one thread uses it at a time */
Thread_lock_mutex(log_mutex);
Paho_thread_lock_mutex(log_mutex);
if (format == NULL && (temp = Messages_get(msgno, log_level)) != NULL)
format = temp;
@ -439,7 +439,7 @@ void Log(enum LOG_LEVELS log_level, int msgno, const char *format, ...)
Log_trace(log_level, msg_buf);
va_end(args);
Thread_unlock_mutex(log_mutex);
Paho_thread_unlock_mutex(log_mutex);
}
}
@ -463,7 +463,7 @@ void Log_stackTrace(enum LOG_LEVELS log_level, int msgno, thread_id_type thread_
if (log_level < trace_settings.trace_level)
return;
Thread_lock_mutex(log_mutex);
Paho_thread_lock_mutex(log_mutex);
cur_entry = Log_pretrace();
memcpy(&(cur_entry->ts), &now_ts, sizeof(now_ts));
@ -483,7 +483,7 @@ void Log_stackTrace(enum LOG_LEVELS log_level, int msgno, thread_id_type thread_
}
Log_posttrace(log_level, cur_entry);
Thread_unlock_mutex(log_mutex);
Paho_thread_unlock_mutex(log_mutex);
}

8
src/MQTTAsync.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2022 IBM Corp., Ian Craggs and others
* Copyright (c) 2009, 2023 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
@ -648,7 +648,7 @@ int MQTTAsync_connect(MQTTAsync handle, const MQTTAsync_connectOptions* options)
m->connectTimeout = options->connectTimeout;
/* don't lock async mutex if we are being called from a callback */
thread_id = Thread_getid();
thread_id = Paho_thread_getid();
if (thread_id != sendThread_id && thread_id != receiveThread_id)
{
MQTTAsync_lock_mutex(mqttasync_mutex);
@ -658,12 +658,12 @@ int MQTTAsync_connect(MQTTAsync handle, const MQTTAsync_connectOptions* options)
if (sendThread_state != STARTING && sendThread_state != RUNNING)
{
sendThread_state = STARTING;
Thread_start(MQTTAsync_sendThread, NULL);
Paho_thread_start(MQTTAsync_sendThread, NULL);
}
if (receiveThread_state != STARTING && receiveThread_state != RUNNING)
{
receiveThread_state = STARTING;
Thread_start(MQTTAsync_receiveThread, handle);
Paho_thread_start(MQTTAsync_receiveThread, handle);
}
if (locked)
MQTTAsync_unlock_mutex(mqttasync_mutex);

10
src/MQTTAsyncUtils.c
View File

@ -165,7 +165,7 @@ static int clientSockCompare(void* a, void* b)
void MQTTAsync_lock_mutex(mutex_type amutex)
{
int rc = Thread_lock_mutex(amutex);
int rc = Paho_thread_lock_mutex(amutex);
if (rc != 0)
Log(LOG_ERROR, 0, "Error %s locking mutex", strerror(rc));
}
@ -173,7 +173,7 @@ void MQTTAsync_lock_mutex(mutex_type amutex)
void MQTTAsync_unlock_mutex(mutex_type amutex)
{
int rc = Thread_unlock_mutex(amutex);
int rc = Paho_thread_unlock_mutex(amutex);
if (rc != 0)
Log(LOG_ERROR, 0, "Error %s unlocking mutex", strerror(rc));
}
@ -1786,7 +1786,7 @@ thread_return_type WINAPI MQTTAsync_sendThread(void* n)
Thread_set_name("MQTTAsync_send");
MQTTAsync_lock_mutex(mqttasync_mutex);
sendThread_state = RUNNING;
sendThread_id = Thread_getid();
sendThread_id = Paho_thread_getid();
MQTTAsync_unlock_mutex(mqttasync_mutex);
while (!MQTTAsync_tostop)
{
@ -2015,7 +2015,7 @@ thread_return_type WINAPI MQTTAsync_receiveThread(void* n)
Thread_set_name("MQTTAsync_rcv");
MQTTAsync_lock_mutex(mqttasync_mutex);
receiveThread_state = RUNNING;
receiveThread_id = Thread_getid();
receiveThread_id = Paho_thread_getid();
while (!MQTTAsync_tostop)
{
int rc = SOCKET_ERROR;
@ -2735,7 +2735,7 @@ int MQTTAsync_assignMsgId(MQTTAsyncs* m)
/* need to check: commands list and response list for a client */
FUNC_ENTRY;
/* We might be called in a callback. In which case, this mutex will be already locked. */
thread_id = Thread_getid();
thread_id = Paho_thread_getid();
if (thread_id != sendThread_id && thread_id != receiveThread_id)
{
MQTTAsync_lock_mutex(mqttasync_mutex);

168
src/MQTTClient.c
View File

@ -373,7 +373,7 @@ int MQTTClient_createWithOptions(MQTTClient* handle, const char* serverURI, cons
BOOL bStatus = InitOnceExecuteOnce(&g_InitOnce, InitOnceFunction, NULL, NULL);
#endif
FUNC_ENTRY;
if ((rc = Thread_lock_mutex(mqttclient_mutex)) != 0)
if ((rc = Paho_thread_lock_mutex(mqttclient_mutex)) != 0)
goto nounlock_exit;
if (serverURI == NULL || clientId == NULL)
@ -517,7 +517,7 @@ int MQTTClient_createWithOptions(MQTTClient* handle, const char* serverURI, cons
ListAppend(bstate->clients, m->c, sizeof(Clients) + 3*sizeof(List));
exit:
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
nounlock_exit:
FUNC_EXIT_RC(rc);
return rc;
@ -578,8 +578,8 @@ void MQTTClient_destroy(MQTTClient* handle)
MQTTClients* m = *handle;
FUNC_ENTRY;
Thread_lock_mutex(connect_mutex);
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(connect_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (m == NULL)
goto exit;
@ -612,8 +612,8 @@ void MQTTClient_destroy(MQTTClient* handle)
MQTTClient_terminate();
exit:
Thread_unlock_mutex(mqttclient_mutex);
Thread_unlock_mutex(connect_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(connect_mutex);
FUNC_EXIT;
}
@ -721,7 +721,7 @@ int MQTTClient_setDisconnected(MQTTClient handle, void* context, MQTTClient_disc
MQTTClients* m = handle;
FUNC_ENTRY;
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (m == NULL || m->c->connect_state != NOT_IN_PROGRESS)
rc = MQTTCLIENT_FAILURE;
@ -731,7 +731,7 @@ int MQTTClient_setDisconnected(MQTTClient handle, void* context, MQTTClient_disc
m->disconnected = disconnected;
}
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
FUNC_EXIT_RC(rc);
return rc;
}
@ -762,7 +762,7 @@ int MQTTClient_setPublished(MQTTClient handle, void* context, MQTTClient_publish
MQTTClients* m = handle;
FUNC_ENTRY;
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (m == NULL || m->c->connect_state != NOT_IN_PROGRESS)
rc = MQTTCLIENT_FAILURE;
@ -772,7 +772,7 @@ int MQTTClient_setPublished(MQTTClient handle, void* context, MQTTClient_publish
m->published = published;
}
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
FUNC_EXIT_RC(rc);
return rc;
}
@ -785,7 +785,7 @@ int MQTTClient_setHandleAuth(MQTTClient handle, void* context, MQTTClient_handle
MQTTClients* m = handle;
FUNC_ENTRY;
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (m == NULL || m->c->connect_state != NOT_IN_PROGRESS)
rc = MQTTCLIENT_FAILURE;
@ -795,7 +795,7 @@ int MQTTClient_setHandleAuth(MQTTClient handle, void* context, MQTTClient_handle
m->auth_handle = auth_handle;
}
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
FUNC_EXIT_RC(rc);
return rc;
}
@ -827,9 +827,9 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
FUNC_ENTRY;
Thread_set_name("MQTTClient_run");
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
run_id = Thread_getid();
run_id = Paho_thread_getid();
running = 1;
while (!tostop)
{
@ -838,9 +838,9 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
MQTTClients* m = NULL;
MQTTPacket* pack = NULL;
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
pack = MQTTClient_cycle(&sock, timeout, &rc);
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (tostop)
break;
timeout = 100L;
@ -889,9 +889,9 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
Log(TRACE_MIN, -1, "Calling messageArrived for client %s, queue depth %d",
m->c->clientID, m->c->messageQueue->count);
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
rc = (*(m->ma))(m->context, qe->topicName, topicLen, qe->msg);
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
/* if 0 (false) is returned by the callback then it failed, so we don't remove the message from
* the queue, and it will be retried later. If 1 is returned then the message data may have been freed,
* so we must be careful how we use it.
@ -946,7 +946,7 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
*(dp->properties) = disc->properties;
MQTTClient_disconnect1(m, 10, 0, 1, MQTTREASONCODE_SUCCESS, NULL);
Log(TRACE_MIN, -1, "Calling disconnected for client %s", m->c->clientID);
Thread_start(call_disconnected, dp);
Paho_thread_start(call_disconnected, dp);
}
else
free(dp);
@ -964,7 +964,7 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
dp.reasonCode = disc->rc;
free(pack);
Log(TRACE_MIN, -1, "Calling auth_handle for client %s", m->c->clientID);
Thread_start(call_auth_handle, &dp);
Paho_thread_start(call_auth_handle, &dp);
}
#endif
}
@ -1012,7 +1012,7 @@ static thread_return_type WINAPI MQTTClient_run(void* n)
}
run_id = 0;
running = tostop = 0;
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
FUNC_EXIT;
#if defined(_WIN32) || defined(_WIN64)
ExitThread(0);
@ -1047,14 +1047,14 @@ static int MQTTClient_stop(void)
{
int count = 0;
tostop = 1;
if (Thread_getid() != run_id)
if (Paho_thread_getid() != run_id)
{
while (running && ++count < 100)
{
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
Log(TRACE_MIN, -1, "sleeping");
MQTTTime_sleep(100L);
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
}
}
rc = 1;
@ -1072,7 +1072,7 @@ int MQTTClient_setCallbacks(MQTTClient handle, void* context, MQTTClient_connect
MQTTClients* m = handle;
FUNC_ENTRY;
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (m == NULL || ma == NULL || m->c->connect_state != NOT_IN_PROGRESS)
rc = MQTTCLIENT_FAILURE;
@ -1084,7 +1084,7 @@ int MQTTClient_setCallbacks(MQTTClient handle, void* context, MQTTClient_connect
m->dc = dc;
}
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
FUNC_EXIT_RC(rc);
return rc;
}
@ -1100,7 +1100,7 @@ static void MQTTClient_closeSession(Clients* client, enum MQTTReasonCodes reason
{
if (client->connected)
MQTTPacket_send_disconnect(client, reason, props);
Thread_lock_mutex(socket_mutex);
Paho_thread_lock_mutex(socket_mutex);
WebSocket_close(&client->net, WebSocket_CLOSE_NORMAL, NULL);
#if defined(OPENSSL)
@ -1108,7 +1108,7 @@ static void MQTTClient_closeSession(Clients* client, enum MQTTReasonCodes reason
client->session = NULL; /* show the session has been freed */
SSLSocket_close(&client->net);
#endif
Thread_unlock_mutex(socket_mutex);
Paho_thread_unlock_mutex(socket_mutex);
Socket_close(client->net.socket);
client->net.socket = 0;
#if defined(OPENSSL)
@ -1212,7 +1212,7 @@ static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_c
{
int count = 0;
Thread_start(MQTTClient_run, handle);
Paho_thread_start(MQTTClient_run, handle);
if (MQTTTime_elapsed(start) >= millisecsTimeout)
{
rc = SOCKET_ERROR;
@ -1221,9 +1221,9 @@ static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_c
while (!running && ++count < 5)
{
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
MQTTTime_sleep(100L);
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
}
if (!running)
{
@ -1259,9 +1259,9 @@ static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_c
if (m->c->connect_state == TCP_IN_PROGRESS) /* TCP connect started - wait for completion */
{
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
MQTTClient_waitfor(handle, CONNECT, &rc, millisecsTimeout - MQTTTime_elapsed(start));
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (rc != 0)
{
rc = SOCKET_ERROR;
@ -1364,9 +1364,9 @@ static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_c
#if defined(OPENSSL)
if (m->c->connect_state == SSL_IN_PROGRESS) /* SSL connect sent - wait for completion */
{
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
MQTTClient_waitfor(handle, CONNECT, &rc, millisecsTimeout - MQTTTime_elapsed(start));
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (rc != 1)
{
rc = SOCKET_ERROR;
@ -1400,9 +1400,9 @@ static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_c
if (m->c->connect_state == WEBSOCKET_IN_PROGRESS) /* websocket request sent - wait for upgrade */
{
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
MQTTClient_waitfor(handle, CONNECT, &rc, millisecsTimeout - MQTTTime_elapsed(start));
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
m->c->connect_state = WAIT_FOR_CONNACK; /* websocket upgrade complete */
if (MQTTPacket_send_connect(m->c, MQTTVersion, connectProperties, willProperties) == SOCKET_ERROR)
{
@ -1414,9 +1414,9 @@ static MQTTResponse MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_c
if (m->c->connect_state == WAIT_FOR_CONNACK) /* MQTT connect sent - wait for CONNACK */
{
MQTTPacket* pack = NULL;
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
pack = MQTTClient_waitfor(handle, CONNACK, &rc, millisecsTimeout - MQTTTime_elapsed(start));
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (pack == NULL)
rc = SOCKET_ERROR;
else
@ -1743,8 +1743,8 @@ MQTTResponse MQTTClient_connectAll(MQTTClient handle, MQTTClient_connectOptions*
MQTTResponse rc = MQTTResponse_initializer;
FUNC_ENTRY;
Thread_lock_mutex(connect_mutex);
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(connect_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
rc.reasonCode = SOCKET_ERROR;
if (!library_initialized)
@ -1905,8 +1905,8 @@ exit:
free(m->c->will);
m->c->will = NULL;
}
Thread_unlock_mutex(mqttclient_mutex);
Thread_unlock_mutex(connect_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(connect_mutex);
FUNC_EXIT_RC(rc.reasonCode);
return rc;
}
@ -1941,9 +1941,9 @@ static int MQTTClient_disconnect1(MQTTClient handle, int timeout, int call_conne
{ /* wait for all inflight message flows to finish, up to timeout */
if (MQTTTime_elapsed(start) >= (ELAPSED_TIME_TYPE)timeout)
break;
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
MQTTClient_yield();
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
}
}
@ -1956,7 +1956,7 @@ exit:
{
sync.sem = Thread_create_sem(&rc);
Log(TRACE_MIN, -1, "Calling connectionLost for client %s", m->c->clientID);
Thread_start(connectionLost_call, &sync);
Paho_thread_start(connectionLost_call, &sync);
Thread_wait_sem(sync.sem, 5000);
Thread_destroy_sem(sync.sem);
}
@ -1987,9 +1987,9 @@ int MQTTClient_disconnect(MQTTClient handle, int timeout)
{
int rc = 0;
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
rc = MQTTClient_disconnect1(handle, timeout, 0, 1, MQTTREASONCODE_SUCCESS, NULL);
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
return rc;
}
@ -1998,9 +1998,9 @@ int MQTTClient_disconnect5(MQTTClient handle, int timeout, enum MQTTReasonCodes
{
int rc = 0;
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
rc = MQTTClient_disconnect1(handle, timeout, 0, 1, reason, props);
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
return rc;
}
@ -2011,10 +2011,10 @@ int MQTTClient_isConnected(MQTTClient handle)
int rc = 0;
FUNC_ENTRY;
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (m && m->c)
rc = m->c->connected;
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
FUNC_EXIT_RC(rc);
return rc;
}
@ -2032,8 +2032,8 @@ MQTTResponse MQTTClient_subscribeMany5(MQTTClient handle, int count, char* const
int msgid = 0;
FUNC_ENTRY;
Thread_lock_mutex(subscribe_mutex);
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(subscribe_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
resp.reasonCode = MQTTCLIENT_FAILURE;
if (m == NULL || m->c == NULL)
@ -2082,9 +2082,9 @@ MQTTResponse MQTTClient_subscribeMany5(MQTTClient handle, int count, char* const
{
MQTTPacket* pack = NULL;
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
pack = MQTTClient_waitfor(handle, SUBACK, &rc, m->commandTimeout);
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (pack != NULL)
{
Suback* sub = (Suback*)pack;
@ -2149,8 +2149,8 @@ MQTTResponse MQTTClient_subscribeMany5(MQTTClient handle, int count, char* const
exit:
if (rc < 0)
resp.reasonCode = rc;
Thread_unlock_mutex(mqttclient_mutex);
Thread_unlock_mutex(subscribe_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(subscribe_mutex);
FUNC_EXIT_RC(resp.reasonCode);
return resp;
}
@ -2210,8 +2210,8 @@ MQTTResponse MQTTClient_unsubscribeMany5(MQTTClient handle, int count, char* con
int msgid = 0;
FUNC_ENTRY;
Thread_lock_mutex(subscribe_mutex);
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(subscribe_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
resp.reasonCode = MQTTCLIENT_FAILURE;
if (m == NULL || m->c == NULL)
@ -2248,9 +2248,9 @@ MQTTResponse MQTTClient_unsubscribeMany5(MQTTClient handle, int count, char* con
{
MQTTPacket* pack = NULL;
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
pack = MQTTClient_waitfor(handle, UNSUBACK, &rc, m->commandTimeout);
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (pack != NULL)
{
Unsuback* unsub = (Unsuback*)pack;
@ -2297,8 +2297,8 @@ MQTTResponse MQTTClient_unsubscribeMany5(MQTTClient handle, int count, char* con
exit:
if (rc < 0)
resp.reasonCode = rc;
Thread_unlock_mutex(mqttclient_mutex);
Thread_unlock_mutex(subscribe_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(subscribe_mutex);
FUNC_EXIT_RC(resp.reasonCode);
return resp;
}
@ -2347,7 +2347,7 @@ MQTTResponse MQTTClient_publish5(MQTTClient handle, const char* topicName, int p
MQTTResponse resp = MQTTResponse_initializer;
FUNC_ENTRY;
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (m == NULL || m->c == NULL)
rc = MQTTCLIENT_FAILURE;
@ -2368,9 +2368,9 @@ MQTTResponse MQTTClient_publish5(MQTTClient handle, const char* topicName, int p
blocked = 1;
Log(TRACE_MIN, -1, "Blocking publish on queue full for client %s", m->c->clientID);
}
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
MQTTClient_yield();
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (m->c->connected == 0)
{
rc = MQTTCLIENT_FAILURE;
@ -2433,16 +2433,16 @@ MQTTResponse MQTTClient_publish5(MQTTClient handle, const char* topicName, int p
{
pending_writes* writing = NULL;
Thread_lock_mutex(socket_mutex);
Paho_thread_lock_mutex(socket_mutex);
writing = SocketBuffer_getWrite(m->c->net.socket);
Thread_unlock_mutex(socket_mutex);
Paho_thread_unlock_mutex(socket_mutex);
if (writing == NULL)
break;
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
MQTTClient_yield();
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
}
rc = (qos > 0 || m->c->connected == 1) ? MQTTCLIENT_SUCCESS : MQTTCLIENT_FAILURE;
}
@ -2468,7 +2468,7 @@ exit_and_free:
}
exit:
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
resp.reasonCode = rc;
FUNC_EXIT_RC(resp.reasonCode);
return resp;
@ -2577,7 +2577,7 @@ static MQTTPacket* MQTTClient_cycle(SOCKET* sock, ELAPSED_TIME_TYPE timeout, int
#if defined(OPENSSL)
}
#endif
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (*sock > 0 && rc1 == 0)
{
MQTTClients* m = NULL;
@ -2646,7 +2646,7 @@ static MQTTPacket* MQTTClient_cycle(SOCKET* sock, ELAPSED_TIME_TYPE timeout, int
}
}
MQTTClient_retry();
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
FUNC_EXIT_RC(*rc);
return pack;
}
@ -2666,9 +2666,9 @@ static MQTTPacket* MQTTClient_waitfor(MQTTClient handle, int packet_type, int* r
goto exit;
}
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
is_running = running;
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
if (is_running)
{
@ -2841,14 +2841,14 @@ void MQTTClient_yield(void)
{
SOCKET sock = -1;
MQTTClient_cycle(&sock, (timeout > elapsed) ? timeout - elapsed : 0L, &rc);
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (rc == SOCKET_ERROR && ListFindItem(handles, &sock, clientSockCompare))
{
MQTTClients* m = (MQTTClient)(handles->current->content);
if (m->c->connect_state != DISCONNECTING)
MQTTClient_disconnect_internal(m, 0);
}
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
elapsed = MQTTTime_elapsed(start);
}
while (elapsed < timeout);
@ -2872,7 +2872,7 @@ int MQTTClient_waitForCompletion(MQTTClient handle, MQTTClient_deliveryToken mdt
MQTTClients* m = handle;
FUNC_ENTRY;
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
elapsed = MQTTTime_elapsed(start);
while (elapsed < timeout)
@ -2892,14 +2892,14 @@ int MQTTClient_waitForCompletion(MQTTClient handle, MQTTClient_deliveryToken mdt
rc = MQTTCLIENT_SUCCESS; /* well we couldn't find it */
goto exit;
}
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
MQTTClient_yield();
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
elapsed = MQTTTime_elapsed(start);
}
exit:
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
FUNC_EXIT_RC(rc);
return rc;
}
@ -2912,7 +2912,7 @@ int MQTTClient_getPendingDeliveryTokens(MQTTClient handle, MQTTClient_deliveryTo
*tokens = NULL;
FUNC_ENTRY;
Thread_lock_mutex(mqttclient_mutex);
Paho_thread_lock_mutex(mqttclient_mutex);
if (m == NULL)
{
@ -2940,7 +2940,7 @@ int MQTTClient_getPendingDeliveryTokens(MQTTClient handle, MQTTClient_deliveryTo
}
exit:
Thread_unlock_mutex(mqttclient_mutex);
Paho_thread_unlock_mutex(mqttclient_mutex);
FUNC_EXIT_RC(rc);
return rc;
}

32
src/Socket.c
View File

@ -270,7 +270,7 @@ int Socket_addSocket(SOCKET newSd)
int rc = 0;
FUNC_ENTRY;
Thread_lock_mutex(socket_mutex);
Paho_thread_lock_mutex(socket_mutex);
mod_s.nfds++;
if (mod_s.fds_read)
mod_s.fds_read = realloc(mod_s.fds_read, mod_s.nfds * sizeof(mod_s.fds_read[0]));
@ -310,7 +310,7 @@ int Socket_addSocket(SOCKET newSd)
Log(LOG_ERROR, -1, "addSocket: setnonblocking");
exit:
Thread_unlock_mutex(socket_mutex);
Paho_thread_unlock_mutex(socket_mutex);
FUNC_EXIT_RC(rc);
return rc;
}
@ -384,7 +384,7 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
int timeout_ms = 1000;
FUNC_ENTRY;
Thread_lock_mutex(mutex);
Paho_thread_lock_mutex(mutex);
if (mod_s.clientsds->count == 0)
goto exit;
@ -423,9 +423,9 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
goto exit; /* no work to do */
}
/* Prevent performance issue by unlocking the socket_mutex while waiting for a ready socket. */
Thread_unlock_mutex(mutex);
Paho_thread_unlock_mutex(mutex);
*rc = select(maxfdp1_saved, &(mod_s.rset), &pwset, NULL, &timeout_tv);
Thread_lock_mutex(mutex);
Paho_thread_lock_mutex(mutex);
if (*rc == SOCKET_ERROR)
{
Socket_error("read select", 0);
@ -473,7 +473,7 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
ListNextElement(mod_s.clientsds, &mod_s.cur_clientsds);
}
exit:
Thread_unlock_mutex(mutex);
Paho_thread_unlock_mutex(mutex);
FUNC_EXIT_RC(sock);
return sock;
} /* end getReadySocket */
@ -493,7 +493,7 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
int timeout_ms = 1000;
FUNC_ENTRY;
Thread_lock_mutex(mutex);
Paho_thread_lock_mutex(mutex);
if (mod_s.nfds == 0 && mod_s.saved.nfds == 0)
goto exit;
@ -566,9 +566,9 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
}
/* Prevent performance issue by unlocking the socket_mutex while waiting for a ready socket. */
Thread_unlock_mutex(mutex);
Paho_thread_unlock_mutex(mutex);
*rc = poll(mod_s.saved.fds_read, mod_s.saved.nfds, timeout_ms);
Thread_lock_mutex(mutex);
Paho_thread_lock_mutex(mutex);
if (*rc == SOCKET_ERROR)
{
Socket_error("poll", 0);
@ -600,7 +600,7 @@ SOCKET Socket_getReadySocket(int more_work, int timeout, mutex_type mutex, int*
mod_s.saved.cur_fd = (mod_s.saved.cur_fd == mod_s.saved.nfds - 1) ? -1 : mod_s.saved.cur_fd + 1;
}
exit:
Thread_unlock_mutex(mutex);
Paho_thread_unlock_mutex(mutex);
FUNC_EXIT_RC(sock);
return sock;
} /* end getReadySocket */
@ -965,7 +965,7 @@ int Socket_close(SOCKET socket)
int rc = 0;
FUNC_ENTRY;
Thread_lock_mutex(socket_mutex);
Paho_thread_lock_mutex(socket_mutex);
Socket_close_only(socket);
Socket_abortWrite(socket);
SocketBuffer_cleanup(socket);
@ -1033,7 +1033,7 @@ int Socket_close(SOCKET socket)
else
Log(LOG_ERROR, -1, "Failed to remove socket %d", socket);
exit:
Thread_unlock_mutex(socket_mutex);
Paho_thread_unlock_mutex(socket_mutex);
FUNC_EXIT_RC(rc);
return rc;
}
@ -1206,9 +1206,9 @@ int Socket_new(const char* addr, size_t addr_len, int port, SOCKET* sock)
goto exit;
}
*pnewSd = *sock;
Thread_lock_mutex(socket_mutex);
Paho_thread_lock_mutex(socket_mutex);
result = ListAppend(mod_s.connect_pending, pnewSd, sizeof(SOCKET));
Thread_unlock_mutex(socket_mutex);
Paho_thread_unlock_mutex(socket_mutex);
if (!result)
{
free(pnewSd);
@ -1433,9 +1433,9 @@ int Socket_continueWrites(SOCKET* sock, mutex_type mutex)
if (writecomplete)
{
Thread_unlock_mutex(mutex);
Paho_thread_unlock_mutex(mutex);
(*writecomplete)(socket, rc);
Thread_lock_mutex(mutex);
Paho_thread_lock_mutex(mutex);
}
}
else

12
src/StackTrace.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2022 IBM Corp.
* Copyright (c) 2009, 2023 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -80,7 +80,7 @@ int setStack(int create);
int setStack(int create)
{
int i = -1;
thread_id_type curid = Thread_getid();
thread_id_type curid = Paho_thread_getid();
my_thread = NULL;
for (i = 0; i < MAX_THREADS && i < thread_count; ++i)
@ -105,7 +105,7 @@ int setStack(int create)
void StackTrace_entry(const char* name, int line, enum LOG_LEVELS trace_level)
{
Thread_lock_mutex(stack_mutex);
Paho_thread_lock_mutex(stack_mutex);
if (!setStack(1))
goto exit;
if (trace_level != -1)
@ -117,13 +117,13 @@ void StackTrace_entry(const char* name, int line, enum LOG_LEVELS trace_level)
if (my_thread->current_depth >= MAX_STACK_DEPTH)
Log(LOG_FATAL, -1, "Max stack depth exceeded");
exit:
Thread_unlock_mutex(stack_mutex);
Paho_thread_unlock_mutex(stack_mutex);
}
void StackTrace_exit(const char* name, int line, void* rc, enum LOG_LEVELS trace_level)
{
Thread_lock_mutex(stack_mutex);
Paho_thread_lock_mutex(stack_mutex);
if (!setStack(0))
goto exit;
if (--(my_thread->current_depth) < 0)
@ -138,7 +138,7 @@ void StackTrace_exit(const char* name, int line, void* rc, enum LOG_LEVELS trace
Log_stackTrace(trace_level, 11, my_thread->id, my_thread->current_depth, name, line, (int*)rc);
}
exit:
Thread_unlock_mutex(stack_mutex);
Paho_thread_unlock_mutex(stack_mutex);
}

16
src/Thread.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2022 IBM Corp. and Ian Craggs
* Copyright (c) 2009, 2023 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
@ -56,7 +56,7 @@
* @param fn the function to run, must be of the correct signature
* @param parameter pointer to the function parameter, can be NULL
*/
void Thread_start(thread_fn fn, void* parameter)
void Paho_thread_start(thread_fn fn, void* parameter)
{
#if defined(_WIN32) || defined(_WIN64)
thread_type thread = NULL;
@ -104,7 +104,7 @@ int Thread_set_name(const char* thread_name)
#else
#if defined(__GNUC__) && defined(__linux__)
#if __GLIBC__ >= 2 && __GLIBC_MINOR__ >= 12
rc = pthread_setname_np(Thread_getid(), thread_name);
rc = pthread_setname_np(Paho_thread_getid(), thread_name);
#endif
#endif
#endif
@ -118,7 +118,7 @@ int Thread_set_name(const char* thread_name)
* @param rc return code: 0 for success, negative otherwise
* @return the new mutex
*/
mutex_type Thread_create_mutex(int* rc)
mutex_type Paho_thread_create_mutex(int* rc)
{
mutex_type mutex = NULL;
@ -141,7 +141,7 @@ mutex_type Thread_create_mutex(int* rc)
* Lock a mutex which has alrea
* @return completion code, 0 is success
*/
int Thread_lock_mutex(mutex_type mutex)
int Paho_thread_lock_mutex(mutex_type mutex)
{
int rc = -1;
@ -162,7 +162,7 @@ int Thread_lock_mutex(mutex_type mutex)
* @param mutex the mutex
* @return completion code, 0 is success
*/
int Thread_unlock_mutex(mutex_type mutex)
int Paho_thread_unlock_mutex(mutex_type mutex)
{
int rc = -1;
@ -185,7 +185,7 @@ int Thread_unlock_mutex(mutex_type mutex)
* Destroy a mutex which has already been created
* @param mutex the mutex
*/
int Thread_destroy_mutex(mutex_type mutex)
int Paho_thread_destroy_mutex(mutex_type mutex)
{
int rc = 0;
@ -205,7 +205,7 @@ int 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 Thread_getid(void)
thread_id_type Paho_thread_getid(void)
{
#if defined(_WIN32) || defined(_WIN64)
return GetCurrentThreadId();

14
src/Thread.h
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2022 IBM Corp.
* Copyright (c) 2009, 2023 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -68,15 +68,15 @@
int Thread_destroy_cond(cond_type);
#endif
LIBMQTT_API void Thread_start(thread_fn, void*);
LIBMQTT_API void Paho_thread_start(thread_fn, void*);
int Thread_set_name(const char* thread_name);
LIBMQTT_API mutex_type Thread_create_mutex(int*);
LIBMQTT_API int Thread_lock_mutex(mutex_type);
LIBMQTT_API int Thread_unlock_mutex(mutex_type);
int Thread_destroy_mutex(mutex_type);
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 Thread_getid();
LIBMQTT_API thread_id_type Paho_thread_getid();
sem_type Thread_create_sem(int*);
int Thread_wait_sem(sem_type sem, int timeout);

14
test/test2.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2022 IBM Corp., Ian Craggs and others
* Copyright (c) 2009, 2023 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
@ -288,14 +288,14 @@ mutex_type deliveryCompleted_mutex = &deliveryCompleted_mutex_store;
void lock_mutex(mutex_type amutex)
{
int rc = Thread_lock_mutex(amutex);
int rc = Paho_thread_lock_mutex(amutex);
if (rc != 0)
MyLog(LOGA_INFO, "Error %s locking mutex", strerror(rc));
}
void unlock_mutex(mutex_type amutex)
{
int rc = Thread_unlock_mutex(amutex);
int rc = Paho_thread_unlock_mutex(amutex);
if (rc != 0)
MyLog(LOGA_INFO, "Error %s unlocking mutex", strerror(rc));
}
@ -365,7 +365,7 @@ thread_return_type WINAPI test1_sendAndReceive(void* n)
rc = MQTTClient_subscribe(c, test_topic, subsqos);
assert("Good rc from subscribe", rc == MQTTCLIENT_SUCCESS, "rc was %d", rc);
MyLog(LOGA_INFO, "Thread %u, %d messages at QoS %d", Thread_getid(), iterations, qos);
MyLog(LOGA_INFO, "Thread %u, %d messages at QoS %d", Paho_thread_getid(), iterations, qos);
test1_pubmsg.payload = test1_pubmsg_check.payload;
test1_pubmsg.payloadlen = test1_pubmsg_check.payloadlen;
test1_pubmsg.retained = 0;
@ -462,13 +462,13 @@ int test1(struct Options options)
test1_deliveryCompleted = test1_arrivedcount = 0;
struct thread_parms parms0 = {c, 0, test_topic};
Thread_start(test1_sendAndReceive, (void*)&parms0);
Paho_thread_start(test1_sendAndReceive, (void*)&parms0);
struct thread_parms parms1 = {c, 1, test_topic};
Thread_start(test1_sendAndReceive, (void*)&parms1);
Paho_thread_start(test1_sendAndReceive, (void*)&parms1);
struct thread_parms parms2 = {c, 2, test_topic};
Thread_start(test1_sendAndReceive, (void*)&parms2);
Paho_thread_start(test1_sendAndReceive, (void*)&parms2);
/* MQTT servers can send a message to a subscriber before the server has
completed the QoS 2 handshake with the publisher. For QoS 1 and 2,

8
test/test_connect_destroy.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2020 IBM Corp. and others
* Copyright (c) 2009, 2023 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
@ -285,14 +285,14 @@ void myassert(char* filename, int lineno, char* description, int value, char* fo
void lock_mutex(mutex_type amutex)
{
int rc = Thread_lock_mutex(amutex);
int rc = Paho_thread_lock_mutex(amutex);
if (rc != 0)
MyLog(LOGA_INFO, "Error %s locking mutex", strerror(rc));
}
void unlock_mutex(mutex_type amutex)
{
int rc = Thread_unlock_mutex(amutex);
int rc = Paho_thread_unlock_mutex(amutex);
if (rc != 0)
MyLog(LOGA_INFO, "Error %s unlocking mutex", strerror(rc));
}
@ -365,7 +365,7 @@ int test1(struct Options options)
opts.will = NULL;
struct thread_parms parms = {&c};
Thread_start(test1_destroy, (void*)&parms);
Paho_thread_start(test1_destroy, (void*)&parms);
MQTTClient_connect(c, &opts);

22
test/thread.c
View File

@ -1,5 +1,5 @@
/*******************************************************************************
* Copyright (c) 2009, 2020 IBM Corp.
* Copyright (c) 2009, 2023 IBM Corp.
*
* All rights reserved. This program and the accompanying materials
* are made available under the terms of the Eclipse Public License v2.0
@ -305,7 +305,7 @@ int test_sem(struct Options options)
assert("duration is 2s", duration >= 1500L, "duration was %ld", duration);
MyLog(LOGA_DEBUG, "Starting secondary thread");
Thread_start(sem_secondary, (void*)sem);
Paho_thread_start(sem_secondary, (void*)sem);
mysleep(2);
MyLog(LOGA_DEBUG, "post secondary");
@ -397,7 +397,7 @@ int test_cond(struct Options options)
assert("rc 0 from signal cond", rc == 0, "rc was %d", rc);
MyLog(LOGA_DEBUG, "Starting secondary thread");
Thread_start(cond_secondary, (void*)cond);
Paho_thread_start(cond_secondary, (void*)cond);
MyLog(LOGA_DEBUG, "wait for secondary thread to enter second wait");
mysleep(2);
@ -428,13 +428,13 @@ static thread_return_type WINAPI mutex_secondary(void* n)
/* this should take 2s, as there is another lock held */
start = start_clock();
rc = Thread_lock_mutex(mutex);
rc = Paho_thread_lock_mutex(mutex);
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 >= 1000L, "duration was %ld", duration);
rc = Thread_unlock_mutex(mutex);
rc = Paho_thread_unlock_mutex(mutex);
assert("rc 0 from unlock mutex", rc == 0, "rc was %d", rc);
MyLog(LOGA_DEBUG, "Secondary thread ending");
return 0;
@ -445,7 +445,7 @@ int test_mutex(struct Options options)
{
char* testname = "test_mutex";
int rc = 0;
mutex_type mutex = Thread_create_mutex(&rc);
mutex_type mutex = Paho_thread_create_mutex(&rc);
START_TIME_TYPE start;
long duration;
@ -455,27 +455,27 @@ int test_mutex(struct Options options)
/* this should happen immediately, as there is no other lock held */
start = start_clock();
rc = Thread_lock_mutex(mutex);
rc = Paho_thread_lock_mutex(mutex);
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 very low", duration < 5L, "duration was %ld", duration);
MyLog(LOGA_DEBUG, "Starting secondary thread");
Thread_start(mutex_secondary, (void*)mutex);
Paho_thread_start(mutex_secondary, (void*)mutex);
mysleep(2);
rc = Thread_unlock_mutex(mutex); /* let background thread have it */
rc = Paho_thread_unlock_mutex(mutex); /* let background thread have it */
assert("rc 0 from unlock mutex", rc == 0, "rc was %d", rc);
start = start_clock();
rc = Thread_lock_mutex(mutex); /* make sure background thread hasn't locked it */
rc = Paho_thread_lock_mutex(mutex); /* make sure background thread hasn't locked it */
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 very low", duration < 5L, "duration was %ld", duration);
Thread_destroy_mutex(mutex);
Paho_thread_destroy_mutex(mutex);
MyLog(LOGA_DEBUG, "Main thread ending");