samples: net: MQTT: transport: Respect Run To Completion

Ensure that state changes are executed in order by queueing an event
on callbacks from the internal MQTT thread. This ensures correct order
and state changes in the state machine.

Signed-off-by: Simen S. Røstad <[email protected]>

Author: simensrostad

samples/net/mqtt/src/modules/transport/transport.c

@@ -46,6 +46,24 @@ static char client_id[CONFIG_MQTT_SAMPLE_TRANSPORT_CLIENT_ID_BUFFER_SIZE];
 static uint8_t pub_topic[sizeof(client_id) + sizeof(CONFIG_MQTT_SAMPLE_TRANSPORT_PUBLISH_TOPIC)];
 static uint8_t sub_topic[sizeof(client_id) + sizeof(CONFIG_MQTT_SAMPLE_TRANSPORT_SUBSCRIBE_TOPIC)];
 
+enum transport_event_type {
+	CONNECTED,
+	DISCONNECTED,
+};
+
+struct transport_event {
+	enum transport_event_type type;
+};
+
+/* Private channel for internal events */
+ZBUS_CHAN_DEFINE(TRANSPORT_PRIVATE_CHANNEL,
+		 struct transport_event,
+		 NULL,
+		 NULL,
+		 ZBUS_OBSERVERS(transport),
+		 ZBUS_MSG_INIT(0)
+);
+
 /* User defined state object.
  * Used to transfer data between state changes.
  */
@@ -70,15 +88,37 @@ static struct s_object {
 static void on_mqtt_connack(enum mqtt_conn_return_code return_code, bool session_present)
 {
 	ARG_UNUSED(return_code);
+	ARG_UNUSED(session_present);
+
+	int err;
+	struct transport_event event = {
+		.type = CONNECTED,
+	};
 
-	smf_set_state(SMF_CTX(&s_obj), &state[MQTT_CONNECTED]);
+	if (return_code != MQTT_CONNECTION_ACCEPTED) {
+		LOG_ERR("MQTT broker rejected connection, return code: %d", return_code);
+		return;
+	}
+
+	err = zbus_chan_pub(&TRANSPORT_PRIVATE_CHANNEL, &event, K_SECONDS(1));
+	if (err) {
+		LOG_ERR("zbus_chan_pub, error: %d", err);
+		SEND_FATAL_ERROR();
+	}
 }
 
 static void on_mqtt_disconnect(int result)
 {
-	ARG_UNUSED(result);
+	int err;
+	struct transport_event event = {
+		.type = DISCONNECTED,
+	};
 
-	smf_set_state(SMF_CTX(&s_obj), &state[MQTT_DISCONNECTED]);
+	err = zbus_chan_pub(&TRANSPORT_PRIVATE_CHANNEL, &event, K_SECONDS(1));
+	if (err) {
+		LOG_ERR("zbus_chan_pub, error: %d", err);
+		SEND_FATAL_ERROR();
+	}
 }
 
 static void on_mqtt_publish(struct mqtt_helper_buf topic, struct mqtt_helper_buf payload)
@@ -381,6 +421,31 @@ static void transport_task(void)
 				return;
 			}
 		}
+
+		if (&TRANSPORT_PRIVATE_CHANNEL == chan) {
+
+			struct transport_event event;
+
+			err = zbus_chan_read(&TRANSPORT_PRIVATE_CHANNEL, &event, K_SECONDS(1));
+			if (err) {
+				LOG_ERR("zbus_chan_read, error: %d", err);
+				SEND_FATAL_ERROR();
+				return;
+			}
+
+			/* Process MQTT events and change state in the correct thread context */
+			switch (event.type) {
+			case CONNECTED:
+				smf_set_state(SMF_CTX(&s_obj), &state[MQTT_CONNECTED]);
+				break;
+			case DISCONNECTED:
+				smf_set_state(SMF_CTX(&s_obj), &state[MQTT_DISCONNECTED]);
+				break;
+			default:
+				LOG_WRN("Unknown MQTT event type: %d", event.type);
+				break;
+			}
+		}
 	}
 }