add: mongoose for connections and mqtt

include/database.h

@@ -14,4 +14,6 @@
 void
 database_setup(MDB_env **mdb_env, config_t *config);
 
+extern MDB_env *global_mdb_env;
+
 #endif /* CONTROLLER_DATABASE_H */

include/handlers.h

@@ -1,7 +1,7 @@
 #ifndef CONTROLLER_HANDLERS_H
 #define CONTROLLER_HANDLERS_H
 
-#include <poll.h>
+#include <mongoose.h>
 
 #include <models/controller.h>
 
@@ -12,7 +12,7 @@
  * @param controller Controller to use for answering command
  */
 void
-handler_command(int fd, controller_t *controller);
+handler_command(struct mg_connection *c, int ev, void *ev_data);
 
 /**
  * @brief Handle the discovery processing
@@ -21,12 +21,12 @@ handler_command(int fd, controller_t *controller);
  * @param controller Controller to use for answering discovery
  */
 void
-handler_discovery(int fd, controller_t *controller);
+handler_discovery(struct mg_connection *c, int ev, void *ev_data);
 
 void
-handler_loop(controller_t *this_controller);
+handler_mqtt(struct mg_connection *c, int ev, void *ev_data);
 
 void
-handler_poll(struct pollfd *fds, controller_t *controller, MDB_env *mdb_env);
+handler_loop(struct mg_connection *c_mqtt);
 
 #endif /* CONTROLLER_HANDLERS_H */

include/models/controller.h

@@ -107,4 +107,6 @@ controller_free(controller_t *controller);
 void
 controller_debug(controller_t *controller);
 
+extern controller_t *global_controller;
+
 #endif //CONTROLLER_CONTROLLER_H

include/models/relay.h

@@ -11,6 +11,7 @@
 typedef struct
 {
     uint8_t number;
+    int is_on;
     char name[MAX_NAME_LENGTH + 1];
     schedule_t *schedules[7];
 } relay_t;

src/handlers/command.c

@@ -11,6 +11,7 @@
 #include <errno.h>
 
 #include <logger.h>
+#include <database.h>
 #include <handlers.h>
 #include <helpers.h>
 #include <enums.h>
@@ -68,75 +69,59 @@ handler_command_set_relay_name(mpack_node_t map, controller_t *controller)
 }
 
 void
-handler_command(int fd, controller_t *controller)
+handler_command(struct mg_connection *c, int ev, void *ev_data)
 {
-    struct sockaddr_storage their_addr;
+    (void)ev_data;
-    socklen_t addr_size;
+    if(ev == MG_EV_RECV)
-    int client_fd;
-
-    addr_size = sizeof(their_addr);
-
-    if((client_fd = accept(fd, (struct sockaddr *) &their_addr, &addr_size)) < 0)
     {
-        LOG_ERROR("could not accept client: %s\n", strerror(errno));
+        uint32_t payload_length = *((uint32_t*)c->recv_mbuf.buf);
-        return;
+        LOG_DEBUG("payload_length %d\n", payload_length);
+
+        if(c->recv_mbuf.len < payload_length + sizeof(payload_length))
+        {
+            return;
+        }
+
+        char *payload = c->recv_mbuf.buf + sizeof(payload_length);
+
+        mpack_tree_t tree;
+        mpack_tree_init_data(&tree, payload, payload_length);
+        mpack_tree_parse(&tree);
+        mpack_node_t root = mpack_tree_root(&tree);
+
+        uint8_t command_code = mpack_node_u8(mpack_node_map_uint(root, COMMAND_MAPPING_CODE));
+
+        LOG_INFO("received command %d\n", command_code);
+
+        switch(command_code)
+        {
+            case COMMAND_CODE_GET_TIME:
+                break;
+            case COMMAND_CODE_GET_ID:
+                break;
+            case COMMAND_CODE_SET_NAME:
+                handler_command_set_name(root, global_controller);
+                break;
+            case COMMAND_CODE_GET_NAME:
+                break;
+            case COMMAND_CODE_SET_SCHEDULE:
+                handler_command_set_schedule(root, global_controller);
+                break;
+            case COMMAND_CODE_GET_SCHEDULE:
+                break;
+            case COMMAND_CODE_SET_RELAY_NAME:
+                handler_command_set_relay_name(root, global_controller);
+                break;
+            case COMMAND_CODE_GET_RELAY_NAME:
+                break;
+            default:
+                LOG_ERROR("received invalid command\n");
+        }
+
+        if(mpack_tree_destroy(&tree) != mpack_ok)
+        {
+            LOG_WARN("error when destroying mpack tree\n");
+        }
     }
-
+    controller_save(global_controller, global_mdb_env);
-    uint32_t payload_length;
-
-    if(recv(client_fd, &payload_length, sizeof(payload_length), 0) <= 0)
-    {
-        LOG_ERROR("unable to receive header: %s\n", strerror(errno));
-        return;
-    }
-
-    void *payload = malloc((payload_length + 1));
-    ssize_t bytes_transferred;
-
-    if((bytes_transferred = recv(client_fd, payload, payload_length, 0)) <= 0)
-    {
-        LOG_ERROR("unable to receive payload: %s\n", strerror(errno));
-        return;
-    }
-
-    mpack_tree_t tree;
-    mpack_tree_init_data(&tree, payload, payload_length);
-    mpack_tree_parse(&tree);
-    mpack_node_t root = mpack_tree_root(&tree);
-
-    uint8_t command_code = mpack_node_u8(mpack_node_map_uint(root, COMMAND_MAPPING_CODE));
-
-    LOG_INFO("received command %d\n", command_code);
-
-    switch(command_code)
-    {
-        case COMMAND_CODE_GET_TIME:
-            break;
-        case COMMAND_CODE_GET_ID:
-            break;
-        case COMMAND_CODE_SET_NAME:
-            handler_command_set_name(root, controller);
-            break;
-        case COMMAND_CODE_GET_NAME:
-            break;
-        case COMMAND_CODE_SET_SCHEDULE:
-            handler_command_set_schedule(root, controller);
-            break;
-        case COMMAND_CODE_GET_SCHEDULE:
-            break;
-        case COMMAND_CODE_SET_RELAY_NAME:
-            handler_command_set_relay_name(root, controller);
-            break;
-        case COMMAND_CODE_GET_RELAY_NAME:
-            break;
-        default:
-            LOG_ERROR("received invalid command\n");
-    }
-
-    if(mpack_tree_destroy(&tree) != mpack_ok)
-    {
-        LOG_WARN("error when destroying mpack tree\n");
-    }
-    free(payload);
-    close(client_fd);
 }

src/handlers/discovery.c

@@ -11,79 +11,85 @@
 #include <errno.h>
 
 #include <logger.h>
+#include <mongoose.h>
 #include <handlers.h>
 #include <helpers.h>
 #include <mpack.h>
 #include <enums.h>
 
 void
-handler_discovery(int fd, controller_t *controller)
+handler_discovery(struct mg_connection *c, int ev, void *ev_data)
 {
-    ssize_t bytes_transferred;
+    (void)ev_data;
-    uint16_t discovery_answer_port;
+    if(ev == MG_EV_RECV)
-    struct sockaddr_in si_other;
-    socklen_t slen = sizeof(si_other);
-
-    if((bytes_transferred = recvfrom(fd, &discovery_answer_port, sizeof(discovery_answer_port), 0, (struct sockaddr *) &si_other, &slen)) <= 0)
     {
-        LOG_ERROR("received invalid discovery from %s\n", inet_ntoa(si_other.sin_addr));
+        uint16_t discovery_answer_port;
-        return;
+        char ip_buf[32];
-    }
+        mg_conn_addr_to_str(c, ip_buf, sizeof(ip_buf), MG_SOCK_STRINGIFY_IP | MG_SOCK_STRINGIFY_REMOTE);
-    LOG_INFO("received discovery from %s:%d\n", inet_ntoa(si_other.sin_addr), discovery_answer_port);
 
-    if(discovery_answer_port == 0)
+        if(c->recv_mbuf.len != sizeof(discovery_answer_port))
-    {
+        {
-        LOG_ERROR("invalid port received\n");
+            LOG_ERROR("received invalid discovery from %s\n", ip_buf);
-        return;
+            return;
-    }
+        }
+        discovery_answer_port = *((uint16_t*)c->recv_mbuf.buf);
 
-    char* payload;
+        LOG_INFO("received discovery from %s:%d\n", ip_buf, discovery_answer_port);
-    size_t payload_size;
-    mpack_writer_t writer;
-    mpack_writer_init_growable(&writer, &payload, &payload_size);
 
-    mpack_start_map(&writer, 4);
+        if(discovery_answer_port == 0)
-    mpack_write_uint(&writer, DISCOVERY_MAPPING_ID);
+        {
-    mpack_write_bin(&writer, (char*)controller->id, sizeof(uuid_t));
+            LOG_ERROR("invalid port received\n");
-    mpack_write_uint(&writer, DISCOVERY_MAPPING_COMMAND_PORT);
+            return;
-    mpack_write_u16(&writer, controller->command_port);
+        }
-    mpack_write_uint(&writer, DISCOVERY_MAPPING_RELAY_COUNT);
-    mpack_write_u8(&writer, controller->relay_count);
-    mpack_write_uint(&writer, DISCOVERY_MAPPING_NAME);
-    mpack_write_cstr(&writer, controller->name);
-    mpack_finish_map(&writer);
 
-    // finish writing
+        char* payload;
-    if(mpack_writer_destroy(&writer) != mpack_ok)
+        size_t payload_size;
-    {
+        mpack_writer_t writer;
-        LOG_ERROR("error writing discovery answer payload\n");
+        mpack_writer_init_growable(&writer, &payload, &payload_size);
-        return;
 
-    }
+        mpack_start_map(&writer, 4);
+        mpack_write_uint(&writer, DISCOVERY_MAPPING_ID);
+        mpack_write_bin(&writer, (char*)global_controller->id, sizeof(uuid_t));
+        mpack_write_uint(&writer, DISCOVERY_MAPPING_COMMAND_PORT);
+        mpack_write_u16(&writer, global_controller->command_port);
+        mpack_write_uint(&writer, DISCOVERY_MAPPING_RELAY_COUNT);
+        mpack_write_u8(&writer, global_controller->relay_count);
+        mpack_write_uint(&writer, DISCOVERY_MAPPING_NAME);
+        mpack_write_cstr(&writer, global_controller->name);
+        mpack_finish_map(&writer);
+
+        // finish writing
+        if(mpack_writer_destroy(&writer) != mpack_ok)
+        {
+            LOG_ERROR("error writing discovery answer payload\n");
+            return;
+
+        }
+
+        size_t bytes_transferred;
+        int fd_answer = helper_connect_tcp_server(ip_buf, discovery_answer_port);
+        if(fd_answer == -1)
+        {
+            LOG_ERROR("error during connecting\n");
+            free(payload);
+            return;
+        }
+
+        if((bytes_transferred = send(fd_answer, &payload_size, sizeof(payload_size), 0)) <= 0)
+        {
+            LOG_ERROR("error during sending\n");
+            free(payload);
+            close(fd_answer);
+            return;
+        }
+        if((bytes_transferred = send(fd_answer, payload, payload_size, 0)) <= 0)
+        {
+            LOG_ERROR("error during sending\n");
+            free(payload);
+            close(fd_answer);
+            return;
+        }
 
-    int fd_answer = helper_connect_tcp_server(inet_ntoa(si_other.sin_addr), discovery_answer_port);
-    if(fd_answer == -1)
-    {
-        LOG_ERROR("error during connecting\n");
         free(payload);
-        return;
     }
-
-    if((bytes_transferred = send(fd_answer, &payload_size, sizeof(payload_size), 0)) <= 0)
-    {
-        LOG_ERROR("error during sending\n");
-        free(payload);
-        close(fd_answer);
-        return;
-    }
-    if((bytes_transferred = send(fd_answer, payload, payload_size, 0)) <= 0)
-    {
-        LOG_ERROR("error during sending\n");
-        free(payload);
-        close(fd_answer);
-        return;
-    }
-
-    free(payload);
-    close(fd_answer);
 }

src/handlers/loop.c

@@ -2,6 +2,7 @@
 #include <string.h>
 #include <stdio.h>
 #include <time.h>
+#include <uuid/uuid.h>
 
 #include <logger.h>
 #include <models/controller.h>
@@ -13,20 +14,35 @@
 #include <wiring_debug.h>
 
 void
-handler_loop(controller_t *controller)
+handler_loop(struct mg_connection *c_mqtt)
 {
+    char topic_buf[100];
+    char payload_buf[2];
+    char controller_uid[UUID_STR_LEN];
+    uuid_unparse(global_controller->id, controller_uid);
+
     time_t timestamp = time(NULL);
     struct tm *time_struct = localtime(&timestamp);
     LOG_DEBUG("===== IDLE LOOP START =====\n");
-    for(uint_fast8_t i = 0; i < controller->relay_count; ++i)
+    for(uint_fast8_t i = 0; i < global_controller->relay_count; ++i)
     {
-        relay_t *relay = controller->relays[i];
+        relay_t *relay = global_controller->relays[i];
         int is_active = 0;
         if(relay_is_active(relay, time_struct))
         {
             LOG_DEBUG("relay %d is active\n", i);
             is_active = 1;
         }
+
+        if(relay->is_on != is_active)
+        {
+            sprintf(topic_buf, "controller/%s/relay/%u", controller_uid, i);
+            sprintf(payload_buf, "%u", is_active);
+            mg_mqtt_publish(c_mqtt, topic_buf, 0, MG_MQTT_QOS(0), payload_buf, strlen(payload_buf));
+        }
+
+        relay->is_on = is_active;
+
         if(global_config.relay_configs[i].inverted)
         {
             is_active = !is_active;

src/handlers/mqtt.c

@@ -0,0 +1,37 @@
+#include <logger.h>
+#include <handlers.h>
+#include <models/controller.h>
+
+void
+handler_mqtt(struct mg_connection *nc, int ev, void *p) {
+  struct mg_mqtt_message *msg = (struct mg_mqtt_message *) p;
+  (void) nc;
+
+  if (ev != MG_EV_POLL) LOG_DEBUG("USER HANDLER GOT EVENT %d\n", ev);
+
+  switch (ev) {
+    case MG_EV_CONNECT: {
+      struct mg_send_mqtt_handshake_opts opts;
+      memset(&opts, 0, sizeof(opts));
+      // TODO add password
+
+      mg_set_protocol_mqtt(nc);
+      mg_send_mqtt_handshake_opt(nc, global_controller->name, opts);
+      break;
+    }
+    case MG_EV_MQTT_CONNACK:
+      if (msg->connack_ret_code != MG_EV_MQTT_CONNACK_ACCEPTED) {
+        LOG_DEBUG("Got mqtt connection error: %d\n", msg->connack_ret_code);
+        exit(1);
+      }
+      break;
+    case MG_EV_MQTT_PUBACK:
+      LOG_DEBUG("Message publishing acknowledged (msg_id: %d)\n", msg->message_id);
+      break;
+    case MG_EV_MQTT_SUBACK:
+      LOG_DEBUG("Subscription acknowledged, forwarding to '/test'\n");
+      break;
+    case MG_EV_CLOSE:
+      LOG_DEBUG("Connection closed\n");
+  }
+}

src/handlers/poll.c

@@ -1,50 +0,0 @@
-#include <stdlib.h>
-#include <string.h>
-#include <stdio.h>
-#include <time.h>
-#include <lmdb.h>
-#include <unistd.h>
-#include <sys/socket.h>
-#include <poll.h>
-#include <signal.h>
-
-#include <logger.h>
-#include <models/controller.h>
-#include <database.h>
-#include <config.h>
-#include <constants.h>
-#include <handlers.h>
-#include <drivers.h>
-#include <enums.h>
-#include <helpers.h>
-#include <wiringPi.h>
-#include <piFace.h>
-#include <wiring_debug.h>
-
-void
-handler_poll(struct pollfd *fds, controller_t *controller, MDB_env *mdb_env)
-{
-    /* An event on one of the fds has occurred. */
-    for(int i = 0; i < POLL_FDS_COUNT; i++) {
-        if(fds[i].revents & POLLIN)
-        {
-            /* data may be read on device number i. */
-            LOG_DEBUG("fd %i may read data\n", fds[i].fd);
-            switch(i)
-            {
-                case POLL_FDS_DISCOVERY:
-                    handler_discovery(fds[i].fd, controller);
-                    break;
-                case POLL_FDS_COMMAND:
-                    handler_command(fds[i].fd, controller);
-                    controller_save(controller, mdb_env);
-                    break;
-            }
-        }
-        if(fds[i].revents & POLLHUP)
-        {
-            /* A hangup has occurred on device number i. */
-            LOG_DEBUG("fd %i got closed\n", fds[i].fd);
-        }
-    }
-}

src/main.c

@@ -3,12 +3,10 @@
 #include <stdio.h>
 #include <time.h>
 #include <lmdb.h>
-#include <unistd.h>
-#include <sys/socket.h>
-#include <poll.h>
 #include <signal.h>
 
 #include <logger.h>
+#include <mongoose.h>
 #include <models/controller.h>
 #include <database.h>
 #include <config.h>
@@ -24,26 +22,28 @@
 #include <confini.h>
 
 config_t global_config;
+controller_t *global_controller;
+MDB_env *global_mdb_env;
 
-static MDB_env *mdb_env;
+static struct mg_mgr mgr;
-static controller_t *this_controller;
-static struct pollfd poll_fds[POLL_FDS_COUNT];
 
 static void
 terminate(int signum)
 {
     LOG_INFO("terminating controller (%d)\n", signum);
 
-    for(int i = 0; i < POLL_FDS_COUNT; ++i)
+    LOG_DEBUG("freeing mongoose manager\n");
-    {
+    mg_mgr_free(&mgr);
-        close(poll_fds[i].fd);
-    }
 
-    mdb_env_close(mdb_env);
+    LOG_DEBUG("closing database\n");
+    mdb_env_close(global_mdb_env);
 
-    controller_free(this_controller);
+    LOG_DEBUG("freeing global controller\n");
+    controller_free(global_controller);
 
+    LOG_DEBUG("freeing database config\n");
     free(global_config.database);
+    LOG_DEBUG("freeing relay configs config\n");
     free(global_config.relay_configs);
 
     exit(signum);
@@ -93,18 +93,27 @@ main(int argc, const char** argv)
         LOG_WARN("this system is not using 8-bit time\n");
     }
 
-    /******************** SETUP DATABASE AND THIS CONTROLLER ********************/
 
-    database_setup(&mdb_env, &global_config);
+    /******************** SETUP DATABASE, SOCKETS AND THIS CONTROLLER ********************/
 
-    this_controller = controller_load(mdb_env);
+    mg_mgr_init(&mgr, NULL);
 
-    int fd_discovery = helper_open_discovery_socket(this_controller->discovery_port);
+    database_setup(&global_mdb_env, &global_config);
-    int fd_command = helper_bind_tcp_server("0.0.0.0", this_controller->command_port, 128);
 
-    this_controller->command_port = helper_get_port(fd_command);
+    global_controller = controller_load(global_mdb_env);
 
-    controller_save(this_controller, mdb_env);
+    char address[100];
+    sprintf(address, "udp://0.0.0.0:%u", global_controller->discovery_port);
+    struct mg_connection *c_discovery = mg_bind(&mgr, address, handler_discovery);
+    sprintf(address, "tcp://0.0.0.0:%u", global_controller->command_port);
+    struct mg_connection *c_command = mg_bind(&mgr, address, handler_command);
+    sprintf(address, "tcp://localhost:%u", 1883);
+    struct mg_connection *c_mqtt = mg_connect(&mgr, address, handler_mqtt);
+
+    (void)c_discovery; // still unused
+    global_controller->command_port = helper_get_port(c_command->sock);
+
+    controller_save(global_controller, global_mdb_env);
 
     
     /******************** SETUP WIRINGPI ********************/
@@ -112,7 +121,7 @@ main(int argc, const char** argv)
     wiringPiSetup();
     piFaceSetup(PIFACE_GPIO_BASE);
 
-    for(uint_fast8_t i = 0; i < this_controller->relay_count; ++i)
+    for(uint_fast8_t i = 0; i < global_controller->relay_count; ++i)
     {
         if(global_config.relay_configs[i].driver == RELAY_DRIVER_GPIO)
         {
@@ -121,25 +130,11 @@ main(int argc, const char** argv)
     }
 
 
-    /******************** SETUP SOCKETS ********************/
-
-    int timeout_msecs = ACCEPT_TIMEOUT_MSECONDS;
-    int ret;
-
-    /* Open STREAMS device. */
-    poll_fds[POLL_FDS_DISCOVERY].fd = fd_discovery;
-    poll_fds[POLL_FDS_DISCOVERY].events = POLLIN;
-    LOG_DEBUG("setup fd_discovery as %i on index %i\n", fd_discovery, POLL_FDS_DISCOVERY);
-    poll_fds[POLL_FDS_COMMAND].fd = fd_command;
-    poll_fds[POLL_FDS_COMMAND].events = POLLIN;
-    LOG_DEBUG("setup fd_command as %i on index %i\n", fd_command, POLL_FDS_COMMAND);
-
-
     /******************** CHECK FOR TESTING RUN ********************/
 
     if(global_config.run_type == RUN_TYPE_TEST)
     {
-        runner_test(this_controller);
+        runner_test(global_controller);
         terminate(0);
     }
     
@@ -148,21 +143,11 @@ main(int argc, const char** argv)
 
     for(;;)
     {
-        ret = poll(poll_fds, POLL_FDS_COUNT, timeout_msecs);
+        mg_mgr_poll(&mgr, 1000);
-
+        handler_loop(c_mqtt);
-        if(ret == 0)
-        {
-            handler_loop(this_controller);
-        }
-        if(ret > 0)
-        {
-            handler_poll(poll_fds, this_controller, mdb_env);
-        }
     }
 
-    close(fd_discovery);
+    terminate(0);
-
-    mdb_env_close(mdb_env);
 
     return 0;
 }

src/models/relay.c

@@ -14,6 +14,8 @@ relay_create(uint8_t number)
     new_relay->number = number;
     new_relay->name[0] = '\0';
 
+    new_relay->is_on = -1;
+
     uuid_t off_id;
     memset(off_id, 0, sizeof(uuid_t));
     memcpy(off_id, "off", 3);

src/models/relay_load.c

@@ -65,6 +65,7 @@ relay_load(MDB_env *mdb_env, uint8_t num)
 
     new_relay = malloc(sizeof(relay_t));
     new_relay->number = num;
+    new_relay->is_on = -1;
 
     MDB_val value;