1 files changed, 269 insertions, 0 deletions

diff --git a/mfi-mqtt/mfi-mqtt.c b/mfi-mqtt/mfi-mqtt.c
new file mode 100644
index 0000000..25527bc
--- /dev/null
+++ b/mfi-mqtt/mfi-mqtt.c
@@ -0,0 +1,269 @@
+#include <errno.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include <unistd.h>
+#include <signal.h>
+#include <pthread.h>
+
+#include <json-c/json.h>
+#include <mosquitto.h>
+
+#include "mosquitto.h"
+#include "reporting.h"
+
+static volatile sig_atomic_t closing_time = 0;
+
+typedef struct control_message {
+    int output;
+    int state;
+} control_message;
+
+control_message * parse_control_message(const struct mosquitto_message *message) {
+    struct json_tokener *tok;
+    enum json_tokener_error jerr;
+    control_message *out_message = NULL;
+    json_object *msg, *state_key, *output_key;
+
+    tok = json_tokener_new();
+    msg = json_tokener_parse_ex(tok, message->payload, message->payloadlen);
+    jerr = json_tokener_get_error(tok);
+    if (jerr != json_tokener_success) {
+        fprintf(stderr, "Invalid message format: %s\n", json_tokener_error_desc(jerr));
+        goto cleanup;
+    }
+
+    output_key = json_object_object_get(msg, "output");
+    if (!output_key) {
+        fprintf(stderr, "Invalid message format: no output key\n");
+        goto cleanup;
+    }
+
+    state_key = json_object_object_get(msg, "state");
+    if (!state_key) {
+        fprintf(stderr, "Invalid message format: no state key\n");
+        goto cleanup;
+    }
+
+    out_message = calloc(sizeof(control_message), 1);
+    out_message->state = json_object_get_int(state_key);
+    out_message->output = json_object_get_int(output_key);
+    // Clamp the value to on or off
+    out_message->state = out_message->state > 0 ? 1 : 0;
+
+cleanup:
+    if (msg) json_object_put(msg);
+    json_tokener_free(tok);
+
+    return out_message;
+}
+
+void set_relay(control_message *msg) {
+    FILE *f;
+    char *basename;
+    char filename[255];
+
+    basename = "/proc/power/relay";
+
+    memset(filename, 0, sizeof(filename));
+    snprintf(filename, strlen(basename) + 2, "%s%i", basename, msg->output);
+
+    f = fopen(filename, "w");
+    if (!f) {
+        fprintf(stderr, "Failed to open relay file %s\n", filename);
+        return;
+    }
+
+    fprintf(f, "%i\n", msg->state);
+    fclose(f);
+}
+
+void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *message)
+{
+    control_message *msg;
+
+    msg = parse_control_message(message);
+    if (!msg) {
+        return;
+    }
+
+    if (msg->output > get_output_count() || msg->output <= 0) {
+        fprintf(stderr, "Invalid output number: %i\n", msg->output);
+        goto cleanup;
+    }
+
+    fprintf(stderr, "Set output %i to %i\n", msg->output, msg->state);
+    set_relay(msg);
+
+cleanup:
+    free(msg);
+}
+
+void my_connect_callback(struct mosquitto *mosq, void *obj, int result, int flags)
+{
+    char *topic;
+    char *prefix;
+    char hostname[255];
+
+    memset(hostname, 0, sizeof(hostname));
+    gethostname(hostname, 255);
+
+    prefix = "/mfi/devices";
+    topic = calloc(sizeof(char), strlen(prefix) + strlen(hostname) + 2);
+    sprintf(topic, "%s/%s", prefix, hostname);
+
+    fprintf(stderr, "Subscribed to topic %s\n", topic);
+
+        if (!result) {
+        mosquitto_subscribe(mosq, NULL, topic, 0);
+        } else {
+        fprintf(stderr, "%s\n", mosquitto_connack_string(result));
+                mosquitto_disconnect(mosq);
+        }
+}
+
+static void signal_handler(int signo) {
+    closing_time = 1;
+}
+
+void set_signal_handler() {
+    struct sigaction action;
+
+    action.sa_handler = signal_handler;
+    action.sa_flags = SA_SIGINFO | SA_RESTART;
+
+    if (sigaction(SIGINT, &action, NULL) == -1) {
+        perror("Error connecting signal");
+        exit(-1);
+    }
+}
+
+void mask_sig() {
+    sigset_t mask;
+    sigemptyset(&mask);
+    sigaddset(&mask, SIGINT);
+    pthread_sigmask(SIG_BLOCK, &mask, NULL);
+}
+
+void * control_thread(struct mosquitto *mosq) {
+    mask_sig();
+        mosquitto_loop_forever(mosq, 1000*86400, 1);
+    return NULL;
+}
+
+void * pinger_thread(struct mosquitto *mosq) {
+    power_statistics **stats;
+    int report_count;
+    char *output2;
+
+    mask_sig();
+
+    do {
+        if (closing_time) {
+            break;
+        }
+
+        report_count = get_stats_all_outputs(&stats);
+        output2 = format_report_all_outputs(stats, report_count);
+
+        mosquitto_publish(mosq, NULL, "/mfi/reports", strlen(output2), output2, 0, false);
+
+        free(output2);
+        free(stats);
+
+        sleep(2);
+    } while(true);
+
+    return NULL;
+}
+
+// Stop all the blinking!
+void * light_management_thread(void *arg) {
+    FILE *light_fp;
+    FILE *freq_fp;
+
+    do {
+        if (closing_time) {
+            break;
+        }
+
+        light_fp = fopen("/proc/led/status", "w");
+        if (light_fp) {
+            fprintf(light_fp, "1\n");
+            fclose(light_fp);
+        }
+
+        freq_fp = fopen("/proc/led/freq", "w");
+        if (freq_fp) {
+            fprintf(freq_fp, "0\n");
+            fclose(freq_fp);
+        }
+
+        sleep(1);
+    } while(true);
+
+    return NULL;
+}
+
+// Doing this the "C way" is a real pain in the ass, just shell out and forget
+// about it
+void cleanup_crap_processes() {
+    // Otherwise init will continue to respawn them
+    system("sed -i "
+        "-e '/ubnt-websockets/s/^/#/' "
+        "-e '/telnetd/s/^/#/' "
+        "-e '/mca[-d]/s/^/#/' "
+        "-e '/lighttpd/s/^/#/' "
+        "/etc/inittab");
+
+    system("kill -HUP 1");
+
+    // Most of these kill cleanly but a few are stubborn so don't ask, tell.
+    system("pkill -9 ubnt-websockets");
+    system("pkill -9 lighttpd");
+    system("pkill upnpd");
+    system("pkill telnetd");
+    system("pkill mca-monitor");
+    system("pkill mcad");
+    system("pkill avahi-daemon");
+}
+
+int main(int argc, char *argv[])
+{
+    struct mosquitto *mosq;
+    pthread_t pinger_thread_h, control_thread_h, light_management_thread_h;
+
+    cleanup_crap_processes();
+
+    set_signal_handler();
+
+    mosq = connect_to_broker("172.16.0.191", 1883);
+    if (!mosq) {
+        return 1;
+    }
+    fprintf(stderr, "Connected to broker\n");
+
+        mosquitto_connect_with_flags_callback_set(mosq, my_connect_callback);
+        mosquitto_message_callback_set(mosq, my_message_callback);
+
+    pthread_create(&pinger_thread_h, NULL, (void * (*)(void *))pinger_thread, mosq);
+    pthread_create(&control_thread_h, NULL, (void * (*)(void *))control_thread, mosq);
+    pthread_create(&light_management_thread_h, NULL, (void * (*)(void *))light_management_thread, NULL);
+
+    do {
+        sleep(2);
+        if (closing_time) {
+            fprintf(stderr, "Shutting down\n");
+
+            mosquitto_disconnect(mosq);
+            pthread_join(pinger_thread_h, NULL);
+            pthread_join(control_thread_h, NULL);
+            break;
+        }
+    } while(true);
+
+    shutdown_broker(mosq);
+
+        return 0;
+}