add simple commands processing (at process)

main.c

@@ -32,6 +32,7 @@
 
 #include "auth.h"
 #include "cmdlnopts.h"
+#include "netproto.h"
 #include "websockets.h"
 
 // temporary
@@ -107,21 +108,33 @@ static void runServer(){
     signal(SIGTSTP, SIG_IGN);
     signal(SIGHUP, SIG_IGN);
     // if(G.logfilename) Cl_createlog();
-    pthread_t pg_thread;//, ws_thread;
+    pthread_t pg_thread, main_thread;
     if(pthread_create(&pg_thread, NULL, runPostGet, NULL)){
         ERR("pthread_create()");
     }
+    if(pthread_create(&main_thread, NULL, runMainProc, NULL)){
+        ERR("pthread_create()");
+    }
     do{
         if(STOP) return;
         if(pthread_kill(pg_thread, 0) == ESRCH){ // server died
-            WARNX("POST/GET server thread died");
-            putlog("POST/GET server thread died");
+            WARNX("Server thread died");
+            putlog("Server thread died");
             pthread_join(pg_thread, NULL);
             if(pthread_create(&pg_thread, NULL, runPostGet, NULL)){
                 putlog("pthread_create() failed");
                 ERR("pthread_create()");
             }
         }
+        if(pthread_kill(main_thread, 0) == ESRCH){ // server died
+            WARNX("Main thread died");
+            putlog("Main thread died");
+            pthread_join(main_thread, NULL);
+            if(pthread_create(&main_thread, NULL, runMainProc, NULL)){
+                putlog("pthread_create() failed");
+                ERR("pthread_create()");
+            }
+        }
 #if 0
         usleep(1000); // sleep a little or thread's won't be able to lock mutex
         if(dtime() - tgot < T_INTERVAL) continue;

netproto.c

@@ -0,0 +1,137 @@
+/*
+ * This file is part of the Onion_test project.
+ * Copyright 2020 Edward V. Emelianov <edward.emelianoff@gmail.com>.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "netproto.h"
+
+#include <onion/types_internal.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <string.h>
+#include <time.h>
+#include <usefull_macros.h>
+
+typedef struct list_{
+    onion_websocket *ws;
+    struct list_ *next;
+    struct list_ *prev;
+    struct list_ *last;
+} WSlist;
+
+static WSlist *wslist = NULL;
+static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
+static char message[256];
+
+/**
+ * @brief runMainProc - main data process
+ * @param data - unused
+ * @return
+ */
+void *runMainProc(_U_ void *data){
+    while(1){
+        int changed = 0;
+        pthread_mutex_lock(&mutex);
+        time_t t = time(NULL);
+        struct tm *tmp = localtime(&t);
+        if(tmp){
+            strftime(message, 256, "%d/%m/%y, %T", tmp);
+            changed = 1;
+        }
+        pthread_mutex_unlock(&mutex);
+        if(changed) send_all_WS(message);
+        sleep(1);
+    }
+    return NULL;
+}
+
+/**
+ * @brief process_WS_signal - get signal from websocket and do something with it
+ * @param ws (i)     - websocket
+ * @param signal (i) - command received
+ */
+void process_WS_signal(onion_websocket *ws, char *signal){
+    char *eq = strchr(signal, '=');
+    if(eq){
+        *eq++ = 0;
+        onion_websocket_printf(ws, "parameter: '%s', its value: '%s'", signal, eq);
+    }else{
+        onion_websocket_printf(ws, "Echo: %s", signal);
+    }
+}
+
+/**
+ * @brief register_WS - add recently opened websocket to common list
+ * @param ws - websocket
+ */
+void register_WS(onion_websocket *ws){
+    green("Add NEW websocket\n");
+    if(!wslist){
+        wslist = MALLOC(WSlist, 1);
+        wslist->ws = ws;
+        wslist->last = wslist;
+    }else{
+        wslist->last->next = MALLOC(WSlist, 1);
+        wslist->last->next->ws = ws;
+        wslist->last->next->prev = wslist->last;
+        wslist->last = wslist->last->next;
+    }
+}
+
+/**
+ * @brief send_all_WS - send data to all opened websockets
+ * @param data (i) - data to send
+ */
+void send_all_WS(char *data){
+    if(strlen(data) == 0) return; // zero length
+    WSlist *l = wslist;
+    int cnt = 0;
+    unregister_WS(); // check for dead ws
+    while(l){
+        DBG("try to send");
+        if(onion_websocket_printf(l->ws, "%s", data) <= 0){ // dead websocket? remove it from list
+            DBG("Error printing - check for dead");
+            unregister_WS();
+        }else ++cnt;
+        l = l->next;
+    }
+    DBG("Send message %s to %d clients", data, cnt);
+}
+
+void unregister_WS(){
+    WSlist *l = wslist;
+    while(l){
+        if(l->ws->req->connection.fd < 0){
+            red("Found dead WS, remove it\n");
+            if(l->prev){
+                //DBG("l->prev->next = l->next");
+                l->prev->next = l->next;
+            }
+            if(l->next){
+                //DBG("l->next->prev = l->prev");
+                l->next->prev = l->prev;
+            }else{
+                //DBG("wslist->last = l->prev");
+                wslist->last = l->prev; // we should delete last element
+            }
+            WSlist *nxt = l->next;
+            FREE(l);
+            l = nxt;
+            continue;
+        }
+        l = l->next;
+    }
+}

netproto.h

@@ -0,0 +1,32 @@
+/*
+ * This file is part of the Onion_test project.
+ * Copyright 2020 Edward V. Emelianov <edward.emelianoff@gmail.com>.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+#ifndef NETPROTO_H__
+
+#include "websockets.h"
+
+void *runMainProc(void *data);
+
+void process_WS_signal(onion_websocket *ws, char *signal);
+void register_WS(onion_websocket *ws);
+void unregister_WS();
+void send_all_WS(char *data);
+
+#define NETPROTO_H__
+#endif // NETPROTO_H__

websockets.c

@@ -17,11 +17,13 @@
  */
 
 #include "auth.h"
+#include "netproto.h"
 #include "websockets.h"
 
 #include <errno.h>
 #include <onion/dict.h>
 #include <onion/log.h>
+#include <onion/types_internal.h>
 #include <stdio.h>
 #include <string.h>
 #include <usefull_macros.h>
@@ -54,9 +56,11 @@ static onion_connection_status websocket_cont(void *data, onion_websocket *ws, s
     if(dlen > BUFLEN) dlen = BUFLEN;
 
     int len = onion_websocket_read(ws, tmp, dlen);
-    if(len <= 0){
+    if(!len) return OCS_NEED_MORE_DATA;
+    if(len < 0){
         ONION_ERROR("Error reading data: %d: %s (%d)", errno, strerror(errno), dlen);
-        return OCS_NEED_MORE_DATA;
+        unregister_WS();
+        return OCS_CLOSE_CONNECTION;
     }
     tmp[len] = 0;
     //ONION_INFO("Read from websocket: %s (len=%d)", tmp, len);
@@ -93,13 +97,7 @@ static onion_connection_status websocket_cont(void *data, onion_websocket *ws, s
         wsdata->flags &= ~WS_FLAG_NOTAUTHORIZED; // clear non-authorized flag
         return OCS_NEED_MORE_DATA;
     }
-    char *eq = strchr(tmp, '=');
-    if(eq){
-        *eq++ = 0;
-        onion_websocket_printf(ws, "parameter: '%s', its value: '%s'", tmp, eq);
-    }else{
-        onion_websocket_printf(ws, "Echo: %s", tmp);
-    }
+    process_WS_signal(ws, tmp);
     return OCS_NEED_MORE_DATA;
 }
 
@@ -120,5 +118,6 @@ onion_connection_status websocket_run(_U_ void *data, onion_request *req, onion_
     wsdata->UAhash = MurmurOAAT64(UA);
     onion_websocket_set_userdata(ws, (void*)wsdata, free);
     onion_websocket_set_callback(ws, websocket_cont);
+    register_WS(ws);
     return OCS_WEBSOCKET;
 }