WolfSSL
/
wolfMQTT
mirror of https://github.com/wolfSSL/wolfMQTT.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Improvements to the connect and read code to handle runtime option for block/non-block. Enabled the mqttclient example support with `--enable-nonblock`.

pull/130/head
David Garske 2019-09-05 14:55:42 -07:00
parent 60ea40a1a7
commit 0f4ef8805e
4 changed files with 90 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
examples/mqttclient/mqttclient.c
View File

@ -470,6 +470,12 @@ int mqttclient_test(MQTTCtx *mqttCtx)
rc = MqttClient_WaitMessage(&mqttCtx->client,
mqttCtx->cmd_timeout_ms);
#ifdef WOLFMQTT_NONBLOCK
/* Track elapsed time with no activity and trigger timeout */
rc = mqtt_check_timeout(rc, &mqttCtx->start_sec,
mqttCtx->cmd_timeout_ms/1000);
#endif
/* check for test mode */
if (mStopRead) {
rc = MQTT_CODE_SUCCESS;
@ -605,7 +611,6 @@ exit:
int main(int argc, char** argv)
{
int rc;
#ifndef WOLFMQTT_NONBLOCK
MQTTCtx mqttCtx;
/* init defaults */
@ -617,7 +622,7 @@ int main(int argc, char** argv)
if (rc != 0) {
return rc;
}
#endif
#ifdef USE_WINDOWS_API
if (SetConsoleCtrlHandler((PHANDLER_ROUTINE)CtrlHandler,
TRUE) == FALSE)
@ -630,17 +635,7 @@ int main(int argc, char** argv)
}
#endif
#ifndef WOLFMQTT_NONBLOCK
rc = mqttclient_test(&mqttCtx);
#else
(void)argc;
(void)argv;
/* This example requires non-blocking mode to be disabled
./configure --disable-nonblock */
PRINTF("Example not compiled in!");
rc = EXIT_FAILURE;
#endif
return (rc == 0) ? 0 : EXIT_FAILURE;

3
examples/mqttexample.h
View File

@ -146,6 +146,9 @@ typedef struct _MQTTCtx {
byte subId_not_avail; /* Server property */
byte enable_eauth; /* Enhanced authentication */
#endif
#ifdef WOLFMQTT_NONBLOCK
unsigned int useNonBlockMode:1;
#endif
} MQTTCtx;

133
examples/mqttnet.c
View File

@ -627,28 +627,31 @@ static int NetConnect(void *context, const char* host, word16 port,
#endif /* !WOLFMQTT_NO_TIMEOUT */
#if !defined(WOLFMQTT_NO_TIMEOUT) && defined(WOLFMQTT_NONBLOCK)
/* Set socket as non-blocking */
tcp_set_nonblocking(&sock->fd);
if (mqttCtx->useNonBlockMode) {
/* Set socket as non-blocking */
tcp_set_nonblocking(&sock->fd);
}
#endif
/* Start connect */
rc = SOCK_CONNECT(sock->fd, (struct sockaddr*)&sock->addr, sizeof(sock->addr));
#ifndef WOLFMQTT_NO_TIMEOUT
/* Wait for connect */
if (rc < 0 || select((int)SELECT_FD(sock->fd), NULL, &fdset, NULL, &tv) > 0)
#else
if (rc < 0)
#endif /* !WOLFMQTT_NO_TIMEOUT */
{
if (rc < 0) {
/* Check for error */
socklen_t len = sizeof(so_error);
getsockopt(sock->fd, SOL_SOCKET, SO_ERROR, &so_error, &len);
if (so_error == 0) {
rc = 0; /* Success */
}
#if !defined(WOLFMQTT_NO_TIMEOUT) && defined(WOLFMQTT_NONBLOCK)
else if (so_error == EINPROGRESS) {
/* set default error case */
rc = MQTT_CODE_ERROR_NETWORK;
#ifdef WOLFMQTT_NONBLOCK
if (errno == EINPROGRESS || so_error == EINPROGRESS) {
#ifndef WOLFMQTT_NO_TIMEOUT
/* Wait for connect */
if (select((int)SELECT_FD(sock->fd), NULL, &fdset, NULL, &tv) > 0) {
rc = MQTT_CODE_SUCCESS;
}
#else
rc = MQTT_CODE_CONTINUE;
#endif
}
#endif
}
@ -816,10 +819,11 @@ static int NetRead_ex(void *context, byte* buf, int buf_len,
SOERROR_T so_error = 0;
int bytes = 0;
int flags = 0;
#if !defined(WOLFMQTT_NO_TIMEOUT) && !defined(WOLFMQTT_NONBLOCK)
#ifndef WOLFMQTT_NO_TIMEOUT
fd_set recvfds;
fd_set errfds;
struct timeval tv;
MQTTCtx* mqttCtx = sock->mqttCtx;
#endif
if (context == NULL || buf == NULL || buf_len <= 0) {
@ -833,7 +837,7 @@ static int NetRead_ex(void *context, byte* buf, int buf_len,
flags |= MSG_PEEK;
}
#if !defined(WOLFMQTT_NO_TIMEOUT) && !defined(WOLFMQTT_NONBLOCK)
#ifndef WOLFMQTT_NO_TIMEOUT
/* Setup timeout and FD's */
setup_timeout(&tv, timeout_ms);
FD_ZERO(&recvfds);
@ -851,50 +855,69 @@ static int NetRead_ex(void *context, byte* buf, int buf_len,
/* Loop until buf_len has been read, error or timeout */
while (bytes < buf_len) {
int do_read = 0;
#if !defined(WOLFMQTT_NO_TIMEOUT) && !defined(WOLFMQTT_NONBLOCK)
/* Wait for rx data to be available */
rc = select((int)SELECT_FD(sock->fd), &recvfds, NULL, &errfds, &tv);
if (rc > 0)
{
/* Check if rx or error */
if (FD_ISSET(sock->fd, &recvfds)) {
#endif /* !WOLFMQTT_NO_TIMEOUT && !WOLFMQTT_NONBLOCK */
/* Try and read number of buf_len provided,
minus what's already been read */
rc = (int)SOCK_RECV(sock->fd,
&buf[bytes],
buf_len - bytes,
flags);
if (rc <= 0) {
rc = -1;
goto exit; /* Error */
}
else {
bytes += rc; /* Data */
}
#if !defined(WOLFMQTT_NO_TIMEOUT) && !defined(WOLFMQTT_NONBLOCK)
}
#ifdef WOLFMQTT_ENABLE_STDIN_CAP
else if (FD_ISSET(STDIN, &recvfds)) {
return MQTT_CODE_STDIN_WAKE;
}
#endif
if (FD_ISSET(sock->fd, &errfds)) {
rc = -1;
break;
}
#ifndef WOLFMQTT_NO_TIMEOUT
#ifdef WOLFMQTT_NONBLOCK
if (mqttCtx->useNonBlockMode) {
do_read = 1;
}
else {
timeout = 1;
break; /* timeout or signal */
else
#endif
{
/* Wait for rx data to be available */
rc = select((int)SELECT_FD(sock->fd), &recvfds, NULL, &errfds, &tv);
if (rc > 0)
{
if (FD_ISSET(sock->fd, &recvfds))
{
do_read = 1;
}
/* Check if rx or error */
#ifdef WOLFMQTT_ENABLE_STDIN_CAP
else if (FD_ISSET(STDIN, &recvfds)) {
return MQTT_CODE_STDIN_WAKE;
}
#endif
if (FD_ISSET(sock->fd, &errfds)) {
rc = -1;
break;
}
}
else {
timeout = 1;
break; /* timeout or signal */
}
}
#else
/* non-blocking should always exit loop */
do_read = 1;
#endif /* !WOLFMQTT_NO_TIMEOUT */
if (do_read) {
/* Try and read number of buf_len provided,
minus what's already been read */
rc = (int)SOCK_RECV(sock->fd,
&buf[bytes],
buf_len - bytes,
flags);
if (rc <= 0) {
rc = -1;
goto exit; /* Error */
}
else {
bytes += rc; /* Data */
}
}
/* no timeout and non-block should always exit loop */
#ifdef WOLFMQTT_NONBLOCK
if (mqttCtx->useNonBlockMode) {
break;
}
#endif
#ifdef WOLFMQTT_NO_TIMEOUT
break;
#endif /* !WOLFMQTT_NO_TIMEOUT && !WOLFMQTT_NONBLOCK */
#endif
} /* while */
exit:

3
examples/nbclient/nbclient.c
View File

@ -113,6 +113,8 @@ int mqttclient_test(MQTTCtx *mqttCtx)
PRINTF("MQTT Client: QoS %d, Use TLS %d", mqttCtx->qos,
mqttCtx->use_tls);
mqttCtx->useNonBlockMode = 1;
FALL_THROUGH;
}
@ -535,4 +537,3 @@ exit:
}
#endif /* NO_MAIN_DRIVER */