Added buggy implementation of wildcard subscription #, removed useless functions

Author: codepr

README.md

@@ -40,7 +40,7 @@ features expected from a MQTT broker.
 - [X] QoS 0, 1, 2 are handled
 - [X] Trie as underlying structure to handle topic hierarchies
 - [X] Periodic tasks like stats publishing
-- [ ] Wildcards on subscriptions (though simple to implement, will be added soon)
+- [X] Wildcards on subscriptions (though simple to implement, will be added soon)
 - [ ] QoS 1 and 2 tracking of pending clients and re-send
 - [ ] Session present check and handling
 - [ ] Authentication

src/network.c

@@ -210,44 +210,6 @@ int accept_connection(int serversock) {
     return clientsock;
 }
 
-/*
- * Create a socket and use it to connect to the specified host and port
- */
-int open_connection(const char *host, int port) {
-
-    struct sockaddr_in serveraddr;
-    struct hostent *server;
-
-    /* socket: create the socket */
-    int sfd = socket(AF_INET, SOCK_STREAM, 0);
-    if (sfd < 0)
-        goto err;
-
-    /* gethostbyname: get the server's DNS entry */
-    server = gethostbyname(host);
-    if (server == NULL)
-        goto err;
-
-    /* build the server's address */
-    bzero((char *) &serveraddr, sizeof(serveraddr));
-    serveraddr.sin_family = AF_INET;
-    bcopy((char *) server->h_addr,
-          (char *) &serveraddr.sin_addr.s_addr, server->h_length);
-    serveraddr.sin_port = htons(port);
-
-    /* connect: create a connection with the server */
-    if (connect(sfd, (const struct sockaddr *) &serveraddr,
-                sizeof(serveraddr)) < 0)
-        goto err;
-
-    return sfd;
-
-err:
-
-    perror("open_connection");
-    return -1;
-}
-
 /* Send all bytes contained in buf, updating sent bytes counter */
 ssize_t send_bytes(int fd, const unsigned char *buf, size_t len) {
 

src/network.h

@@ -100,9 +100,6 @@ int make_listen(const char *, const char *, int);
 /* Accept a connection and add it to the right epollfd */
 int accept_connection(int);
 
-/* Open a connection with a target host:port */
-int open_connection(const char *, int);
-
 struct evloop *evloop_create(int, int);
 
 void evloop_init(struct evloop *, int, int);

src/server.c

@@ -54,14 +54,19 @@ static struct sol sol;
  * Statistics topics, published every N seconds defined by configuration
  * interval
  */
-static const char *sys_topics[7] = {
-    "$SOL/broker/uptime",
-    "$SOL/broker/clients/connected",
-    "$SOL/broker/clients/disconnected",
-    "$SOL/broker/bytes/sent",
-    "$SOL/broker/bytes/received",
-    "$SOL/broker/messages/sent",
-    "$SOL/broker/messages/received"
+static const char *sys_topics[12] = {
+    "$SOL/",
+    "$SOL/broker/",
+    "$SOL/broker/clients/",
+    "$SOL/broker/bytes/",
+    "$SOL/broker/messages/",
+    "$SOL/broker/uptime/",
+    "$SOL/broker/clients/connected/",
+    "$SOL/broker/clients/disconnected/",
+    "$SOL/broker/bytes/sent/",
+    "$SOL/broker/bytes/received/",
+    "$SOL/broker/messages/sent/",
+    "$SOL/broker/messages/received/"
 };
 
 
@@ -280,9 +285,28 @@ static int disconnect_handler(struct closure *cb, union mqtt_packet *pkt) {
 }
 
 
+static void recursive_subscription(struct trie_node *node, void *arg) {
+
+    if (!node || !node->data)
+        return;
+
+    struct list_node *child = node->children->head;
+    for (; child; child = child->next)
+        recursive_subscription(child->data, arg);
+
+    struct topic *t = node->data;
+
+    struct subscriber *s = arg;
+
+    t->subscribers = list_push(t->subscribers, s);
+}
+
+
 static int subscribe_handler(struct closure *cb, union mqtt_packet *pkt) {
 
     struct sol_client *c = cb->obj;
+    bool wildcard = false;
+    bool alloced = false;
 
     /* Subscribe packets contains a list of topics and QoS tuples */
     for (unsigned i = 0; i < pkt->subscribe.tuples_len; i++) {
@@ -293,19 +317,43 @@ static int subscribe_handler(struct closure *cb, union mqtt_packet *pkt) {
          * Check if the topic exists already or in case create it and store in
          * the global map
          */
-        const char *topic_name = (const char *) pkt->subscribe.tuples[i].topic;
-        struct topic *t = sol_topic_get(&sol, topic_name);
+        char *topic = (char *) pkt->subscribe.tuples[i].topic;
+
+        sol_debug("\t%s (QoS %i)", topic, pkt->subscribe.tuples[i].qos);
+
+        /* Recursive subscribe to all children topics if the topic ends with "/#" */
+        if (topic[pkt->subscribe.tuples[i].topic_len - 1] == '#' &&
+            topic[pkt->subscribe.tuples[i].topic_len - 2] == '/') {
+            topic = remove_occur(topic, '#');
+            wildcard = true;
+        } else if (topic[pkt->subscribe.tuples[i].topic_len - 1] != '/') {
+            topic = sol_malloc(pkt->subscribe.tuples[i].topic_len + 2);
+            memcpy(topic, pkt->subscribe.tuples[i].topic,
+                   pkt->subscribe.tuples[i].topic_len);
+            topic[pkt->subscribe.tuples[i].topic_len] = '/';
+            topic[pkt->subscribe.tuples[i].topic_len + 1] = '\0';
+            alloced = true;
+        }
 
-        sol_debug("\t%s (QoS %i)", topic_name, pkt->subscribe.tuples[i].qos);
+        struct topic *t = sol_topic_get(&sol, topic);
 
         // TODO check for callback correctly set to obj
 
         if (!t) {
-            t = topic_create(sol_strdup(topic_name));
+            t = topic_create(sol_strdup(topic));
             sol_topic_put(&sol, t);
+        } else if (wildcard == true) {
+            struct subscriber *sub = sol_malloc(sizeof(*sub));
+            sub->client = cb->obj;
+            sub->qos = pkt->subscribe.tuples[i].qos;
+            trie_prefix_map_tuple(&sol.topics, topic,
+                                  recursive_subscription, sub);
         }
 
         topic_add_subscriber(t, cb->obj, pkt->subscribe.tuples[i].qos);
+
+        if (alloced)
+            sol_free(topic);
     }
 
     /*
@@ -374,17 +422,31 @@ static int publish_handler(struct closure *cb, union mqtt_packet *pkt) {
 
     info.messages_recv++;
 
+    char *topic = (char *) pkt->publish.topic;
+    bool alloced = false;
+
+    if (topic[pkt->publish.topiclen - 1] != '/') {
+        topic = sol_malloc(pkt->publish.topiclen + 2);
+        memcpy(topic, pkt->publish.topic, pkt->publish.topiclen);
+        topic[pkt->publish.topiclen] = '/';
+        topic[pkt->publish.topiclen + 1] = '\0';
+        alloced = true;
+    }
+
     /*
      * Retrieve the topic from the global map, if it wasn't created before,
      * create a new one with the name selected
      */
-    struct topic *t = sol_topic_get(&sol, (const char *) pkt->publish.topic);
+    struct topic *t = sol_topic_get(&sol, topic);
 
     if (!t) {
-        t = topic_create(sol_strdup((const char *) pkt->publish.topic));
+        t = topic_create(sol_strdup(topic));
         sol_topic_put(&sol, t);
     }
 
+    if (alloced == true)
+        sol_free(topic);
+
     // TODO Check for QoS of subscriber, it should override the PUBLISH one
 
     unsigned char *pub = pack_mqtt_packet(pkt, PUBLISH_TYPE);
@@ -822,16 +884,16 @@ static void publish_stats(struct evloop *loop, void *args) {
     unsigned char *pmrecv = (unsigned char *) &mrecv[0];
     unsigned char *putime = (unsigned char *) &utime[0];
 
-    publish_message(0, strlen(sys_topics[0]),
-                    sys_topics[0], strlen(utime), putime);
-    publish_message(0, strlen(sys_topics[1]),
-                    sys_topics[1], strlen(cclients), pcclients);
-    publish_message(0, strlen(sys_topics[3]),
-                    sys_topics[3], strlen(bsent), pbsent);
     publish_message(0, strlen(sys_topics[5]),
-                    sys_topics[5], strlen(msent), pmsent);
+                    sys_topics[5], strlen(utime), putime);
     publish_message(0, strlen(sys_topics[6]),
-                    sys_topics[6], strlen(mrecv), pmrecv);
+                    sys_topics[6], strlen(cclients), pcclients);
+    publish_message(0, strlen(sys_topics[8]),
+                    sys_topics[8], strlen(bsent), pbsent);
+    publish_message(0, strlen(sys_topics[10]),
+                    sys_topics[10], strlen(msent), pmsent);
+    publish_message(0, strlen(sys_topics[11]),
+                    sys_topics[11], strlen(mrecv), pmrecv);
 
 }
 
@@ -891,7 +953,7 @@ int start_server(const char *addr, const char *port) {
     generate_uuid(server_closure.closure_id);
 
     /* Generate stats topics */
-    for (int i = 0; i < 7; i++)
+    for (int i = 0; i < 12; i++)
         sol_topic_put(&sol, topic_create(sol_strdup(sys_topics[i])));
 
     struct evloop *event_loop = evloop_create(EPOLL_MAX_EVENTS, EPOLL_TIMEOUT);

src/trie.c

@@ -480,6 +480,24 @@ static void trie_prefix_map_func(struct trie_node *node,
 
 }
 
+/* Iterate through children of each node starting from a given node, applying
+   a defined function which take a struct trie_node as argument */
+static void trie_prefix_map_func2(struct trie_node *node,
+                                  void (*mapfunc)(struct trie_node *, void *), void *arg) {
+
+    if (trie_is_free_node(node)) {
+        mapfunc(node, arg);
+        return;
+    }
+
+    struct list_node *child = node->children->head;
+    for (; child; child = child->next)
+        trie_prefix_map_func2(child->data, mapfunc, arg);
+
+    mapfunc(node, arg);
+
+}
+
 /*
  * Apply a function to every key below a given prefix, if prefix is null the
  * function will be applied to all the trie
@@ -505,6 +523,30 @@ void trie_prefix_map(Trie *trie, const char *prefix,
     }
 }
 
+/* Apply a function to every key below a given prefix, if prefix is null the
+   function will be applied to all the trie */
+void trie_prefix_map_tuple(Trie *trie, const char *prefix,
+                           void (*mapfunc)(struct trie_node *, void *), void *arg) {
+
+    assert(trie);
+
+    if (!prefix) {
+        trie_prefix_map_func2(trie->root, mapfunc, arg);
+    } else {
+
+        // Walk the trie till the end of the key
+        struct trie_node *node = trie_node_find(trie->root, prefix);
+
+        // No complete key found
+        if (!node)
+            return;
+
+        // Check all possible sub-paths and add to count where there is a leaf
+        trie_prefix_map_func2(node, mapfunc, arg);
+    }
+}
+
+
 /* Release memory of a node while updating size of the trie */
 void trie_node_free(struct trie_node *node, size_t *size) {
 

src/trie.h

@@ -109,4 +109,7 @@ List *trie_prefix_find(const Trie *, const char *);
 void trie_prefix_map(Trie *, const char *, void (*mapfunc)(struct trie_node *));
 
 
+void trie_prefix_map_tuple(Trie *, const char *,
+                           void (*mapfunc)(struct trie_node *, void *), void *);
+
 #endif

src/util.c

@@ -117,19 +117,16 @@ int parse_int(const char *string) {
 }
 
 
-char *update_integer_string(char *str, int num) {
-
-    int n = parse_int(str);
-    n += num;
-    /*
-     * Check for realloc if the new value is "larger" then
-     * previous
-     */
-    char tmp[number_len(n) + 1];  // max size in bytes
-    sprintf(tmp, "%d", n);  // XXX Unsafe
-    size_t len = strlen(tmp);
-    str = sol_realloc(str, len + 1);
-    strncpy(str, tmp, len + 1);
+char *remove_occur(char *str, char c) {
+    char *p = str;
+    char *pp = str;
+
+    while (*p) {
+        *pp = *p++;
+        pp += (*pp != c);
+    }
+
+    *pp = '\0';
 
     return str;
 }

src/util.h

@@ -59,7 +59,7 @@ void *sol_realloc(void *, size_t);
 size_t malloc_size(void *);
 void sol_free(void *);
 char *sol_strdup(const char *);
-char *update_integer_string(char *, int);
+char *remove_occur(char *str, char c);
 
 size_t memory_used(void);