riot-ci/RIOT
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Merge pull request #15653 from haukepetersen/fix_asymcute_subflags

Browse Source 
net/asymcute: fix handling of short and pre-defined topics
This commit is contained in:
Martine Lenders 2020-12-18 13:52:52 +01:00 committed by GitHub
commit b5a36e9af7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 170 additions and 103 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

163
examples/asymcute_mqttsn/main.c
View File

@ -89,51 +89,37 @@ static asymcute_sub_t *_find_sub(const char *name)
return NULL; return NULL;
} }
static uint16_t _topic_parse_pre(const char *name) static uint16_t _parse_predef_id(const char *name)
{ {
if (strncmp(name, "pre_", 4) == 0) { uint16_t id = 0;
return (uint16_t)atoi(&name[4]); if ((strlen(name) > 4) && (strncmp(name, "pre_", 4) == 0)) {
id = (uint16_t)atoi(&name[4]);
} }
return 0; return id;
} }
static int _topic_init(asymcute_topic_t *t, const char *name) static asymcute_topic_t *_topic_init(asymcute_topic_t *t, const char *name)
{ {
uint16_t id = _topic_parse_pre(name); uint16_t id = _parse_predef_id(name);
if (id != 0) { if (id != 0) {
name = NULL; name = NULL;
} }
if (asymcute_topic_init(t, name, id) != ASYMCUTE_OK) { if (asymcute_topic_init(t, name, id) != ASYMCUTE_OK) {
return 1; return NULL;
} }
return 0; return t;
} }
static asymcute_topic_t *_topic_find(const char *name)
static int _topic_find(asymcute_topic_t *t, const char *name)
{ {
size_t len = strlen(name); uint16_t id = _parse_predef_id(name);
uint16_t id = _topic_parse_pre(name);
if ((id != 0) || (len == 2)) {
if (t) {
return _topic_init(t, name);
}
return 0;
}
/* need to find topic in list of registered ones */
for (unsigned i = 0; i < TOPIC_BUF_NUMOF; i++) { for (unsigned i = 0; i < TOPIC_BUF_NUMOF; i++) {
if (asymcute_topic_is_reg(&_topics[i]) && if ((asymcute_topic_is_predef(&_topics[i]) && (id == _topics[i].id)) ||
(strncmp(name, _topics[i].name, sizeof(_topics[i].name)) == 0)) { (strncmp(name, _topics[i].name, sizeof(_topics[i].name)) == 0)) {
if (t) { return &_topics[i];
*t = _topics[i];
}
return 0;
} }
} }
return 1; return NULL;
} }
static void _topics_clear(void) static void _topics_clear(void)
@ -151,6 +137,18 @@ static asymcute_topic_t *_topic_get_free(void)
return NULL; return NULL;
} }
static void _topic_print_info(const asymcute_topic_t *topic)
{
printf(" id: %u, name: %s", (unsigned)topic->id, topic->name);
if (asymcute_topic_is_short(topic)) {
printf(" (SHORT)");
}
else if (asymcute_topic_is_predef(topic)) {
printf(" (PREDEF)");
}
puts("");
}
static void _topic_print_help(void) static void _topic_print_help(void)
{ {
puts(" topic can be\n" puts(" topic can be\n"
@ -300,34 +298,35 @@ static int _cmd_reg(int argc, char **argv)
return 1; return 1;
} }
if (_topic_find(NULL, argv[1]) == 0) { asymcute_topic_t *t = _topic_find(argv[1]);
puts("success: topic already registered (or no registration needed)\n"); if ((t != NULL) && (asymcute_topic_is_reg(t))) {
puts("success: topic already registered");
return 0; return 0;
} }
/* get registration request context */
asymcute_req_t *req = _get_req_ctx();
if (req == NULL) {
puts("error: unable to allocate request context");
return 1;
}
/* find unused slot */ /* find unused slot */
asymcute_topic_t *t = NULL; if (t == NULL) {
for (unsigned i = 0; i < TOPIC_BUF_NUMOF; i++) { t = _topic_get_free();
if (!asymcute_topic_is_reg(&_topics[i])) {
t = &_topics[i];
break;
}
}
if (t == NULL) { if (t == NULL) {
puts("error: no empty slot left for storing the topic\n"); puts("error: no empty slot left for storing the topic\n");
return 1; return 1;
} }
if (_topic_init(t, argv[1]) == NULL) {
/* send registration request */
asymcute_req_t *req = _get_req_ctx();
if (req == NULL) {
return 1;
}
if (_topic_init(t, argv[1]) != 0) {
puts("error: unable to initialize topic"); puts("error: unable to initialize topic");
return 1; return 1;
} }
if (asymcute_register(&_connection, req, t) != ASYMCUTE_OK) { }
int res = asymcute_register(&_connection, req, t);
if (res != ASYMCUTE_OK) {
printf("res: %i\n", res);
puts("error: unable to send REGISTER request\n"); puts("error: unable to send REGISTER request\n");
return 1; return 1;
} }
@ -341,21 +340,18 @@ static int _cmd_unreg(int argc, char **argv)
return 1; return 1;
} }
unsigned i = 0; asymcute_topic_t *t = _topic_find(argv[1]);
for (; i < TOPIC_BUF_NUMOF; i++) { if (t) {
if (strcmp(argv[1], _topics[i].name) == 0) {
for (unsigned s = 0; s < SUB_CTX_NUMOF; s++) { for (unsigned s = 0; s < SUB_CTX_NUMOF; s++) {
if (_subscriptions[i].topic == &_topics[i]) { if (_subscriptions[s].topic == t) {
puts("error: topic used in active subscription"); puts("error: topic used in active subscription");
return 1; return 1;
} }
} }
memset(&_topics[i], 0, sizeof(asymcute_topic_t)); asymcute_topic_reset(t);
puts("success: removed local entry for given topic"); puts("success: removed local entry for given topic");
break;
} }
} else {
if (i == TOPIC_BUF_NUMOF) {
puts("error: unable to find topic in list of registered topics"); puts("error: unable to find topic in list of registered topics");
} }
@ -370,13 +366,6 @@ static int _cmd_pub(int argc, char **argv)
return 1; return 1;
} }
/* parse and register topic */
asymcute_topic_t t;
if (_topic_find(&t, argv[1]) != 0) {
puts("error: given topic is not registered");
return 1;
}
/* parse QoS level */ /* parse QoS level */
unsigned flags = 0; unsigned flags = 0;
int qos = _qos_parse(argc, argv, 3, &flags); int qos = _qos_parse(argc, argv, 3, &flags);
@ -388,12 +377,20 @@ static int _cmd_pub(int argc, char **argv)
/* get request context */ /* get request context */
asymcute_req_t *req = _get_req_ctx(); asymcute_req_t *req = _get_req_ctx();
if (req == NULL) { if (req == NULL) {
puts("error: unable to obtain request context");
return 1;
}
/* get topic */
asymcute_topic_t *t = _topic_find(argv[1]);
if (t == NULL) {
puts("error: given topic is not registered");
return 1; return 1;
} }
/* publish data */ /* publish data */
size_t len = strlen(argv[2]); size_t len = strlen(argv[2]);
if (asymcute_publish(&_connection, req, &t, argv[2], len, flags) != if (asymcute_publish(&_connection, req, t, argv[2], len, flags) !=
ASYMCUTE_OK) { ASYMCUTE_OK) {
puts("error: unable to send PUBLISH message"); puts("error: unable to send PUBLISH message");
return 1; return 1;
@ -424,6 +421,7 @@ static int _cmd_sub(int argc, char **argv)
/* get request context */ /* get request context */
asymcute_req_t *req = _get_req_ctx(); asymcute_req_t *req = _get_req_ctx();
if (req == NULL) { if (req == NULL) {
puts("error: unable to obtain request context");
return 1; return 1;
} }
@ -434,23 +432,39 @@ static int _cmd_sub(int argc, char **argv)
return 1; return 1;
} }
/* parse topic */ /* parse topic, see if it exists, otherwise take empty topic */
asymcute_topic_t *t = _topic_get_free(); asymcute_topic_t *t = _topic_find(argv[1]);
if (t == NULL) {
puts("info: given topic does not exist, creating it now");
t = _topic_get_free();
if (t == NULL) { if (t == NULL) {
puts("error: no free topic memory"); puts("error: no free topic memory");
return 1; return 1;
} }
if (_topic_init(t, argv[1]) != 0) { if (_topic_init(t, argv[1]) == 0) {
puts("error: unable to initialize topic"); puts("error: unable to initialize topic");
return 1; return 1;
} }
}
else {
if (!asymcute_topic_is_reg(t)) {
puts("error: given topic is not registered\n");
return 1;
}
}
printf("using req %p, sub %p\n", (void *)req, (void *)sub); int res = asymcute_subscribe(&_connection, req, sub, t, _on_pub_evt,
NULL, flags);
if (asymcute_subscribe(&_connection, req, sub, t, _on_pub_evt, NULL, flags) if (res != ASYMCUTE_OK) {
!= ASYMCUTE_OK) { if (!asymcute_topic_is_reg(t)) {
asymcute_topic_reset(t); asymcute_topic_reset(t);
}
if (res == ASYMCUTE_SUBERR) {
puts("error: already subscribed to given topic");
}
else {
puts("error: unable to send SUBSCRIBE request"); puts("error: unable to send SUBSCRIBE request");
}
return 1; return 1;
} }
@ -496,8 +510,12 @@ static int _cmd_info(int argc, char **argv)
for (unsigned i = 0; i < TOPIC_BUF_NUMOF; i++) { for (unsigned i = 0; i < TOPIC_BUF_NUMOF; i++) {
printf("topic #%2u - ", i); printf("topic #%2u - ", i);
if (asymcute_topic_is_reg(&_topics[i])) { if (asymcute_topic_is_reg(&_topics[i])) {
printf("[registered] id: %u, name: %s\n", printf("[registered] ");
(unsigned)_topics[i].id, _topics[i].name); _topic_print_info(&_topics[i]);
}
else if (asymcute_topic_is_init(&_topics[i])) {
printf("[initialized] ");
_topic_print_info(&_topics[i]);
} }
else { else {
puts("[unused]"); puts("[unused]");
@ -508,9 +526,8 @@ static int _cmd_info(int argc, char **argv)
for (unsigned i = 0; i < SUB_CTX_NUMOF; i++) { for (unsigned i = 0; i < SUB_CTX_NUMOF; i++) {
printf("sub #%2u - ", i); printf("sub #%2u - ", i);
if (asymcute_sub_active(&_subscriptions[i])) { if (asymcute_sub_active(&_subscriptions[i])) {
printf("[subscribed] id: %u, name: %s\n", printf("[subscribed]");
(unsigned)_subscriptions[i].topic->id, _topic_print_info(_subscriptions[i].topic);
_subscriptions[i].topic->name);
} }
else { else {
puts("[unused]"); puts("[unused]");

28
sys/include/net/asymcute.h
View File

@ -394,6 +394,34 @@ static inline bool asymcute_topic_is_reg(const asymcute_topic_t *topic)
return (topic->con != NULL); return (topic->con != NULL);
} }
/**
* @brief Check if a given topic is a short topic
*
* @param[in] topic topic to check
*
* @return true if topic is a short topic
* @return false if topic is not short topic
*/
static inline bool asymcute_topic_is_short(const asymcute_topic_t *topic)
{
assert(topic);
return ((topic->flags & MQTTSN_TIT_SHORT) != 0);
}
/**
* @brief Check if a given topic is a pre-defined topic
*
* @param[in] topic topic to check
*
* @return true if topic is pre-defined
* @return false if topic is not pre-defined
*/
static inline bool asymcute_topic_is_predef(const asymcute_topic_t *topic)
{
assert(topic);
return ((topic->flags & MQTTSN_TIT_PREDEF) != 0);
}
/** /**
* @brief Check if a given topic is initialized * @brief Check if a given topic is initialized
* *

32
sys/net/application_layer/asymcute/asymcute.c
View File

@ -165,7 +165,12 @@ static void _compile_sub_unsub(asymcute_req_t *req, asymcute_con_t *con,
req->data[pos] = type; req->data[pos] = type;
req->data[pos + 1] = sub->topic->flags; req->data[pos + 1] = sub->topic->flags;
byteorder_htobebufs(&req->data[pos + 2], req->msg_id); byteorder_htobebufs(&req->data[pos + 2], req->msg_id);
if (sub->topic->flags & MQTTSN_TIT_MASK) {
memcpy(&req->data[pos + 4], &sub->topic->id, 2);
}
else {
memcpy(&req->data[pos + 4], sub->topic->name, topic_len); memcpy(&req->data[pos + 4], sub->topic->name, topic_len);
}
req->data_len = (pos + 4 + topic_len); req->data_len = (pos + 4 + topic_len);
req->arg = sub; req->arg = sub;
} }
@ -474,20 +479,26 @@ static void _on_suback(asymcute_con_t *con, const uint8_t *data, size_t len)
} }
unsigned ret = ASYMCUTE_REJECTED; unsigned ret = ASYMCUTE_REJECTED;
if (data[7] == MQTTSN_ACCEPTED) {
/* parse and apply assigned topic id */ /* parse and apply assigned topic id */
asymcute_sub_t *sub = req->arg; asymcute_sub_t *sub = req->arg;
if (sub == NULL) { if (sub == NULL) {
return; return;
} }
if (data[7] == MQTTSN_ACCEPTED) {
/* do not assign a topic ID for short and predefined topics */
if (!(sub->topic->flags & MQTTSN_TIT_MASK)) {
sub->topic->id = byteorder_bebuftohs(&data[3]); sub->topic->id = byteorder_bebuftohs(&data[3]);
}
sub->topic->con = con; sub->topic->con = con;
/* insert subscription to connection context */ /* insert subscription to connection context */
sub->next = con->subscriptions; sub->next = con->subscriptions;
con->subscriptions = sub; con->subscriptions = sub;
ret = ASYMCUTE_SUBSCRIBED; ret = ASYMCUTE_SUBSCRIBED;
} }
else {
sub->topic = NULL;
}
/* notify the user */ /* notify the user */
mutex_unlock(&req->lock); mutex_unlock(&req->lock);
@ -668,12 +679,12 @@ int asymcute_topic_init(asymcute_topic_t *topic, const char *topic_name,
{ {
assert(topic); assert(topic);
size_t len = 0; size_t len = 0;
if (asymcute_topic_is_reg(topic)) { if (asymcute_topic_is_reg(topic)) {
return ASYMCUTE_REGERR; return ASYMCUTE_REGERR;
} }
if (topic_name == NULL) { if (topic_name == NULL) {
if ((topic_id == 0) || (topic_id == UINT16_MAX)) { if ((topic_id == 0) || (topic_id == UINT16_MAX)) {
return ASYMCUTE_OVERFLOW; return ASYMCUTE_OVERFLOW;
@ -692,8 +703,7 @@ int asymcute_topic_init(asymcute_topic_t *topic, const char *topic_name,
if (topic_name == NULL) { if (topic_name == NULL) {
topic->id = topic_id; topic->id = topic_id;
topic->flags = MQTTSN_TIT_PREDEF; topic->flags = MQTTSN_TIT_PREDEF;
memcpy(topic->name, &topic_id, 2); memcpy(topic->name, "..\0", 3);
topic->name[2] = '\0';
} }
else { else {
strncpy(topic->name, topic_name, sizeof(topic->name)); strncpy(topic->name, topic_name, sizeof(topic->name));
@ -823,6 +833,14 @@ int asymcute_register(asymcute_con_t *con, asymcute_req_t *req,
goto end; goto end;
} }
/* if we have a short or predefined topic, there is no need to send a
* registration message. We assign the connection right away */
if (topic->flags & MQTTSN_TIT_MASK) {
topic->con = con;
mutex_unlock(&req->lock);
goto end;
}
/* prepare topic */ /* prepare topic */
req->arg = topic; req->arg = topic;
size_t topic_len = strlen(topic->name); size_t topic_len = strlen(topic->name);
@ -936,13 +954,16 @@ int asymcute_subscribe(asymcute_con_t *con, asymcute_req_t *req,
ret = ASYMCUTE_GWERR; ret = ASYMCUTE_GWERR;
goto end; goto end;
} }
/* check if we are already subscribed to the given topic */ /* check if we are already subscribed to the given topic, but only if the
* topic was already registered */
if (asymcute_topic_is_reg(topic)) {
for (asymcute_sub_t *sub = con->subscriptions; sub; sub = sub->next) { for (asymcute_sub_t *sub = con->subscriptions; sub; sub = sub->next) {
if (asymcute_topic_equal(topic, sub->topic)) { if (asymcute_topic_equal(topic, sub->topic)) {
ret = ASYMCUTE_SUBERR; ret = ASYMCUTE_SUBERR;
goto end; goto end;
} }
} }
}
/* make sure request context is clear to be used */ /* make sure request context is clear to be used */
if (mutex_trylock(&req->lock) != 1) { if (mutex_trylock(&req->lock) != 1) {
ret = ASYMCUTE_BUSY; ret = ASYMCUTE_BUSY;
@ -953,6 +974,7 @@ int asymcute_subscribe(asymcute_con_t *con, asymcute_req_t *req,
sub->cb = callback; sub->cb = callback;
sub->arg = arg; sub->arg = arg;
sub->topic = topic; sub->topic = topic;
topic->flags |= flags;
/* send SUBSCRIBE message */ /* send SUBSCRIBE message */
_compile_sub_unsub(req, con, sub, MQTTSN_SUBSCRIBE); _compile_sub_unsub(req, con, sub, MQTTSN_SUBSCRIBE);