Use pipes to store pubsub messages

include/pubsub.h

@@ -17,27 +17,18 @@
 
 #define MAX_STR_SIZE 1024
 
-struct strNode {
-  char *str;
-  struct strNode *next;
-};
-typedef struct {
-  struct strNode *head;
-  struct strNode *tail;
-} StrQueue;
-
 typedef struct {
     UA_UInt16 id;
     UA_Double latitude;
     UA_Double longitude;
     UA_Double altitudeAbs;
     UA_Double altitudeRel;
-    UA_Int64 timestamp;
+    UA_Int64  timestamp;
     UA_Float  yaw;
     UA_Float  speed;
     UA_Float  climbRate;
-    StrQueue receiveMessageQueue;
-    StrQueue receiveLogQueue;
+    int receiveMessagePipefds[2];
+    int receiveLogPipefds[2];
 } JSDroneData;
 
 typedef struct {

qjs_wrapper.c

@@ -107,16 +107,11 @@ static UA_Boolean pubsubExited = true;
 static UA_UInt32 nbDrone;
 static UA_UInt32 nbSubscriber;
 static JSValueConst *droneObjectIdList;
-static StrQueue sendMessageQueue = {
-    .head = NULL,
-    .tail = NULL,
-};
+
 UA_String currentSendMessage;
-static StrQueue sendLogQueue = {
-    .head = NULL,
-    .tail = NULL,
-};
+int sendMessagePipefds[2];
 UA_String currentSendLog;
+int sendLogPipefds[2];
 
 bool isADrone;
 
@@ -190,30 +185,23 @@ static const JSCFunctionListEntry js_position_proto_funcs[] = {
 
 // Drone class functions
 
-static void delete_str_node(struct strNode *node) {
-    free(node->str);
-    free(node);
-}
-
-static void cleanQueue(StrQueue *pQueue)
-{
-    struct strNode *current;
-
-    while (pQueue->head != NULL) {
-        current = pQueue->head;
-        pQueue->head = current->next;
-        delete_str_node(current);
-    }
-}
-
 static void js_drone_finalizer(JSRuntime *rt, JSValue val)
 {
     JSDroneData *s = (JSDroneData *) JS_GetOpaque(val, jsDroneClassId);
-    cleanQueue(&(s->receiveMessageQueue));
-    cleanQueue(&(s->receiveLogQueue));
+    close(s->receiveMessagePipefds[0]);
+    close(s->receiveMessagePipefds[1]);
+    close(s->receiveLogPipefds[0]);
+    close(s->receiveLogPipefds[1]);
     js_free_rt(rt, s);
 }
 
+static int set_pipe_non_blocking(int rpipefd) {
+    int res = fcntl(rpipefd, F_SETFL, fcntl(rpipefd, F_GETFL) | O_NONBLOCK);
+    if (res < 0)
+        perror("Failed to set pipe non blocking");
+    return res;
+}
+
 static JSValue js_drone_ctor(JSContext *ctx, JSValueConst newTarget, 
                              int argc, JSValueConst *argv)
 {
@@ -228,14 +216,22 @@ static JSValue js_drone_ctor(JSContext *ctx, JSValueConst newTarget,
     if (JS_ToUint32(ctx, &uint32, argv[0]))
         goto fail;
     s->id = (uint16_t)uint32;
-    s->receiveMessageQueue = (StrQueue){
-        .head = NULL,
-        .tail = NULL,
-    };
-    s->receiveLogQueue = (StrQueue){
-        .head = NULL,
-        .tail = NULL,
-    };
+
+    if (pipe(s->receiveMessagePipefds) < 0) {
+        perror("Failed to create receiveMessagePipe");
+        goto fail;
+    }
+
+    if (set_pipe_non_blocking(s->receiveMessagePipefds[0]) < 0)
+        goto fail;
+
+    if (pipe(s->receiveLogPipefds) < 0) {
+        perror("Failed to create receiveLogPipe");
+        goto fail;
+    }
+
+    if (set_pipe_non_blocking(s->receiveLogPipefds[0]) < 0)
+        goto fail;
 
     proto = JS_GetPropertyStr(ctx, newTarget, "prototype");
     if (JS_IsException(proto))
@@ -247,6 +243,10 @@ static JSValue js_drone_ctor(JSContext *ctx, JSValueConst newTarget,
     JS_SetOpaque(obj, s);
     return obj;
  fail:
+    close(s->receiveMessagePipefds[0]);
+    close(s->receiveMessagePipefds[1]);
+    close(s->receiveLogPipefds[0]);
+    close(s->receiveLogPipefds[1]);
     js_free(ctx, s);
     JS_FreeValue(ctx, obj);
     return JS_EXCEPTION;
@@ -257,31 +257,32 @@ static JSValue js_drone_init(JSContext *ctx, JSValueConst thisVal,
 {
     int nb;
     JSDroneData *s = (JSDroneData *) JS_GetOpaque2(ctx, thisVal, jsDroneClassId);
+
     if (!s)
         return JS_EXCEPTION;
+
     if (JS_ToInt32(ctx, &nb, argv[0]))
         return JS_EXCEPTION;
     droneObjectIdList[nb] = thisVal;
     return JS_UNDEFINED;
 }
 
-static JSValue readDroneDataStr(JSContext *ctx, StrQueue *pQueue, bool keepAtLeastAnElement)
+static ssize_t read_from_pipe(int file, char *rd_buf, size_t max_read)
+{
+    ssize_t nread = read(file, rd_buf, max_read);
+
+    if (nread == -1 && errno != EAGAIN)
+        perror("Failed to read pipe");
+    return nread;
+}
+
+static JSValue getStrFromPipe(JSContext *ctx, int rpipefd)
 {
     JSValue res;
-    struct strNode *current;
-
-    current = pQueue->head;
-    if (current != NULL) {
-        res = JS_NewString(ctx, current->str);
-        if (current->next != NULL) {
-            pQueue->head = current->next;
-            delete_str_node(current);
-        } else {
-            if (!keepAtLeastAnElement) {
-                pQueue->head = (pQueue->tail = NULL);
-                delete_str_node(current);
-            }
-        }
+    char rd_buf[MAX_STR_SIZE];
+
+    if (read_from_pipe(rpipefd, rd_buf, MAX_STR_SIZE) > 0) {
+        res = JS_NewString(ctx, rd_buf);
     } else {
         res = JS_NewString(ctx, "");
     }
@@ -311,9 +312,9 @@ static JSValue js_drone_get(JSContext *ctx, JSValueConst thisVal, int magic)
         case 7:
             return JS_NewFloat64(ctx, s->climbRate);
         case 8:
-            return readDroneDataStr(ctx, &(s->receiveMessageQueue), true);
+            return getStrFromPipe(ctx, s->receiveMessagePipefds[0]);
         case 9:
-            return readDroneDataStr(ctx, &(s->receiveLogQueue), false);
+            return getStrFromPipe(ctx, s->receiveLogPipefds[0]);
         case 10:
             return JS_NewInt64(ctx, s->timestamp);
         default:
@@ -321,46 +322,39 @@ static JSValue js_drone_get(JSContext *ctx, JSValueConst thisVal, int magic)
     }
 }
 
-static void addStrToQueue(const char *str, StrQueue *pQueue)
+static void write_to_pipe(int file, const char *wr_buf, size_t max_written)
 {
-    struct strNode *newNode;
-    newNode = (struct strNode*)malloc(sizeof(struct strNode));
-    newNode->str = strdup(str);
-    newNode->next = NULL;
-
-    if (pQueue->tail == NULL) {
-        pQueue->head = pQueue->tail = newNode;
-    } else {
-        pQueue->tail->next = newNode;
-        pQueue->tail = newNode;
-    }
+    ssize_t nwritten = write(file, wr_buf, max_written);
+    if (nwritten < 0)
+        perror("Failed to write pipe");
 }
 
-static JSValue add_jsstr_to_queue(JSContext *ctx, JSValueConst jsStr,
-                                  StrQueue *pQueue)
+static JSValue add_jsstr_to_pipe(JSContext *ctx, JSValueConst jsStr,
+                                  int wpipefd)
 {
-    const char *str;
-    str = JS_ToCString(ctx, jsStr);
-    if (strlen(str) > MAX_STR_SIZE) {
+    const char *wr_buf;
+    wr_buf = JS_ToCString(ctx, jsStr);
+    if (strlen(wr_buf) > MAX_STR_SIZE) {
         UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "String too long");
+        JS_FreeCString(ctx, wr_buf);
         return JS_EXCEPTION;
     }
 
-    addStrToQueue(str, pQueue);
-    JS_FreeCString(ctx, str);
+    write_to_pipe(wpipefd, wr_buf, MAX_STR_SIZE);
+    JS_FreeCString(ctx, wr_buf);
     return JS_UNDEFINED;
 }
 
 static JSValue js_drone_set_message(JSContext *ctx, JSValueConst thisVal,
                                     int argc, JSValueConst *argv)
 {
-    return add_jsstr_to_queue(ctx, argv[0], &sendMessageQueue);
+    return add_jsstr_to_pipe(ctx, argv[0], sendMessagePipefds[1]);
 }
 
 static JSValue js_drone_set_log(JSContext *ctx, JSValueConst thisVal,
                                 int argc, JSValueConst *argv)
 {
-    return add_jsstr_to_queue(ctx, argv[0], &sendLogQueue);
+    return add_jsstr_to_pipe(ctx, argv[0], sendLogPipefds[1]);
 }
 
 static UA_Boolean UA_String_isEmpty(const UA_String *s) {
@@ -372,28 +366,23 @@ static void clear_str(UA_String *pstr) {
         UA_String_clear(pstr); 
 }
 
-static UA_String get_StrQueue_content(StrQueue *pQueue, UA_String currentStr)
+static UA_String get_ua_str_from_pipe(int rpipefd, UA_String currentStr)
 {
-    struct strNode *current;
-
-    current = pQueue->head;
-    if (current != NULL) {
-        clear_str(&currentStr);
-        currentStr = UA_STRING_ALLOC(current->str);
-        pQueue->head = current->next == NULL ? (pQueue->tail = NULL) : current->next;
-        delete_str_node(current);
-    }
+    char rd_buf[MAX_STR_SIZE];
+    clear_str(&currentStr);
+    currentStr = (read_from_pipe(rpipefd, rd_buf, MAX_STR_SIZE) > 0) ?
+      UA_STRING_ALLOC(rd_buf) : UA_STRING("");
     return currentStr;
 }
 
 UA_String get_message(void)
 {
-    return get_StrQueue_content(&sendMessageQueue, currentSendMessage);
+    return get_ua_str_from_pipe(sendMessagePipefds[0], currentSendMessage);
 }
 
 UA_String get_log(void)
 {
-    return get_StrQueue_content(&sendLogQueue, currentSendLog);
+    return get_ua_str_from_pipe(sendLogPipefds[0], currentSendLog);
 }
 
 static JSClassDef jsDroneClass = {
@@ -482,16 +471,18 @@ VariableData pubsub_get_value(UA_String identifier) {
     return varDetails;
 }
 
-static void setDroneDataStr(void *data, StrQueue* pQueue)
+static void pushToPipe(void *data, int wpipefd)
 {
     UA_String uaStr = *(UA_String*) data;
-    char str[MAX_STR_SIZE];
-    memcpy(str, uaStr.data, uaStr.length);
-    str[uaStr.length] = '\0';
-    addStrToQueue(str, pQueue);
+    char wr_buf[MAX_STR_SIZE];
+
+    memcpy(wr_buf, uaStr.data, uaStr.length);
+    wr_buf[uaStr.length] = '\0';
+    write_to_pipe(wpipefd, wr_buf, MAX_STR_SIZE);
 }
 
-static void pubsub_update_variables(UA_UInt32 id, const UA_DataValue *var, bool print)
+static void pubsub_update_variables(UA_UInt32 id,
+                                    const UA_DataValue *var, bool print)
 {
     JSDroneData* s;
     UA_Int64* updatedPositionArray;
@@ -526,11 +517,11 @@ static void pubsub_update_variables(UA_UInt32 id, const UA_DataValue *var, bool
             }
             return;
         } else if (pubsubIdsArray[i].message == id) {
-            setDroneDataStr(var->value.data, &(s->receiveMessageQueue));
+            pushToPipe(var->value.data, s->receiveMessagePipefds[1]);
             return;
         } else if (pubsubIdsArray[i].log == id) {
             if (!isADrone) {
-                setDroneDataStr(var->value.data, &(s->receiveLogQueue));
+                pushToPipe(var->value.data, s->receiveLogPipefds[1]);
             }
             return;
         }
@@ -623,15 +614,37 @@ static JSValue js_init_pubsub(JSContext *ctx, JSValueConst thisVal,
                               int argc, JSValueConst *argv)
 {
     if (JS_ToUint32(ctx, &nbDrone, argv[0]))
-        return JS_EXCEPTION;
+        goto fail;
 
     if (JS_ToUint32(ctx, &nbSubscriber, argv[1]))
-        return JS_EXCEPTION;
+        goto fail;
+
+    if (pipe(sendMessagePipefds) < 0) {
+        perror("Failed to create sendMessagePipe");
+        goto fail;
+    }
+
+    if (set_pipe_non_blocking(sendMessagePipefds[0]) < 0)
+        goto fail;
+
+    if (pipe(sendLogPipefds) < 0) {
+        perror("Failed to create sendLogPipe");
+        goto fail;
+    }
+
+    if (set_pipe_non_blocking(sendLogPipefds[0]) < 0)
+        perror("Failed to set sendLogPipe non blocking");
 
     currentSendMessage = UA_STRING("");
     currentSendLog = UA_STRING("");
     droneObjectIdList = (JSValue *) malloc((nbDrone + nbSubscriber) * sizeof(JSValueConst));
     return JS_NewInt32(ctx, 0);
+fail:
+    close(sendMessagePipefds[0]);
+    close(sendMessagePipefds[1]);
+    close(sendLogPipefds[0]);
+    close(sendLogPipefds[1]);
+    return JS_EXCEPTION;
 }
 
 static JSValue js_stop_pubsub(JSContext *ctx, JSValueConst thisVal,
@@ -641,10 +654,11 @@ static JSValue js_stop_pubsub(JSContext *ctx, JSValueConst thisVal,
     while(!pubsubExited)
         sleep(1);
     free(droneObjectIdList);
-
-    cleanQueue(&sendMessageQueue);
+    close(sendMessagePipefds[0]);
+    close(sendMessagePipefds[1]);
     clear_str(&currentSendMessage);
-    cleanQueue(&sendLogQueue);
+    close(sendLogPipefds[0]);
+    close(sendLogPipefds[1]);
     clear_str(&currentSendLog);
 
     return JS_NewInt32(ctx, 0);