Durable extenstion support part I

samiaghezal · Jul 22, 2019 · adba7cb · adba7cb
1 parent 901749e
commit adba7cb
Show file tree

Hide file tree

Showing 8 changed files with 147 additions and 10 deletions.
diff --git a/libjs/durableEngine.js b/libjs/durableEngine.js
@@ -343,6 +343,10 @@ exports = module.exports = durableEngine = function () {
             r.assertTimers(handle);
         };
 
+        that.setStoreMessageCallback = function (func) {
+            r.setStoreMessageCallback(handle, func);
+        };
+
         var flushActions = function (state, resultContainer, stateOffset, complete) {
             while (resultContainer['message']) {
                 var actionName = null;
@@ -902,6 +906,10 @@ exports = module.exports = durableEngine = function () {
             that.getRuleset(rulesetName).renewActionLease(sid);
         };
 
+        that.setStoreMessageCallback = function (rulesetName, func) {
+            that.getRuleset(rulesetName).setStoreMessageCallback(func);
+        };
+
         var dispatchRules = function (index) {
             if (!rulesList.length) {
                 setTimeout(dispatchRules, 500, index);

diff --git a/src/rules/events.c b/src/rules/events.c
@@ -998,6 +998,7 @@ static unsigned int handleBetaFrame(ruleset *tree,
 
 static unsigned int handleDeleteMessage(ruleset *tree,
                                         stateNode *state,
+                                        char *mid,
                                         unsigned int messageOffset) {
     unsigned int result;
     unsigned int count = 0;
@@ -1139,7 +1140,10 @@ static unsigned int handleDeleteMessage(ruleset *tree,
         }
     }
 
-    result = deleteMessage(state, messageOffset);
+    result = deleteMessage(tree, 
+                           state,
+                           mid, 
+                           messageOffset);
     if (result == ERR_NODE_DELETED) {
         return RULES_OK;
     } else if (result != RULES_OK) {
@@ -1702,7 +1706,7 @@ static unsigned int handleMessageCore(ruleset *tree,
 
     sid[sidProperty->valueLength] = '\0';
 
-    #ifdef _WIN32
+#ifdef _WIN32
     char *mid = (char *)_alloca(sizeof(char)*(midProperty->valueLength + 1));
 #else
     char mid[midProperty->valueLength + 1];
@@ -1725,11 +1729,13 @@ static unsigned int handleMessageCore(ruleset *tree,
         if (sidState->factOffset != UNDEFINED_HASH_OFFSET) {
             CHECK_RESULT(handleDeleteMessage(tree,
                                              sidState,
+                                             mid,
                                              sidState->factOffset));
 
         }
 
-        CHECK_RESULT(storeMessage(sidState,
+        CHECK_RESULT(storeMessage(tree,
+                                  sidState,
                                   mid,
                                   jo,
                                   MESSAGE_TYPE_FACT,
@@ -1752,10 +1758,12 @@ static unsigned int handleMessageCore(ruleset *tree,
         if (*messageOffset != UNDEFINED_HASH_OFFSET) {
             CHECK_RESULT(handleDeleteMessage(tree,
                                              sidState,
+                                             mid,
                                              *messageOffset));
         }
     } else {
-        CHECK_RESULT(storeMessage(sidState,
+        CHECK_RESULT(storeMessage(tree,
+                                  sidState,
                                   mid,
                                   jo,
                                   (actionType == ACTION_ASSERT_FACT ? MESSAGE_TYPE_FACT : MESSAGE_TYPE_EVENT),
@@ -2054,8 +2062,25 @@ static unsigned int deleteCurrentAction(ruleset *tree,
 
 
                 if (currentMessageNode->messageType == MESSAGE_TYPE_EVENT) {
+                    jsonObject *jo = &currentMessageNode->jo;
+                    jsonProperty *midProperty = &jo->properties[jo->idIndex];
+
+#ifdef _WIN32
+                    char *mid = (char *)_alloca(sizeof(char)*(midProperty->valueLength + 1));
+#else
+                    char mid[midProperty->valueLength + 1];
+#endif
+                    if (midProperty->valueOffset) {
+                        strncpy(mid, jo->content + midProperty->valueOffset, midProperty->valueLength);
+                    } else {
+                        strncpy(mid, jo->idBuffer, midProperty->valueLength);
+                    }
+
+                    mid[midProperty->valueLength] = '\0';
+
                     CHECK_RESULT(handleDeleteMessage(tree,
-                                                     state, 
+                                                     state,
+                                                     mid, 
                                                      currentMessageFrame->messageNodeOffset));
                 }
             }

diff --git a/src/rules/rete.c b/src/rules/rete.c
@@ -1852,6 +1852,9 @@ unsigned int createRuleset(unsigned int *handle, char *name, char *rules) {
     tree->actionCount = 0;
     tree->bindingsList = NULL;
     tree->currentStateIndex = 0;
+    tree->storeMessageCallback = NULL;
+    tree->storeMessageCallbackContext = NULL;
+    tree->deleteMessageCallback = NULL;
     memset(tree->stateIndex, 0, MAX_STATE_INDEX_LENGTH * sizeof(unsigned int) * 2);
     memset(tree->reverseStateIndex, 0, MAX_STATE_INDEX_LENGTH * sizeof(unsigned int));
     initStatePool(tree);
@@ -1894,4 +1897,21 @@ unsigned int deleteRuleset(unsigned int handle) {
     return RULES_OK;
 }
 
+unsigned int setStoreMessageCallback(unsigned int handle, 
+                                     void *context, 
+                                     unsigned int (*callback)(void *, char *, char *, char *)) {
+    ruleset *tree;
+    RESOLVE_HANDLE(handle, &tree);
+
+    tree->storeMessageCallbackContext = context;
+    tree->storeMessageCallback = callback;
+    return RULES_OK;
+}
 
+unsigned int setDeleteMessageCallback(unsigned int handle, unsigned int (*callback)(char *, char *)) {
+    ruleset *tree;
+    RESOLVE_HANDLE(handle, &tree);
+
+    tree->deleteMessageCallback = callback;    
+    return RULES_OK;
+}
diff --git a/src/rules/rete.h b/src/rules/rete.h
@@ -146,6 +146,10 @@ typedef struct ruleset {
     unsigned int stateIndex[MAX_STATE_INDEX_LENGTH];
     unsigned int reverseStateIndex[MAX_STATE_INDEX_LENGTH];
     unsigned int currentStateIndex;
+
+    unsigned int (*storeMessageCallback)(void*, char *, char *, char *);
+    void *storeMessageCallbackContext;
+    unsigned int (*deleteMessageCallback)(char *, char *);
 } ruleset;
 
 #ifdef _PRINT

diff --git a/src/rules/rules.h b/src/rules/rules.h
@@ -112,6 +112,12 @@ unsigned int createRuleset(unsigned int *handle,
                            char *name, 
                            char *rules);
 
+unsigned int setStoreMessageCallback(unsigned int handle, 
+                                     void *context, 
+                                     unsigned int (*callback)(void *, char *, char *, char *));
+
+unsigned int setDeleteMessageCallback(unsigned int handle, unsigned int (*callback)(char *, char *));
+
 unsigned int deleteRuleset(unsigned int handle);
 
 unsigned int createClient(unsigned int *handle, 

diff --git a/src/rules/state.c b/src/rules/state.c
@@ -670,10 +670,17 @@ unsigned int createRightFrame(stateNode *state,
     return RULES_OK;
 }
 
-unsigned int deleteMessage(stateNode *state,
+unsigned int deleteMessage(void *tree,
+                           stateNode *state,
+                           char *mid,
                            unsigned int messageNodeOffset) {
 
     messageNode *node = MESSAGE_NODE(state, messageNodeOffset);
+    ruleset *rulesetTree = (ruleset*)tree;
+    if (rulesetTree->deleteMessageCallback) {
+        return rulesetTree->deleteMessageCallback(state->sid, mid);
+    }
+
     if (node->jo.content) {
         free(node->jo.content);
         free(node->locationPool.content);
@@ -686,6 +693,7 @@ unsigned int deleteMessage(stateNode *state,
            MAX_MESSAGE_INDEX_LENGTH,
            state->messagePool,
            messageNodeOffset);
+
     return RULES_OK;
 }
 
@@ -724,7 +732,8 @@ static unsigned int copyMessage(jsonObject *targetMessage,
     return RULES_OK;
 }
 
-unsigned int storeMessage(stateNode *state,
+unsigned int storeMessage(void *tree,
+                          stateNode *state,
                           char *mid,
                           jsonObject *message,
                           unsigned char messageType,
@@ -762,7 +771,17 @@ unsigned int storeMessage(stateNode *state,
     memset(node->locationIndex, 0, MAX_LOCATION_INDEX_LENGTH * sizeof(unsigned int) * 2);
 
     node->messageType = messageType;
-    return copyMessage(&node->jo, message);
+    CHECK_RESULT(copyMessage(&node->jo, message));
+
+    ruleset *rulesetTree = (ruleset*)tree;
+    if (rulesetTree->storeMessageCallback) {
+        return rulesetTree->storeMessageCallback(rulesetTree->storeMessageCallbackContext, 
+                                                 state->sid, 
+                                                 mid, 
+                                                 message->content);
+    }
+
+    return RULES_OK;
 }
 
 unsigned int ensureStateNode(void *tree, 

diff --git a/src/rules/state.h b/src/rules/state.h
@@ -307,14 +307,17 @@ unsigned int createActionFrame(stateNode *state,
                                leftFrameNode **newNode,
                                frameLocation *newLocation);
 
-unsigned int deleteMessage(stateNode *state,
+unsigned int deleteMessage(void *tree,
+                           stateNode *state,
+                           char *mid,
                            unsigned int messageNodeOffset);
 
 unsigned int getMessage(stateNode *state,
                         char *mid,
                         unsigned int *valueOffset);
 
-unsigned int storeMessage(stateNode *state,
+unsigned int storeMessage(void *tree,
+                          stateNode *state,
                           char *mid,
                           jsonObject *message,
                           unsigned char messageType,

diff --git a/src/rulesjs/rules.cc b/src/rulesjs/rules.cc
@@ -31,6 +31,18 @@ class ObjectProxy {
 };
 
 
+class CallbackProxy {
+public:
+
+    explicit CallbackProxy(Handle<Function> value) { 
+        Isolate* isolate = Isolate::GetCurrent();
+        func.Reset(isolate, value); 
+
+    }
+
+    Persistent<Function> func;
+};
+
 void jsCreateRuleset(const FunctionCallbackInfo<Value>& args) {
     Isolate* isolate;
     isolate = args.GetIsolate();
@@ -515,6 +527,43 @@ void jsRenewActionLease(const FunctionCallbackInfo<Value>& args) {
     }
 }
 
+static unsigned int storeMessageCallback(void *context, char *sid, char *mid, char *content) {
+    CallbackProxy *p = (CallbackProxy *)context;
+    Isolate* isolate = Isolate::GetCurrent();
+
+    Local<Value> args[3];
+    args[0] = String::NewFromUtf8(isolate, sid);
+    args[1] = String::NewFromUtf8(isolate, mid);
+    args[2] = String::NewFromUtf8(isolate, content);
+
+    Local<Function> storeMessageFunc = Local<Function>::New(isolate, p->func);
+    Local<Value> result = storeMessageFunc->Call(isolate->GetCurrentContext()->Global(), 3, args);    
+    return TO_NUMBER(isolate, result);
+}
+
+void jsSetStoreMessageCallback(const FunctionCallbackInfo<Value>& args) {
+    Isolate* isolate;
+    isolate = args.GetIsolate();
+    if (args.Length() < 2) {
+        isolate->ThrowException(Exception::TypeError(String::NewFromUtf8(isolate, "Wrong number of arguments")));
+    } else if (!args[0]->IsNumber() || !args[1]->IsFunction()) {
+        isolate->ThrowException(Exception::TypeError(String::NewFromUtf8(isolate, "Wrong argument type")));
+    } else {
+        CallbackProxy *p = new CallbackProxy(Handle<Function>::Cast(args[1]));
+        unsigned int result = setStoreMessageCallback(TO_NUMBER(isolate, args[0]),
+                                                      p,
+                                                      &storeMessageCallback);  
+        if (result != RULES_OK) {
+            char *message = NULL;
+            if (asprintf(&message, "Could not set storage message callback, error code: %d", result) == -1) {
+                isolate->ThrowException(Exception::Error(String::NewFromUtf8(isolate, "Out of memory")));
+            } else {
+                isolate->ThrowException(Exception::Error(String::NewFromUtf8(isolate, message)));
+            }
+        }  
+    }
+}
+
 void jsCreateProxy(const FunctionCallbackInfo<Value>& args) {
     Isolate* isolate;
     isolate = args.GetIsolate();
@@ -631,6 +680,9 @@ void init(Handle<Object> exports) {
 
     exports->Set(String::NewFromUtf8(isolate, "createProxy", String::kInternalizedString),
         FunctionTemplate::New(isolate, jsCreateProxy)->GetFunction());
+
+    exports->Set(String::NewFromUtf8(isolate, "setStoreMessageCallback", String::kInternalizedString),
+        FunctionTemplate::New(isolate, jsSetStoreMessageCallback)->GetFunction());
 }
 
 NODE_MODULE(rulesjs, init)