in_mqtt: fix connection handler and other minor fixes

Signed-off-by: Eduardo Silva <[email protected]>

Author: edsiper

plugins/in_mqtt/mqtt.c

@@ -102,9 +102,9 @@ int in_mqtt_collect(struct flb_config *config, void *in_context)
     flb_debug("[mqtt] new TCP connection arrived FD=%i", fd);
     conn = mqtt_conn_add(fd, ctx);
     if (!conn) {
-        return 0;
+        return -1;
     }
-    return -1;
+    return 0;
 }
 
 /* Plugin reference */

plugins/in_mqtt/mqtt_config.c

@@ -31,6 +31,7 @@ struct flb_in_mqtt_config *mqtt_config_init(struct mk_rconf *conf)
     struct flb_in_mqtt_config *config;
 
     config = malloc(sizeof(struct flb_in_mqtt_config));
+    memset(config, '\0', sizeof(struct flb_in_mqtt_config));
 
     if (conf) {
       section = mk_rconf_section_get(conf, "MQTT");

plugins/in_mqtt/mqtt_conn.c

@@ -79,7 +79,7 @@ struct mqtt_conn *mqtt_conn_add(int fd, struct flb_in_mqtt_config *ctx)
     /* Set data for the event-loop */
     event = &conn->event;
     event->fd           = fd;
-    event->type         = MK_EVENT_CONNECTION;
+    event->type         = FLB_ENGINE_EV_CUSTOM;
     event->mask         = MK_EVENT_EMPTY;
     event->handler      = mqtt_conn_event;
     event->status       = MK_EVENT_NONE;

plugins/in_mqtt/mqtt_prot.c

@@ -161,7 +161,7 @@ static int mqtt_handle_publish(struct mqtt_conn *conn)
     char buf[4];
 
     /*
-     * DUP: we skip dplicated messages.
+     * DUP: we skip duplicated messages.
      * QOS: We process this.
      * Retain: skipped
      */