Mirrors/libwebsockets
1
0
Fork 0
mirror of https://github.com/warmcat/libwebsockets.git synced 2025-03-30 00:00:16 +01:00
Code Issues Releases Wiki Activity

mqtt: Setting the CONNECTED state only when SUBACK is received

Browse source 
Setting the CONNECTED state only when SUBACK is received if the stream has
defined a subscription topic. This is to avoid SS from sending out SUBSCRIBE
right after CONNACK, even when the connection is not valid.
This commit is contained in:
Sakthi Kannan 2021-03-22 11:58:28 -07:00 • committed by Andy Green
parent c15744f1d0
commit 9c7843fbe6
1 changed files with 27 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
lib/secure-streams/protocols/ss-mqtt.c
View file

@ -1,7 +1,7 @@
/* /*
* libwebsockets - small server side websockets and web server implementation * libwebsockets - small server side websockets and web server implementation
* *
* Copyright (C) 2019 - 2020 Andy Green <andy@warmcat.com> * Copyright (C) 2019 - 2021 Andy Green <andy@warmcat.com>
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to * of this software and associated documentation files (the "Software"), to
@ -93,6 +93,16 @@ secstream_mqtt(struct lws *wsi, enum lws_callback_reasons reason, void *user,
h->wsi = wsi; h->wsi = wsi;
h->retry = 0; h->retry = 0;
h->seqstate = SSSEQ_CONNECTED; h->seqstate = SSSEQ_CONNECTED;
/*
* If a subscribe is pending on the stream, then make
* sure the SUBSCRIBE is done before signaling the
* user application.
*/
if (h->policy->u.mqtt.subscribe &&
!wsi->mqtt->done_subscribe) {
lws_callback_on_writable(wsi);
break;
}
lws_sul_cancel(&h->sul); lws_sul_cancel(&h->sul);
#if defined(LWS_WITH_SYS_METRICS) #if defined(LWS_WITH_SYS_METRICS)
/* /*
@ -130,6 +140,17 @@ secstream_mqtt(struct lws *wsi, enum lws_callback_reasons reason, void *user,
return 0; /* don't passthru */ return 0; /* don't passthru */
case LWS_CALLBACK_MQTT_SUBSCRIBED: case LWS_CALLBACK_MQTT_SUBSCRIBED:
/*
* Stream demanded a subscribe while connecting, once
* done notify CONNECTED event to the application.
*/
if (wsi->mqtt->done_subscribe == 0) {
lws_sul_cancel(&h->sul);
r = lws_ss_event_helper(h, LWSSSCS_CONNECTED);
if (r != LWSSSSRET_OK)
return _lws_ss_handle_state_ret_CAN_DESTROY_HANDLE(r,
wsi, &h);
}
wsi->mqtt->done_subscribe = 1; wsi->mqtt->done_subscribe = 1;
lws_callback_on_writable(wsi); lws_callback_on_writable(wsi);
break; break;
@ -185,6 +206,11 @@ secstream_mqtt(struct lws *wsi, enum lws_callback_reasons reason, void *user,
lwsl_notice("%s: unable to subscribe", __func__); lwsl_notice("%s: unable to subscribe", __func__);
return -1; return -1;
} }
/* Expect a SUBACK */
if (lws_change_pollfd(wsi, 0, LWS_POLLIN)) {
lwsl_err("%s: Unable to set LWS_POLLIN\n", __func__);
return -1;
}
return 0; return 0;
} }