diff --git a/page.html b/page.html new file mode 100755 index 0000000..4549d3a --- /dev/null +++ b/page.html @@ -0,0 +1,27 @@ + + + + + Websocket Test + + + + + + \ No newline at end of file diff --git a/xterminal_broker.c b/xterminal_broker.c index a6ca5d4..6133a28 100755 --- a/xterminal_broker.c +++ b/xterminal_broker.c @@ -5,6 +5,14 @@ int asprintf(char **strp, const char *fmt, ...); void *memmem(const void *haystack, size_t haystacklen, const void *needle, size_t needlelen); +static LIST_HEAD(device_list); + +struct device { + char mac[13]; + int alive; + struct list_head node; +}; + static char *macaddr; static char *myid; @@ -14,6 +22,38 @@ static const char *index_page = "" ""; + + +static struct device *find_device_by_mac(struct mg_str *mac) +{ + struct device *dev = NULL; + + list_for_each_entry(dev, &device_list, node) { + if (!mg_vcmp(mac, dev->mac)) + return dev; + } + + return NULL; +} + +static void update_device(struct mg_str *mac) +{ + struct device *dev = find_device_by_mac(mac); + if (!dev) { + dev = calloc(1, sizeof(struct device)); + if (!dev) { + perror("calloc"); + return; + } + + list_add(&dev->node, &device_list); + strncpy(dev->mac, mac->p, mac->len); + } + + dev->alive = 10; + + printf("update_device:%s\n", dev->mac); +} static void ev_handler(struct mg_connection *nc, int ev, void *ev_data) { @@ -27,7 +67,7 @@ static void ev_handler(struct mg_connection *nc, int ev, void *ev_data) } else if (ev == MG_EV_MQTT_CONNACK) { struct mg_mqtt_message *msg = (struct mg_mqtt_message *)ev_data; - struct mg_mqtt_topic_expression topic_expr[2]; + struct mg_mqtt_topic_expression topic_expr[1]; if (msg->connack_ret_code != MG_EV_MQTT_CONNACK_ACCEPTED) { printf("Got mqtt connection error: %d\n", msg->connack_ret_code); @@ -36,15 +76,11 @@ static void ev_handler(struct mg_connection *nc, int ev, void *ev_data) } topic_expr[0].qos = MG_MQTT_QOS(0); - asprintf((char **)&topic_expr[0].topic, "xterminal/%s/response/data/%s", macaddr, myid); + asprintf((char **)&topic_expr[0].topic, "xterminal/heartbeat"); - topic_expr[1].qos = MG_MQTT_QOS(0); - asprintf((char **)&topic_expr[1].topic, "xterminal/%s/response/exit/%s", macaddr, myid); - - mg_mqtt_subscribe(nc, topic_expr, 2, 0); + mg_mqtt_subscribe(nc, topic_expr, 1, 0); free((char *)topic_expr[0].topic); - free((char *)topic_expr[1].topic); } else if (ev == MG_EV_MQTT_SUBACK) { char topic[128] = ""; @@ -53,19 +89,27 @@ static void ev_handler(struct mg_connection *nc, int ev, void *ev_data) } else if (ev == MG_EV_MQTT_PUBLISH) { struct mg_mqtt_message *msg = (struct mg_mqtt_message *)ev_data; - if (memmem(msg->topic.p, msg->topic.len, "data", strlen("data"))) { - int ret = write(STDOUT_FILENO, msg->payload.p, msg->payload.len); - if (ret < 0) - perror("write"); - } else if (memmem(msg->topic.p, msg->topic.len, "exit", strlen("exit"))) { - ev_break(EV_DEFAULT, EVBREAK_ALL); + + if (!mg_vcmp(&msg->topic, "xterminal/heartbeat")) { + update_device(&msg->payload); } } else if (ev == MG_EV_HTTP_REQUEST) { - mg_printf(nc, "HTTP/1.1 200 OK\r\n" - "Content-Length: %d\r\n\r\n" - "%s", (int)strlen(index_page), index_page); - + struct http_message *hm = (struct http_message *)ev_data; + if (!mg_vcmp(&hm->uri, "/list")) { + struct device *dev; + + mg_send_head(nc, 200, -1, NULL); + + list_for_each_entry(dev, &device_list, node) { + mg_printf_http_chunk(nc, "%s|", dev->mac); + } + + mg_send_http_chunk(nc, "", 0); + } else { + mg_send_head(nc, 200, strlen(index_page), NULL); + mg_printf(nc, "%s", index_page); + } } else if (ev == MG_EV_WEBSOCKET_HANDSHAKE_DONE) { char buf[] = "Hello Websocket Client"; printf("New websocket connection:%p\n", nc); @@ -80,23 +124,24 @@ static void ev_handler(struct mg_connection *nc, int ev, void *ev_data) } } +static void device_alive_cb(struct ev_loop *loop, ev_timer *w, int revents) +{ + struct device *dev, *tmp; + + list_for_each_entry_safe(dev, tmp, &device_list, node) { + if (dev->alive-- == 0) { + printf("del device:%s\n", dev->mac); + list_del(&dev->node); + free(dev); + } + } +} + static void signal_cb(struct ev_loop *loop, ev_signal *w, int revents) { printf("Got signal: %d\n", w->signum); ev_break(loop, EVBREAK_ALL); } - -static void usage(const char *name) -{ - fprintf(stderr, - "Usage: %s [OPTION]\n" - " -m macaddr\n" - " -i id\n" - "\n", name - ); - - exit(0); -} int main(int argc, char *argv[]) { @@ -106,30 +151,14 @@ int main(int argc, char *argv[]) struct mg_connection *nc; const char *s_address = "localhost:1883"; const char *s_http_port = "8000"; - int ch; - - while ((ch = getopt(argc, argv, "m:i:")) != -1) { - switch (ch) { - case 'm': - macaddr = optarg; - break; - - case 'i': - myid = optarg; - break; - - default: - usage(argv[0]); - } - } - - if ((optind < argc) || !macaddr || !myid) { - usage(argv[0]); - } + ev_timer device_timer; ev_signal_init(&sig_watcher, signal_cb, SIGINT); ev_signal_start(loop, &sig_watcher); + ev_timer_init(&device_timer, device_alive_cb, 0.1, 2.0); + ev_timer_start(loop, &device_timer); + mg_mgr_init(&mgr, NULL); nc = mg_connect(&mgr, s_address, ev_handler); diff --git a/xterminal_client.c b/xterminal_client.c index 45f816d..4b25fc7 100755 --- a/xterminal_client.c +++ b/xterminal_client.c @@ -18,7 +18,6 @@ void *memmem(const void *haystack, size_t haystacklen, const void *needle, size_ LIST_HEAD(clients); static char *devid; -static const char *s_address = "localhost:1883"; static void pty_read_cb(struct ev_loop *loop, ev_io *w, int revents) { @@ -141,6 +140,14 @@ static void signal_cb(struct ev_loop *loop, ev_signal *w, int revents) } } +static void heartbeat_cb(struct ev_loop *loop, ev_timer *w, int revents) +{ + struct mg_connection *nc = (struct mg_connection *)w->data; + + mg_mqtt_publish(nc, "xterminal/heartbeat", 0, MG_MQTT_QOS(0), devid, strlen(devid)); + printf("send heartbeat:%s\n", devid); +} + static int get_iface_mac(const char *ifname, char mac[6]) { int r, s; @@ -174,9 +181,11 @@ int main(int argc, char *argv[]) struct ev_loop *loop = EV_DEFAULT; ev_signal sigint_watcher; ev_signal sigchld_watcher; + ev_timer heartbeat_timer; struct mg_mgr mgr; struct mg_connection *nc; char mac[6] = ""; + const char *s_address = "localhost:1883"; mg_mgr_init(&mgr, NULL); @@ -193,6 +202,10 @@ int main(int argc, char *argv[]) ev_signal_init(&sigchld_watcher, signal_cb, SIGCHLD); sigchld_watcher.data = nc; ev_signal_start(loop, &sigchld_watcher); + + ev_timer_init(&heartbeat_timer, heartbeat_cb, 0.1, 2.0); + heartbeat_timer.data = nc; + ev_timer_start(loop, &heartbeat_timer); get_iface_mac("enp3s0", mac);