Skip to content
This repository was archived by the owner on Jan 6, 2023. It is now read-only.

Commit fb6678e

Browse files
gguussgguuss
authored
Exponentialbackoff (#93)
Adds exponential backoff and extra subscription topics
1 parent 3e1a206 commit fb6678e

File tree

6 files changed

+365
-20
lines changed

6 files changed

+365
-20
lines changed

‎examples/Esp32-lwmqtt/ciotc_config.h

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -77,3 +77,11 @@ const char *root_cert =
7777
"7a8IVk6wuy6pm+T7HT4LY8ibS5FEZlfAFLSW8NwsVz9SBK2Vqn1N0PIMn5xA6NZV\n"
7878
"c7o835DLAFshEWfC7TIe3g==\n"
7979
"-----END CERTIFICATE-----\n";
80+
81+
// In case we ever need extra topics
82+
const int ex_num_topics = 0;
83+
const char* ex_topics[ex_num_topics];
84+
//const int ex_num_topics = 1;
85+
//const char* ex_topics[ex_num_topics] = {
86+
// "/devices/my-device/tbd/#"
87+
//};

‎examples/Esp32-lwmqtt/esp32-mqtt.h

Lines changed: 114 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -92,16 +92,124 @@ void publishState(String data) {
9292
mqttClient->publish(device->getStateTopic(), data);
9393
}
9494

95+
// FIXME: Move to config?
96+
int __backoff__ = 1000; // current backoff, milliseconds
97+
int __factor__ = 2.5f;
98+
int __minbackoff__ = 1000; // minimum backoff, ms
99+
int __max_backoff__ = 60000; // maximum backoff, ms
100+
int __jitter__ = 500; // max random jitter, ms
95101
void mqttConnect() {
96102
Serial.print("\nconnecting...");
97-
while (!mqttClient->connect(device->getClientId().c_str(), "unused", getJwt().c_str(), false)) {
98-
Serial.println(mqttClient->lastError());
99-
Serial.println(mqttClient->returnCode());
100-
delay(1000);
103+
bool keepgoing = true;
104+
while (keepgoing) {
105+
mqttClient->connect(device->getClientId().c_str(), "unused", getJwt().c_str(), false);
106+
107+
if (mqttClient->lastError() != LWMQTT_SUCCESS){
108+
Serial.println(mqttClient->lastError());
109+
switch(mqttClient->lastError()) {
110+
case (LWMQTT_BUFFER_TOO_SHORT):
111+
Serial.println("LWMQTT_BUFFER_TOO_SHORT");
112+
break;
113+
case (LWMQTT_VARNUM_OVERFLOW):
114+
Serial.println("LWMQTT_VARNUM_OVERFLOW");
115+
break;
116+
case (LWMQTT_NETWORK_FAILED_CONNECT):
117+
Serial.println("LWMQTT_NETWORK_FAILED_CONNECT");
118+
break;
119+
case (LWMQTT_NETWORK_TIMEOUT):
120+
Serial.println("LWMQTT_NETWORK_TIMEOUT");
121+
break;
122+
case (LWMQTT_NETWORK_FAILED_READ):
123+
Serial.println("LWMQTT_NETWORK_FAILED_READ");
124+
break;
125+
case (LWMQTT_NETWORK_FAILED_WRITE):
126+
Serial.println("LWMQTT_NETWORK_FAILED_WRITE");
127+
break;
128+
case (LWMQTT_REMAINING_LENGTH_OVERFLOW):
129+
Serial.println("LWMQTT_REMAINING_LENGTH_OVERFLOW");
130+
break;
131+
case (LWMQTT_REMAINING_LENGTH_MISMATCH):
132+
Serial.println("LWMQTT_REMAINING_LENGTH_MISMATCH");
133+
break;
134+
case (LWMQTT_MISSING_OR_WRONG_PACKET):
135+
Serial.println("LWMQTT_MISSING_OR_WRONG_PACKET");
136+
break;
137+
case (LWMQTT_CONNECTION_DENIED):
138+
Serial.println("LWMQTT_CONNECTION_DENIED");
139+
break;
140+
case (LWMQTT_FAILED_SUBSCRIPTION):
141+
Serial.println("LWMQTT_FAILED_SUBSCRIPTION");
142+
break;
143+
case (LWMQTT_SUBACK_ARRAY_OVERFLOW):
144+
Serial.println("LWMQTT_SUBACK_ARRAY_OVERFLOW");
145+
break;
146+
case (LWMQTT_PONG_TIMEOUT):
147+
Serial.println("LWMQTT_PONG_TIMEOUT");
148+
break;
149+
default:
150+
Serial.println("This error code should never be reached.");
151+
break;
152+
}
153+
154+
Serial.println(mqttClient->returnCode());
155+
switch(mqttClient->returnCode()) {
156+
case (LWMQTT_CONNECTION_ACCEPTED):
157+
Serial.println("OK");
158+
break;
159+
case (LWMQTT_UNACCEPTABLE_PROTOCOL):
160+
Serial.println("LWMQTT_UNACCEPTABLE_PROTOCOLL");
161+
break;
162+
case (LWMQTT_IDENTIFIER_REJECTED):
163+
Serial.println("LWMQTT_IDENTIFIER_REJECTED");
164+
break;
165+
case (LWMQTT_SERVER_UNAVAILABLE):
166+
Serial.println("LWMQTT_SERVER_UNAVAILABLE");
167+
break;
168+
case (LWMQTT_BAD_USERNAME_OR_PASSWORD):
169+
Serial.println("LWMQTT_BAD_USERNAME_OR_PASSWORD");
170+
iss = 0; // Force JWT regeneration
171+
break;
172+
case (LWMQTT_NOT_AUTHORIZED):
173+
Serial.println("LWMQTT_NOT_AUTHORIZED");
174+
iss = 0; // Force JWT regeneration
175+
break;
176+
case (LWMQTT_UNKNOWN_RETURN_CODE):
177+
Serial.println("LWMQTT_UNKNOWN_RETURN_CODE");
178+
break;
179+
default:
180+
Serial.println("This return code should never be reached.");
181+
break;
182+
}
183+
// See https://cloud.google.com/iot/docs/how-tos/exponential-backoff
184+
if (__backoff__ < __minbackoff__) {
185+
__backoff__ = __minbackoff__;
186+
}
187+
__backoff__ = (__backoff__ * __factor__) + random(__jitter__);
188+
if (__backoff__ > __max_backoff__) {
189+
__backoff__ = __max_backoff__;
190+
}
191+
192+
// Clean up the client
193+
mqttClient->disconnect();
194+
Serial.println("Delaying " + String(__backoff__) + "ms");
195+
delay(__backoff__);
196+
keepgoing = true;
197+
} else {
198+
// We're now connected
199+
Serial.println("\nconnected!");
200+
keepgoing = false;
201+
__backoff__ = __minbackoff__;
202+
}
101203
}
102-
Serial.println("\nconnected!");
204+
103205
mqttClient->subscribe(device->getConfigTopic(), 1); // Set QoS to 1 (ack) for configuration messages
104206
mqttClient->subscribe(device->getCommandsTopic(), 0); // QoS 0 (no ack) for commands
207+
if (ex_num_topics > 0) { // Subscribe to the extra topics
208+
for (int i=0; i < ex_num_topics; i++) {
209+
mqttClient->subscribe(ex_topics[i], 0); // QoS 0 (no ack) for commands
210+
}
211+
}
212+
105213
publishState("connected");
106214
}
107215

@@ -128,4 +236,4 @@ void setupCloudIoT() {
128236
mqttClient->setOptions(180, true, 1000); // keepAlive, cleanSession, timeout
129237
startMQTT();
130238
}
131-
#endif //__ESP32_MQTT_H__
239+
#endif //__ESP32_MQTT_H__

‎examples/Esp8266-lwmqtt/ciotc_config.h

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -168,3 +168,12 @@ const char* private_key_str =
168168
// };
169169
//
170170
// size_t ca_crt_len = 1374;
171+
172+
173+
// In case we ever need extra topics
174+
const int ex_num_topics = 0;
175+
const char* ex_topics[ex_num_topics];
176+
//const int ex_num_topics = 1;
177+
//const char* ex_topics[ex_num_topics] = {
178+
// "/devices/my-device/tbd/#"
179+
//};

‎examples/Esp8266-lwmqtt/esp8266_mqtt.h

Lines changed: 113 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -116,21 +116,127 @@ void publishState(String data) {
116116
mqttClient->publish(device->getStateTopic(), data);
117117
}
118118

119+
// FIXME: Move to config?
120+
int __backoff__ = 1000; // current backoff, milliseconds
121+
int __factor__ = 2.5f;
122+
int __minbackoff__ = 1000; // minimum backoff, ms
123+
int __max_backoff__ = 60000; // maximum backoff, ms
124+
int __jitter__ = 500; // max random jitter, ms
119125
void mqttConnect() {
120126
Serial.print("\nconnecting...");
121-
while (!mqttClient->connect(device->getClientId().c_str(), "unused", getJwt().c_str(), false)) {
122-
Serial.print(".");
123-
Serial.println(mqttClient->lastError());
124-
Serial.println(mqttClient->returnCode());
125-
delay(1000);
127+
bool keepgoing = true;
128+
while (keepgoing) {
129+
mqttClient->connect(device->getClientId().c_str(), "unused", getJwt().c_str(), false);
130+
131+
if (mqttClient->lastError() != LWMQTT_SUCCESS){
132+
Serial.println(mqttClient->lastError());
133+
switch(mqttClient->lastError()) {
134+
case (LWMQTT_BUFFER_TOO_SHORT):
135+
Serial.println("LWMQTT_BUFFER_TOO_SHORT");
136+
break;
137+
case (LWMQTT_VARNUM_OVERFLOW):
138+
Serial.println("LWMQTT_VARNUM_OVERFLOW");
139+
break;
140+
case (LWMQTT_NETWORK_FAILED_CONNECT):
141+
Serial.println("LWMQTT_NETWORK_FAILED_CONNECT");
142+
break;
143+
case (LWMQTT_NETWORK_TIMEOUT):
144+
Serial.println("LWMQTT_NETWORK_TIMEOUT");
145+
break;
146+
case (LWMQTT_NETWORK_FAILED_READ):
147+
Serial.println("LWMQTT_NETWORK_FAILED_READ");
148+
break;
149+
case (LWMQTT_NETWORK_FAILED_WRITE):
150+
Serial.println("LWMQTT_NETWORK_FAILED_WRITE");
151+
break;
152+
case (LWMQTT_REMAINING_LENGTH_OVERFLOW):
153+
Serial.println("LWMQTT_REMAINING_LENGTH_OVERFLOW");
154+
break;
155+
case (LWMQTT_REMAINING_LENGTH_MISMATCH):
156+
Serial.println("LWMQTT_REMAINING_LENGTH_MISMATCH");
157+
break;
158+
case (LWMQTT_MISSING_OR_WRONG_PACKET):
159+
Serial.println("LWMQTT_MISSING_OR_WRONG_PACKET");
160+
break;
161+
case (LWMQTT_CONNECTION_DENIED):
162+
Serial.println("LWMQTT_CONNECTION_DENIED");
163+
break;
164+
case (LWMQTT_FAILED_SUBSCRIPTION):
165+
Serial.println("LWMQTT_FAILED_SUBSCRIPTION");
166+
break;
167+
case (LWMQTT_SUBACK_ARRAY_OVERFLOW):
168+
Serial.println("LWMQTT_SUBACK_ARRAY_OVERFLOW");
169+
break;
170+
case (LWMQTT_PONG_TIMEOUT):
171+
Serial.println("LWMQTT_PONG_TIMEOUT");
172+
break;
173+
default:
174+
Serial.println("This error code should never be reached.");
175+
break;
176+
}
177+
178+
Serial.println(mqttClient->returnCode());
179+
switch(mqttClient->returnCode()) {
180+
case (LWMQTT_CONNECTION_ACCEPTED):
181+
Serial.println("OK");
182+
break;
183+
case (LWMQTT_UNACCEPTABLE_PROTOCOL):
184+
Serial.println("LWMQTT_UNACCEPTABLE_PROTOCOLL");
185+
break;
186+
case (LWMQTT_IDENTIFIER_REJECTED):
187+
Serial.println("LWMQTT_IDENTIFIER_REJECTED");
188+
break;
189+
case (LWMQTT_SERVER_UNAVAILABLE):
190+
Serial.println("LWMQTT_SERVER_UNAVAILABLE");
191+
break;
192+
case (LWMQTT_BAD_USERNAME_OR_PASSWORD):
193+
Serial.println("LWMQTT_BAD_USERNAME_OR_PASSWORD");
194+
iss = 0; // Force JWT regeneration
195+
break;
196+
case (LWMQTT_NOT_AUTHORIZED):
197+
Serial.println("LWMQTT_NOT_AUTHORIZED");
198+
iss = 0; // Force JWT regeneration
199+
break;
200+
case (LWMQTT_UNKNOWN_RETURN_CODE):
201+
Serial.println("LWMQTT_UNKNOWN_RETURN_CODE");
202+
break;
203+
default:
204+
Serial.println("This return code should never be reached.");
205+
break;
206+
}
207+
// See https://cloud.google.com/iot/docs/how-tos/exponential-backoff
208+
if (__backoff__ < __minbackoff__) {
209+
__backoff__ = __minbackoff__;
210+
}
211+
__backoff__ = (__backoff__ * __factor__) + random(__jitter__);
212+
if (__backoff__ > __max_backoff__) {
213+
__backoff__ = __max_backoff__;
214+
}
215+
216+
// Clean up the client
217+
mqttClient->disconnect();
218+
Serial.println("Delaying " + String(__backoff__) + "ms");
219+
delay(__backoff__);
220+
keepgoing = true;
221+
} else {
222+
// We're now connected
223+
Serial.println("\nconnected!");
224+
keepgoing = false;
225+
__backoff__ = __minbackoff__;
226+
}
126227
}
127-
Serial.println("\nconnected!");
228+
128229
mqttClient->subscribe(device->getConfigTopic(), 1); // Set QoS to 1 (ack) for configuration messages
129230
mqttClient->subscribe(device->getCommandsTopic(), 0); // QoS 0 (no ack) for commands
231+
if (ex_num_topics > 0) { // Subscribe to the extra topics
232+
for (int i=0; i < ex_num_topics; i++) {
233+
mqttClient->subscribe(ex_topics[i], 0); // QoS 0 (no ack) for commands
234+
}
235+
}
236+
130237
publishState("connected");
131238
}
132239

133-
134240
///////////////////////////////
135241
// Orchestrates various methods from preceeding code.
136242
///////////////////////////////

‎examples/MKR1000-lwmqtt/ciotc_config.h

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -41,4 +41,10 @@ const char* private_key_str =
4141
"07:fd:ed:22:0d:03:2b:a6:b1:b6:04:0b:d5:9b:49:"
4242
"7d:ca";
4343

44-
44+
// In case we ever need extra topics
45+
const int ex_num_topics = 0;
46+
const char* ex_topics[ex_num_topics];
47+
//const int ex_num_topics = 1;
48+
//const char* ex_topics[ex_num_topics] = {
49+
// "/devices/my-device/tbd/#"
50+
//};

0 commit comments

Comments
 (0)