michaelh/zephyr
1
0
Fork 0
Code Releases Activity

net: mqtt: Add MQTT 5.0 support for SUBACK/UNSUBACK

Browse source 
Add support for SUBACK/UNSUBACK messaged specified in MQTT 5.0.

Signed-off-by: Robert Lubos <robert.lubos@nordicsemi.no>
This commit is contained in:
Robert Lubos 2024-12-19 16:09:09 +01:00 • committed by Benjamin Cabé
commit 2459ffae41
5 changed files with 91 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

20
include/zephyr/net/mqtt.h
View file

@ -411,16 +411,32 @@ struct mqtt_pubcomp_param {
struct mqtt_suback_param {
/** Message id of the SUBSCRIBE message being acknowledged */
uint16_t message_id;
/** Return codes indicating maximum QoS level granted for each topic
* in the subscription list.
/** MQTT 3.1 - Return codes indicating maximum QoS level granted for
* each topic in the subscription list.
* MQTT 5.0 - Reason codes corresponding to each topic in the
* subscription list.
*/
struct mqtt_binstr return_codes;
#if defined(CONFIG_MQTT_VERSION_5_0)
/** MQTT 5.0 properties. */
struct mqtt_common_ack_properties prop;
#endif /* CONFIG_MQTT_VERSION_5_0 */
};
/** @brief Parameters for MQTT unsubscribe acknowledgment (UNSUBACK). */
struct mqtt_unsuback_param {
/** Message id of the UNSUBSCRIBE message being acknowledged */
uint16_t message_id;
#if defined(CONFIG_MQTT_VERSION_5_0)
/** Reason codes corresponding to each topic in the unsubscription list. */
struct mqtt_binstr reason_codes;
/** MQTT 5.0 properties. */
struct mqtt_common_ack_properties prop;
#endif /* CONFIG_MQTT_VERSION_5_0 */
};
/** @brief Parameters for a publish message (PUBLISH). */

68
subsys/net/lib/mqtt/mqtt_decoder.c
View file

@ -935,10 +935,28 @@ int publish_complete_decode(const struct mqtt_client *client, struct buf_ctx *bu
}
#endif
return common_ack_decode(buf, &param->message_id, reason_code, prop);
return common_pub_ack_decode(buf, &param->message_id, reason_code, prop);
}
int subscribe_ack_decode(struct buf_ctx *buf, struct mqtt_suback_param *param)
#if defined(CONFIG_MQTT_VERSION_5_0)
static int suback_properties_decode(struct buf_ctx *buf,
struct mqtt_suback_param *param)
{
return common_ack_properties_decode(buf, &param->prop);
}
#else
static int suback_properties_decode(struct buf_ctx *buf,
struct mqtt_suback_param *param)
{
ARG_UNUSED(param);
ARG_UNUSED(buf);
return -ENOTSUP;
}
#endif /* CONFIG_MQTT_VERSION_5_0 */
int subscribe_ack_decode(const struct mqtt_client *client, struct buf_ctx *buf,
struct mqtt_suback_param *param)
{
int err_code;
@ -947,11 +965,53 @@ int subscribe_ack_decode(struct buf_ctx *buf, struct mqtt_suback_param *param)
return err_code;
}
if (mqtt_is_version_5_0(client)) {
err_code = suback_properties_decode(buf, param);
if (err_code < 0) {
return err_code;
}
}
return unpack_raw_data(buf->end - buf->cur, buf, &param->return_codes);
}
int unsubscribe_ack_decode(struct buf_ctx *buf,
#if defined(CONFIG_MQTT_VERSION_5_0)
static int unsuback_5_0_decode(struct buf_ctx *buf,
struct mqtt_unsuback_param *param)
{
return unpack_uint16(buf, &param->message_id);
int err;
err = common_ack_properties_decode(buf, &param->prop);
if (err < 0) {
return err;
}
return unpack_raw_data(buf->end - buf->cur, buf, &param->reason_codes);
}
#else
static int unsuback_5_0_decode(struct buf_ctx *buf,
struct mqtt_unsuback_param *param)
{
ARG_UNUSED(param);
ARG_UNUSED(buf);
return -ENOTSUP;
}
#endif /* CONFIG_MQTT_VERSION_5_0 */
int unsubscribe_ack_decode(const struct mqtt_client *client, struct buf_ctx *buf,
struct mqtt_unsuback_param *param)
{
int err;
err = unpack_uint16(buf, &param->message_id);
if (err < 0) {
return 0;
}
if (mqtt_is_version_5_0(client)) {
return unsuback_5_0_decode(buf, param);
}
return 0;
}

6
subsys/net/lib/mqtt/mqtt_internal.h
View file

@ -423,24 +423,26 @@ int publish_complete_decode(const struct mqtt_client *client, struct buf_ctx *bu
/**@brief Decode MQTT Subscribe packet.
*
* @param[in] MQTT client for which packet is decoded.
* @param[inout] buf A pointer to the buf_ctx structure containing current
* buffer position.
* @param[out] param Pointer to buffer for decoded Subscribe parameters.
*
* @return 0 if the procedure is successful, an error code otherwise.
*/
int subscribe_ack_decode(struct buf_ctx *buf,
int subscribe_ack_decode(const struct mqtt_client *client, struct buf_ctx *buf,
struct mqtt_suback_param *param);
/**@brief Decode MQTT Unsubscribe packet.
*
* @param[in] MQTT client for which packet is decoded.
* @param[inout] buf A pointer to the buf_ctx structure containing current
* buffer position.
* @param[out] param Pointer to buffer for decoded Unsubscribe parameters.
*
* @return 0 if the procedure is successful, an error code otherwise.
*/
int unsubscribe_ack_decode(struct buf_ctx *buf,
int unsubscribe_ack_decode(const struct mqtt_client *client, struct buf_ctx *buf,
struct mqtt_unsuback_param *param);
/**

5
subsys/net/lib/mqtt/mqtt_rx.c
View file

@ -113,7 +113,7 @@ static int mqtt_handle_packet(struct mqtt_client *client,
NET_DBG("[CID %p]: Received MQTT_PKT_TYPE_SUBACK!", client);
evt.type = MQTT_EVT_SUBACK;
err_code = subscribe_ack_decode(buf, &evt.param.suback);
err_code = subscribe_ack_decode(client, buf, &evt.param.suback);
evt.result = err_code;
break;
@ -121,7 +121,8 @@ static int mqtt_handle_packet(struct mqtt_client *client,
NET_DBG("[CID %p]: Received MQTT_PKT_TYPE_UNSUBACK!", client);
evt.type = MQTT_EVT_UNSUBACK;
err_code = unsubscribe_ack_decode(buf, &evt.param.unsuback);
err_code = unsubscribe_ack_decode(client, buf,
&evt.param.unsuback);
evt.result = err_code;
break;

4
tests/net/lib/mqtt/v3_1_1/mqtt_packet/src/mqtt_packet.c
View file

@ -874,7 +874,7 @@ static int eval_msg_suback(struct mqtt_test *mqtt_test)
zassert_false(rc, "fixed_header_decode failed");
rc = subscribe_ack_decode(&buf, &dec_param);
rc = subscribe_ack_decode(&client, &buf, &dec_param);
/**TESTPOINT: Check subscribe_ack_decode function*/
zassert_false(rc, "subscribe_ack_decode failed");
@ -1083,7 +1083,7 @@ static int eval_msg_unsuback(struct mqtt_test *mqtt_test)
zassert_false(rc, "fixed_header_decode failed");
rc = unsubscribe_ack_decode(&buf, &dec_param);
rc = unsubscribe_ack_decode(&client, &buf, &dec_param);
zassert_false(rc, "unsubscribe_ack_decode failed");