diff --git a/include/neuron/connection/mqtt_client.h b/include/neuron/connection/mqtt_client.h index faec6e75..91641db5 100644 --- a/include/neuron/connection/mqtt_client.h +++ b/include/neuron/connection/mqtt_client.h @@ -53,6 +53,20 @@ typedef enum { typedef struct neu_mqtt_client_s neu_mqtt_client_t; +/** + * Check that `topic_filter` is a valid MQTT topic filter. + */ +bool neu_mqtt_topic_filter_is_valid(const char *topic_filter); + +/** + * Check the given `topic_filter` matches the given `topic_name`. + * + * @Precondition `topic_filter` must be a valid MQTT topic filter. + * @Precondition `topic_name` must be a valid MQTT topic name. + */ +bool neu_mqtt_topic_filter_is_match(const char *topic_filter, + const char *topic_name); + typedef void (*neu_mqtt_client_connection_cb_t)(void *data); typedef void (*neu_mqtt_client_publish_cb_t)(int errcode, neu_mqtt_qos_e qos, char *topic, uint8_t *payload, diff --git a/src/connection/mqtt_client.c b/src/connection/mqtt_client.c index 5ef4bc09..75e3fe9b 100644 --- a/src/connection/mqtt_client.c +++ b/src/connection/mqtt_client.c @@ -33,6 +33,7 @@ #include #include "connection/mqtt_client.h" +#include "errcodes.h" #include "event/event.h" #include "utils/asprintf.h" #include "utils/time.h" @@ -179,6 +180,93 @@ static inline uint8_t neu_mqtt_version_to_nng_mqtt_version(neu_mqtt_version_e v) } } +bool neu_mqtt_topic_filter_is_valid(const char *topic_filter) +{ + if (NULL == topic_filter || '\0' == *topic_filter || + strlen(topic_filter) > 65535) { + return false; + } + + char c; + while ((c = *topic_filter++)) { + switch (c) { + case '#': + // '#' must be last character + return '\0' == *topic_filter; + case '+': + // '+' must occupy an entire level + if (*topic_filter && '/' != *topic_filter) { + return false; + } + break; + default: + for (; '/' != c; c = *topic_filter++) { + if ('#' == c || '+' == c) { + return false; + } else if ('\0' == c) { + return true; + } + } + } + } + + return true; +} + +static bool topic_filter_is_match(const char *topic_filter, + const char *topic_name, + uint32_t topic_name_len) +{ + // The Server MUST NOT match Topic Filters starting with a wildcard + // character (# or +) with Topic Names beginning with a $ character + // [MQTT-4.7.2-1]. + if ('$' == topic_name[0]) { + if ('$' != topic_filter[0]) { + return false; + } + } else if (0 == strcmp("#", topic_filter)) { + return true; + } + + uint32_t i = 0; + while (true) { + switch (*topic_filter) { + case '#': + return true; + case '+': + while (i < topic_name_len && '/' != topic_name[i]) { + ++i; + }; + ++topic_filter; + break; + case '/': + if (i < topic_name_len && '/' == topic_name[i]) { + ++i; + } else if ('#' != topic_filter[1]) { // '#' include parent level + return false; + } + ++topic_filter; + break; + case '\0': + return i == topic_name_len; + default: + for (; *topic_filter && '/' != *topic_filter; ++topic_filter, ++i) { + if (!(i < topic_name_len && *topic_filter == topic_name[i])) { + return false; + } + } + } + } + + return true; +} + +bool neu_mqtt_topic_filter_is_match(const char *topic_filter, + const char *topic_name) +{ + return topic_filter_is_match(topic_filter, topic_name, strlen(topic_name)); +} + static inline task_t *task_new(neu_mqtt_client_t *client) { task_t *task = calloc(1, sizeof(*task)); @@ -405,6 +493,26 @@ static inline void subscriptions_free(subscription_t *subscriptions) } } +static inline subscription_t * +subscription_find_match(subscription_t *subscriptions, const char *topic_name, + uint32_t topic_name_len) +{ + subscription_t *sub = NULL, *tmp = NULL; + HASH_FIND(hh, subscriptions, topic_name, topic_name_len, sub); + if (NULL == sub) { + // subscription wildcard matching + // TODO: linear scanning is enough for current usages, + // but may pose performance issues in the future + HASH_ITER(hh, subscriptions, sub, tmp) + { + if (topic_filter_is_match(sub->topic, topic_name, topic_name_len)) { + return sub; + } + } + } + return sub; +} + static int resub_cb(void *data) { neu_mqtt_client_t *client = data; @@ -517,10 +625,9 @@ static void recv_cb(void *arg) log(debug, "recv [%.*s, QoS%d] %" PRIu32 " bytes", (unsigned) topic_len, topic, (int) qos, payload_len); - // find matching subscription - // NOTE: does not support wildcards nng_mtx_lock(client->mtx); - HASH_FIND(hh, client->subscriptions, topic, topic_len, subscription); + subscription = + subscription_find_match(client->subscriptions, topic, topic_len); if (NULL != subscription) { task_t *task = client_alloc_task(client); if (NULL != task) { @@ -1375,6 +1482,10 @@ int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos, int rv = 0; subscription_t *subscription = NULL; + if (!neu_mqtt_topic_filter_is_valid(topic)) { + return NEU_ERR_MQTT_SUBSCRIBE_FAILURE; + } + nng_mtx_lock(client->mtx); if (NULL == client->recv_aio) { if ((rv = nng_aio_alloc(&client->recv_aio, recv_cb, client)) != 0) { diff --git a/tests/ut/CMakeLists.txt b/tests/ut/CMakeLists.txt index f4feb6a2..88340072 100644 --- a/tests/ut/CMakeLists.txt +++ b/tests/ut/CMakeLists.txt @@ -64,6 +64,13 @@ target_include_directories(rolling_counter_test PRIVATE ) target_link_libraries(rolling_counter_test neuron-base gtest_main gtest) +add_executable(mqtt_client_test mqtt_client_test.cc) +target_include_directories(mqtt_client_test PRIVATE + ${CMAKE_SOURCE_DIR}/src + ${CMAKE_SOURCE_DIR}/include +) +target_link_libraries(mqtt_client_test neuron-base gtest_main gtest) + include(GoogleTest) gtest_discover_tests(json_test) gtest_discover_tests(http_test) @@ -73,3 +80,4 @@ gtest_discover_tests(tag_sort_test) gtest_discover_tests(modbus_test) gtest_discover_tests(async_queue_test) gtest_discover_tests(rolling_counter_test) +gtest_discover_tests(mqtt_client_test) diff --git a/tests/ut/mqtt_client_test.cc b/tests/ut/mqtt_client_test.cc new file mode 100644 index 00000000..0f7fab21 --- /dev/null +++ b/tests/ut/mqtt_client_test.cc @@ -0,0 +1,129 @@ +#include + +#include "connection/mqtt_client.h" +#include "utils/log.h" + +zlog_category_t *neuron = NULL; + +TEST(MQTTClientTest, test_topic_filter_is_valid) +{ + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid(NULL)); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("")); + + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport/tennis#")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport/tennis/#/ranking")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("#")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/#")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/tennis/player/#")); + + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport+")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport/+player")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("+")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("+/+")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/+/player")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("+/tennis/#")); + + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("$SYS")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("$SYS/#")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/tennis")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/tennis/player")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_valid("sport/tennis/player/ranking")); +} + +TEST(MQTTClientTest, test_topic_filter_is_match) +{ + const char *filter = "#"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "$sport")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player")); + + filter = "sport/tennis/player/#"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ(false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/play")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/")); + ASSERT_EQ( + true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match( + filter, "sport/tennis/player/ranking/score/wimbledon")); + + filter = "+"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "$finance")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "/finance")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "finance")); + + filter = "/+"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "finance")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "/finance")); + + filter = "+/+"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "$finance")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "finance")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "/finance")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + + filter = "sport/+"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis/")); + ASSERT_EQ(false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/ranking")); + + filter = "sport/tennis/+"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ( + false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player1")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player2")); + + filter = "sport/+/player"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis/")); + ASSERT_EQ(false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/ranking")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player")); + ASSERT_EQ(false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/")); + ASSERT_EQ( + false, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking")); + + filter = "+/tennis/#"; + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/")); + ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/baseball")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis")); + ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis/")); + ASSERT_EQ(true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player")); + ASSERT_EQ( + true, + neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking")); +} + +int main(int argc, char **argv) +{ + zlog_init("./config/dev.conf"); + neuron = zlog_get_category("neuron"); + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +}