Skip to content

Commit ef64e18

Browse files
rlubosrlubos
committed
[nrf fromtree] net: mqtt: Add MQTT 5.0 support for SUBSCRIBE/UNSUBSCRIBE
Add support for SUBSCRIBE/UNSUBSCRIBE messages specified in MQTT 5.0. Signed-off-by: Robert Lubos <[email protected]> (cherry picked from commit efd795b)
1 parent 0276e28 commit ef64e18

File tree

5 files changed

+149
-8
lines changed

5 files changed

+149
-8
lines changed

‎include/zephyr/net/mqtt.h‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -493,7 +493,7 @@ struct mqtt_publish_param {
493493
#endif /* CONFIG_MQTT_VERSION_5_0 */
494494
};
495495

496-
/** @brief List of topics in a subscription request. */
496+
/** @brief Parameters for subscribe/unsubscribe message. */
497497
struct mqtt_subscription_list {
498498
/** Array containing topics along with QoS for each. */
499499
struct mqtt_topic *list;
@@ -503,6 +503,19 @@ struct mqtt_subscription_list {
503503

504504
/** Message id used to identify subscription request. */
505505
uint16_t message_id;
506+
507+
#if defined(CONFIG_MQTT_VERSION_5_0)
508+
/** MQTT 5.0 properties. */
509+
struct {
510+
/** MQTT 5.0, chapter 3.8.2.1.3 / 3.10.2.1.2 User Property. */
511+
struct mqtt_utf8_pair user_prop[CONFIG_MQTT_USER_PROPERTIES_MAX];
512+
513+
/** MQTT 5.0, chapter 3.8.2.1.2 Subscription Identifier.
514+
* Ignored for UNSUBSCRIBE requests.
515+
*/
516+
uint32_t subscription_identifier;
517+
} prop;
518+
#endif /* CONFIG_MQTT_VERSION_5_0 */
506519
};
507520

508521
/**

‎subsys/net/lib/mqtt/mqtt.c‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -495,7 +495,7 @@ int mqtt_subscribe(struct mqtt_client *client,
495495
goto error;
496496
}
497497

498-
err_code = subscribe_encode(param, &packet);
498+
err_code = subscribe_encode(client, param, &packet);
499499
if (err_code < 0) {
500500
goto error;
501501
}
@@ -529,7 +529,7 @@ int mqtt_unsubscribe(struct mqtt_client *client,
529529
goto error;
530530
}
531531

532-
err_code = unsubscribe_encode(param, &packet);
532+
err_code = unsubscribe_encode(client, param, &packet);
533533
if (err_code < 0) {
534534
goto error;
535535
}

‎subsys/net/lib/mqtt/mqtt_encoder.c‎

Lines changed: 128 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -383,6 +383,32 @@ static int encode_uint32_property(uint8_t prop, uint32_t value,
383383
return pack_uint32(value, buf);
384384
}
385385

386+
static size_t var_int_property_length(uint32_t value)
387+
{
388+
if (value == 0) {
389+
return 0;
390+
}
391+
392+
return sizeof(uint8_t) + (size_t)pack_variable_int(value, NULL);
393+
}
394+
395+
static int encode_var_int_property(uint8_t prop, uint32_t value,
396+
struct buf_ctx *buf)
397+
{
398+
int err;
399+
400+
if (value == 0) {
401+
return 0;
402+
}
403+
404+
err = pack_uint8(prop, buf);
405+
if (err < 0) {
406+
return err;
407+
}
408+
409+
return pack_variable_int(value, buf);
410+
}
411+
386412
static size_t string_property_length(const struct mqtt_utf8 *str)
387413
{
388414
if (str->size == 0) {
@@ -1164,7 +1190,54 @@ int disconnect_encode(struct buf_ctx *buf)
11641190
return 0;
11651191
}
11661192

1167-
int subscribe_encode(const struct mqtt_subscription_list *param,
1193+
#if defined(CONFIG_MQTT_VERSION_5_0)
1194+
static uint32_t subscribe_properties_length(
1195+
const struct mqtt_subscription_list *param)
1196+
{
1197+
return var_int_property_length(param->prop.subscription_identifier) +
1198+
user_properties_length(param->prop.user_prop);
1199+
}
1200+
1201+
static int subscribe_properties_encode(const struct mqtt_subscription_list *param,
1202+
struct buf_ctx *buf)
1203+
{
1204+
uint32_t properties_len;
1205+
int err;
1206+
1207+
/* Precalculate total properties length */
1208+
properties_len = subscribe_properties_length(param);
1209+
err = pack_variable_int(properties_len, buf);
1210+
if (err < 0) {
1211+
return err;
1212+
}
1213+
1214+
err = encode_var_int_property(MQTT_PROP_SUBSCRIPTION_IDENTIFIER,
1215+
param->prop.subscription_identifier,
1216+
buf);
1217+
if (err < 0) {
1218+
return err;
1219+
}
1220+
1221+
err = encode_user_properties(param->prop.user_prop, buf);
1222+
if (err < 0) {
1223+
return err;
1224+
}
1225+
1226+
return 0;
1227+
}
1228+
#else
1229+
static int subscribe_properties_encode(const struct mqtt_subscription_list *param,
1230+
struct buf_ctx *buf)
1231+
{
1232+
ARG_UNUSED(param);
1233+
ARG_UNUSED(buf);
1234+
1235+
return -ENOTSUP;
1236+
}
1237+
#endif /* CONFIG_MQTT_VERSION_5_0 */
1238+
1239+
int subscribe_encode(const struct mqtt_client *client,
1240+
const struct mqtt_subscription_list *param,
11681241
struct buf_ctx *buf)
11691242
{
11701243
const uint8_t message_type = MQTT_MESSAGES_OPTIONS(
@@ -1186,6 +1259,13 @@ int subscribe_encode(const struct mqtt_subscription_list *param,
11861259
return err_code;
11871260
}
11881261

1262+
if (mqtt_is_version_5_0(client)) {
1263+
err_code = subscribe_properties_encode(param, buf);
1264+
if (err_code != 0) {
1265+
return err_code;
1266+
}
1267+
}
1268+
11891269
for (i = 0; i < param->list_count; i++) {
11901270
err_code = pack_utf8_str(&param->list[i].topic, buf);
11911271
if (err_code != 0) {
@@ -1201,7 +1281,46 @@ int subscribe_encode(const struct mqtt_subscription_list *param,
12011281
return mqtt_encode_fixed_header(message_type, start, buf);
12021282
}
12031283

1204-
int unsubscribe_encode(const struct mqtt_subscription_list *param,
1284+
#if defined(CONFIG_MQTT_VERSION_5_0)
1285+
static uint32_t unsubscribe_properties_length(
1286+
const struct mqtt_subscription_list *param)
1287+
{
1288+
return user_properties_length(param->prop.user_prop);
1289+
}
1290+
1291+
static int unsubscribe_properties_encode(
1292+
const struct mqtt_subscription_list *param, struct buf_ctx *buf)
1293+
{
1294+
uint32_t properties_len;
1295+
int err;
1296+
1297+
/* Precalculate total properties length */
1298+
properties_len = unsubscribe_properties_length(param);
1299+
err = pack_variable_int(properties_len, buf);
1300+
if (err < 0) {
1301+
return err;
1302+
}
1303+
1304+
err = encode_user_properties(param->prop.user_prop, buf);
1305+
if (err < 0) {
1306+
return err;
1307+
}
1308+
1309+
return 0;
1310+
}
1311+
#else
1312+
static int unsubscribe_properties_encode(
1313+
const struct mqtt_subscription_list *param, struct buf_ctx *buf)
1314+
{
1315+
ARG_UNUSED(param);
1316+
ARG_UNUSED(buf);
1317+
1318+
return -ENOTSUP;
1319+
}
1320+
#endif /* CONFIG_MQTT_VERSION_5_0 */
1321+
1322+
int unsubscribe_encode(const struct mqtt_client *client,
1323+
const struct mqtt_subscription_list *param,
12051324
struct buf_ctx *buf)
12061325
{
12071326
const uint8_t message_type = MQTT_MESSAGES_OPTIONS(
@@ -1218,6 +1337,13 @@ int unsubscribe_encode(const struct mqtt_subscription_list *param,
12181337
return err_code;
12191338
}
12201339

1340+
if (mqtt_is_version_5_0(client)) {
1341+
err_code = unsubscribe_properties_encode(param, buf);
1342+
if (err_code != 0) {
1343+
return err_code;
1344+
}
1345+
}
1346+
12211347
for (i = 0; i < param->list_count; i++) {
12221348
err_code = pack_utf8_str(&param->list[i].topic, buf);
12231349
if (err_code != 0) {

‎subsys/net/lib/mqtt/mqtt_internal.h‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -305,7 +305,8 @@ int disconnect_encode(struct buf_ctx *buf);
305305
*
306306
* @return 0 if the procedure is successful, an error code otherwise.
307307
*/
308-
int subscribe_encode(const struct mqtt_subscription_list *param,
308+
int subscribe_encode(const struct mqtt_client *client,
309+
const struct mqtt_subscription_list *param,
309310
struct buf_ctx *buf);
310311

311312
/**@brief Constructs/encodes Unsubscribe packet.
@@ -318,7 +319,8 @@ int subscribe_encode(const struct mqtt_subscription_list *param,
318319
*
319320
* @return 0 if the procedure is successful, an error code otherwise.
320321
*/
321-
int unsubscribe_encode(const struct mqtt_subscription_list *param,
322+
int unsubscribe_encode(const struct mqtt_client *client,
323+
const struct mqtt_subscription_list *param,
322324
struct buf_ctx *buf);
323325

324326
/**@brief Constructs/encodes Ping Request packet.

‎tests/net/lib/mqtt/v3_1_1/mqtt_packet/src/mqtt_packet.c‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -846,7 +846,7 @@ static int eval_msg_subscribe(struct mqtt_test *mqtt_test)
846846
buf.cur = client.tx_buf;
847847
buf.end = client.tx_buf + client.tx_buf_size;
848848

849-
rc = subscribe_encode(param, &buf);
849+
rc = subscribe_encode(&client, param, &buf);
850850

851851
/**TESTPOINT: Check subscribe_encode function*/
852852
zassert_false(rc, "subscribe_encode failed");

0 commit comments

Comments
 (0)