.gitignore

@@ -1,4 +1,3 @@
 build/
 sdkconfig
 sdkconfig.old
-.cache

main/lib/meshtalos.c

@@ -2,12 +2,12 @@
 #include "lib/mongoose.h"
 #include <string.h>
 
-uint8_t image_buf[IMAGE_SIZE] = {0};
+uint8_t image_buf[152 * 296] = {0};
 bool flag_refresh = true;
 
 static void ev_handler(struct mg_connection *c, int ev, void *ev_data) {
 
-  // struct Storage *storage = (struct Storage *)(ev_data);
+  struct Storage *storage = (struct Storage *)(ev_data);
 
   if (ev == MG_EV_HTTP_MSG) { // New HTTP request received
     struct mg_http_message *hm =

main/lib/meshtalos.h

@@ -12,8 +12,6 @@
 #include "./storage.h"
 
 #define TOPIC_LEN 20
-#define IMAGE_SIZE 152 * 296
-#define IMAGE_BLK_SIZE 1024
 
 struct Meshtalos {
   struct Storage *storage;
@@ -32,7 +30,7 @@ struct sub {
 extern struct sub *sub_list;
 extern QueueHandle_t sub_jobs;
 extern SemaphoreHandle_t sub_mutex;
-extern uint8_t image_buf[IMAGE_SIZE];
+extern uint8_t image_buf[152 * 296];
 
 struct Meshtalos *mtsh_init();
 void mtsh_listen(struct Meshtalos *mtsh, const char *http_endpoint);

main/lib/publish.c

@@ -11,38 +11,56 @@ struct sub *sub_list = NULL;
 SemaphoreHandle_t sub_mutex = NULL;
 QueueHandle_t sub_jobs = NULL;
 
-static void send_image(struct mg_connection *c) {
+static void response(struct mg_connection *c, int status) {
-  char mp_buffer[IMAGE_BLK_SIZE + 50];
+  mpack_writer_t writer;
-  for (int i = 0; i < IMAGE_SIZE; i += IMAGE_BLK_SIZE) {
+  char *data;
-    mpack_writer_t wr;
+  size_t size;
-    mpack_writer_init(&wr, mp_buffer, sizeof(mp_buffer));
+  mpack_writer_init_growable(&writer, &data, &size);
-    mpack_build_map(&wr);
+  mpack_start_array(&writer, 1);
-    mpack_write_cstr(&wr, "command");
+  mpack_write_i16(&writer, status);
-    mpack_write_cstr(&wr, "update_image");
+  if (mpack_writer_destroy(&writer) == mpack_ok) {
-    mpack_write_cstr(&wr, "index");
+    mg_send(c, data, size);
-    mpack_write_u64(&wr, i);
+  }
-    mpack_write_cstr(&wr, "data");
+  mg_free(data);
-    mpack_write_bytes(&wr, (const char *)&image_buf[i],
+}
-                      MIN(IMAGE_SIZE, IMAGE_SIZE - i));
+
-    mpack_complete_map(&wr);
+static int new_sub(mpack_reader_t *rdr, struct mg_connection *c) {
-    size_t used = mpack_writer_buffer_used(&wr);
+  static char topic_buf[TOPIC_LEN];
-    if (mpack_writer_destroy(&wr) != mpack_ok) {
+  mpack_expect_cstr(rdr, topic_buf, sizeof(topic_buf));
-      MG_INFO(("faild build update image mpack"));
+  if (mpack_reader_destroy(rdr) != mpack_ok) {
-      return;
+    return -2;
+  }
+  struct sub **ptr = &sub_list;
+  bool dup = 0;
+  for (; *ptr; ptr = &(*ptr)->next) {
+    if (c == (*ptr)->c) {
+      dup = 1;
+      break;
     }
-    mg_send(c, mp_buffer, used);
   }
-  mpack_writer_t wr;
+  if (dup) {
-  mpack_writer_init(&wr, mp_buffer, sizeof(mp_buffer));
+    return -3;
-  mpack_build_map(&wr);
-  mpack_write_cstr(&wr, "command");
-  mpack_write_cstr(&wr, "push_image");
-  size_t used = mpack_writer_buffer_used(&wr);
-  if (mpack_writer_destroy(&wr) != mpack_ok) {
-    MG_INFO(("faild build update image mpack"));
-    return;
   }
-  mg_send(c, mp_buffer, used);
+  if (xSemaphoreTake(sub_mutex, 10) == pdTRUE) {
+    *ptr = calloc(1, sizeof(struct sub));
+    int stat = 0;
+    if (*ptr == NULL) {
+      stat = -4;
+      goto end_task;
+    }
+    strncpy((*ptr)->topic, topic_buf, TOPIC_LEN);
+    (*ptr)->c = c;
+
+  end_task:
+    xSemaphoreGive(sub_mutex);
+    return stat;
+  } else {
+    return -1;
+  }
+}
+
+static void send_image(struct mg_connection *c) {
+  mg_send(c, image_buf, sizeof(image_buf));
 }
 
 static void sfn(struct mg_connection *c, int ev, void *ev_data) {
@@ -50,24 +68,24 @@ static void sfn(struct mg_connection *c, int ev, void *ev_data) {
     MG_INFO(("SERVER is listening"));
   } else if (ev == MG_EV_ACCEPT) {
     MG_INFO(("SERVER accepted a connection"));
-    struct sub *newsub = calloc(1, sizeof(struct sub));
-    for (struct sub **ptr = &sub_list; *ptr != NULL; ptr = &(*ptr)->next) {
-      if ((*ptr)->next == NULL) {
-        (*ptr)->next = newsub;
-      }
-    }
-
   } else if (ev == MG_EV_READ) {
+    struct mg_iobuf *r = &c->recv;
+    MG_INFO(("SERVER got data: %.*s", r->len, r->buf));
+    // mg_send(c, r->buf, r->len); // echo it back
+    // r->len = 0;                 // Tell Mongoose we've consumed data
+    mpack_reader_t rx;
+    mpack_reader_init_data(&rx, (const char *)r->buf, r->len);
+    mpack_expect_array(&rx);
+    uint8_t id = mpack_expect_u8(&rx);
+    switch (id) {
+    case 1: // sub
+      response(c, new_sub(&rx, c));
+      break;
+    }
+    r->len = 0;
+
   } else if (ev == MG_EV_CLOSE) {
     MG_INFO(("SERVER disconnected"));
-    if (sub_list->c == c) {
-      sub_list = sub_list->next;
-    }
-    for (struct sub *ptr = sub_list; ptr != NULL; ptr = ptr->next) {
-      if (ptr->next && ptr->next->c == c) {
-        ptr->next = ptr->next->next;
-      }
-    }
   } else if (ev == MG_EV_ERROR) {
     MG_INFO(("SERVER error: %s", (char *)ev_data));
   } else if (ev == MG_EV_POLL) {
@@ -75,9 +93,9 @@ static void sfn(struct mg_connection *c, int ev, void *ev_data) {
     if (xQueueReceive(sub_jobs, &job, (TickType_t)20) == pdPASS) {
       MG_INFO(("SERVER CLEAN QUEUE "));
       for (struct sub *child = sub_list; child != NULL; child = child->next) {
-        // if (strncmp(child->topic, job.topic, TOPIC_LEN) == 0) {
+        if (strncmp(child->topic, job.topic, TOPIC_LEN) == 0) {
-        send_image(child->c);
+          send_image(child->c);
-        // }
+        }
       }
     }
   }