Commit eab1a71d authored by Ian Craggs's avatar Ian Craggs

Keepalive timer fix

Bug: 442683
parent 2c969cfa
...@@ -149,7 +149,8 @@ BE*/ ...@@ -149,7 +149,8 @@ BE*/
typedef struct typedef struct
{ {
int socket; int socket;
time_t lastContact; time_t lastSent;
time_t lastReceived;
#if defined(OPENSSL) #if defined(OPENSSL)
SSL* ssl; SSL* ssl;
SSL_CTX* ctx; SSL_CTX* ctx;
......
...@@ -886,7 +886,7 @@ void MQTTAsync_writeComplete(int socket) ...@@ -886,7 +886,7 @@ void MQTTAsync_writeComplete(int socket)
{ {
MQTTAsyncs* m = (MQTTAsyncs*)(found->content); MQTTAsyncs* m = (MQTTAsyncs*)(found->content);
time(&(m->c->net.lastContact)); time(&(m->c->net.lastSent));
/* see if there is a pending write flagged */ /* see if there is a pending write flagged */
if (m->pending_write) if (m->pending_write)
...@@ -1428,13 +1428,16 @@ int MQTTAsync_completeConnection(MQTTAsyncs* m, MQTTPacket* pack) ...@@ -1428,13 +1428,16 @@ int MQTTAsync_completeConnection(MQTTAsyncs* m, MQTTPacket* pack)
rc = MQTTAsync_cleanSession(m->c); rc = MQTTAsync_cleanSession(m->c);
if (m->c->outboundMsgs->count > 0) if (m->c->outboundMsgs->count > 0)
{ {
time_t now;
ListElement* outcurrent = NULL; ListElement* outcurrent = NULL;
while (ListNextElement(m->c->outboundMsgs, &outcurrent)) while (ListNextElement(m->c->outboundMsgs, &outcurrent))
{ {
Messages* m = (Messages*)(outcurrent->content); Messages* m = (Messages*)(outcurrent->content);
m->lastTouch = 0; m->lastTouch = 0;
} }
MQTTProtocol_retry(m->c->net.lastContact, 1); time(&(now));
MQTTProtocol_retry(now, 1);
if (m->c->connected != 1) if (m->c->connected != 1)
rc = MQTTASYNC_DISCONNECTED; rc = MQTTASYNC_DISCONNECTED;
} }
...@@ -2533,11 +2536,13 @@ MQTTPacket* MQTTAsync_cycle(int* sock, unsigned long timeout, int* rc) ...@@ -2533,11 +2536,13 @@ MQTTPacket* MQTTAsync_cycle(int* sock, unsigned long timeout, int* rc)
if (!tostop && *sock == 0 && (tp.tv_sec > 0L || tp.tv_usec > 0L)) if (!tostop && *sock == 0 && (tp.tv_sec > 0L || tp.tv_usec > 0L))
{ {
MQTTAsync_sleep(100L); MQTTAsync_sleep(100L);
#if 0
if (s.clientsds->count == 0) if (s.clientsds->count == 0)
{ {
if (++nosockets_count == 50) /* 5 seconds with no sockets */ if (++nosockets_count == 50) /* 5 seconds with no sockets */
tostop = 1; tostop = 1;
} }
#endif
} }
else else
nosockets_count = 0; nosockets_count = 0;
......
...@@ -896,6 +896,7 @@ int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOptions* o ...@@ -896,6 +896,7 @@ int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOptions* o
rc = MQTTClient_cleanSession(m->c); rc = MQTTClient_cleanSession(m->c);
if (m->c->outboundMsgs->count > 0) if (m->c->outboundMsgs->count > 0)
{ {
time_t now;
ListElement* outcurrent = NULL; ListElement* outcurrent = NULL;
while (ListNextElement(m->c->outboundMsgs, &outcurrent)) while (ListNextElement(m->c->outboundMsgs, &outcurrent))
...@@ -903,7 +904,8 @@ int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOptions* o ...@@ -903,7 +904,8 @@ int MQTTClient_connectURIVersion(MQTTClient handle, MQTTClient_connectOptions* o
Messages* m = (Messages*)(outcurrent->content); Messages* m = (Messages*)(outcurrent->content);
m->lastTouch = 0; m->lastTouch = 0;
} }
MQTTProtocol_retry(m->c->net.lastContact, 1); time(&(now));
MQTTProtocol_retry(now, 1);
if (m->c->connected != 1) if (m->c->connected != 1)
rc = MQTTCLIENT_DISCONNECTED; rc = MQTTCLIENT_DISCONNECTED;
} }
...@@ -1957,7 +1959,7 @@ void MQTTClient_writeComplete(int socket) ...@@ -1957,7 +1959,7 @@ void MQTTClient_writeComplete(int socket)
{ {
MQTTClients* m = (MQTTClients*)(found->content); MQTTClients* m = (MQTTClients*)(found->content);
time(&(m->c->net.lastContact)); time(&(m->c->net.lastSent));
} }
FUNC_EXIT; FUNC_EXIT;
} }
...@@ -156,6 +156,8 @@ void* MQTTPacket_Factory(networkHandles* net, int* error) ...@@ -156,6 +156,8 @@ void* MQTTPacket_Factory(networkHandles* net, int* error)
#endif #endif
} }
} }
if (pack)
time(&(net->lastReceived));
exit: exit:
FUNC_EXIT_RC(*error); FUNC_EXIT_RC(*error);
return pack; return pack;
...@@ -197,7 +199,7 @@ int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buf ...@@ -197,7 +199,7 @@ int MQTTPacket_send(networkHandles* net, Header header, char* buffer, size_t buf
rc = Socket_putdatas(net->socket, buf, buf0len, 1, &buffer, &buflen, &free); rc = Socket_putdatas(net->socket, buf, buf0len, 1, &buffer, &buflen, &free);
if (rc == TCPSOCKET_COMPLETE) if (rc == TCPSOCKET_COMPLETE)
time(&(net->lastContact)); time(&(net->lastSent));
if (rc != TCPSOCKET_INTERRUPTED) if (rc != TCPSOCKET_INTERRUPTED)
free(buf); free(buf);
...@@ -244,7 +246,7 @@ int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffe ...@@ -244,7 +246,7 @@ int MQTTPacket_sends(networkHandles* net, Header header, int count, char** buffe
rc = Socket_putdatas(net->socket, buf, buf0len, count, buffers, buflens, frees); rc = Socket_putdatas(net->socket, buf, buf0len, count, buffers, buflens, frees);
if (rc == TCPSOCKET_COMPLETE) if (rc == TCPSOCKET_COMPLETE)
time(&(net->lastContact)); time(&(net->lastSent));
if (rc != TCPSOCKET_INTERRUPTED) if (rc != TCPSOCKET_INTERRUPTED)
free(buf); free(buf);
......
...@@ -514,7 +514,9 @@ void MQTTProtocol_keepalive(time_t now) ...@@ -514,7 +514,9 @@ void MQTTProtocol_keepalive(time_t now)
{ {
Clients* client = (Clients*)(current->content); Clients* client = (Clients*)(current->content);
ListNextElement(bstate->clients, &current); ListNextElement(bstate->clients, &current);
if (client->connected && client->keepAliveInterval > 0 && (difftime(now, client->net.lastContact) >= client->keepAliveInterval)) if (client->connected && client->keepAliveInterval > 0 &&
(difftime(now, client->net.lastSent) >= client->keepAliveInterval ||
difftime(now, client->net.lastReceived) >= client->keepAliveInterval))
{ {
if (client->ping_outstanding == 0) if (client->ping_outstanding == 0)
{ {
...@@ -527,7 +529,7 @@ void MQTTProtocol_keepalive(time_t now) ...@@ -527,7 +529,7 @@ void MQTTProtocol_keepalive(time_t now)
} }
else else
{ {
client->net.lastContact = now; client->net.lastSent = now;
client->ping_outstanding = 1; client->ping_outstanding = 1;
} }
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment