Introduced stop signals and dependency gating for operations. Added mechanism to RMASS message sending protocol (Confirmed Working)

core/execution/action/action.py

@@ -8,12 +8,18 @@ def buildAction (action, outputs):
 
     if hasattr(action, "allowaction"):
         if not action.allowaction:
-            return outputs, resolved
+            return outputs, None
 
     if match:
-        return outputs, resolved
+        if hasattr(action, "depends"):
+            outputs["stopsignals"].append(action.alias)
+            return outputs, None#resolved
     
     response, outputs = executeRequest(action, outputs, resolved)
+
+    if response is None:
+        return outputs, None
+
     outputs, response = buildRuntimeAttr(action, outputs, resolved, response)
 
     return outputs, response

core/execution/operation/operation.py

@@ -9,16 +9,19 @@ def buildOperation(definition, outputs):
     from core.execution.executioncontract import executeContract
     actions = definition.actions
     responses = []
-
+    outputs["stopsignals"] = []
     for action in actions:
+        if hasattr(action, "depends"):
+            stop =any(
+                dependency in outputs["stopsignals"]
+                for dependency in action.depends
+            )
+                #stop = True
+                #break
 
-        #if action == "recursiveset":
+            if stop:
-        #if hasattr(action, "recursiveset"):
+                continue
-        #    from core.runtime.recursive import recursiveSet
-        #    outputs, responses = recursiveSet(action, outputs, responses)
 
-        #else:
-            #outputs, response = buildAction(action, outputs)
         outputs, response = executeContract(action, outputs)
         
         #if hasattr(action, recursive)

core/execution/recursive.py

@@ -26,7 +26,19 @@ def buildRecursive(definition, outputs):
     for item in items:
         outputs[definition.recursiveset["outputlabel"]] = item
 
+        outputs["stopsignals"] = []
         for job in recursiveset:
+            if hasattr(job, "depends"):
+                stop =any(
+                    dependency in outputs["stopsignals"]
+                    for dependency in job.depends
+                )
+                    #stop = True
+                    #break
+
+                if stop:
+                    continue
+
             #print(f"RECURSIVE DEBUG DURING {job.name}")
             #print(item)
             #outputs, response = buildAction(job, outputs)
@@ -38,4 +50,6 @@ def buildRecursive(definition, outputs):
     if not retain:
         outputs.pop(source, None)
 
+    outputs.pop("stopsignals", None)
+
     return outputs, responses

plugins/rmass/actions/getrmassattachmentid.py

@@ -4,6 +4,12 @@ kind = "action"
 
 alias = "getrmassattachmentid"
 
+depends = [
+    "synapsegetattachment",
+    "getrmassattachmenturl",
+    "synapseattachmenttormass",
+]
+
 showstatus = True
 
 resolve = {

plugins/rmass/actions/getrmassattachmenturl.py

@@ -4,6 +4,11 @@ kind = "action"
 
 alias = "getrmassattachmenturl"
 
+depends = [
+    "synapsegetroomalias",
+    "synapsegetattachment",
+]
+
 showstatus = True
 
 resolve = {
@@ -47,4 +52,10 @@ filters = [
         "filtervalue": "m.text",
         "print": f"No image detected. Bypassing attachment push",
     },
+    {
+        "filterkey": "event_id",
+        "operator": "==",
+        "filtervalue": {"outputs": "db_event_id"},
+        "print": f"Message already delivered. Bypassing {name}",
+    },
 ]

plugins/rmass/actions/getrmassconversation.py

@@ -2,7 +2,13 @@ name = "Save RMASS Message ID"
 
 kind = "action"
 
-alias = "getmessage"
+alias = "savermassmessageid"
+
+depends = [
+    "synapsegetroomalias",
+    "rmassmessagepush",
+    "rmassattachmentpush",
+]
 
 showstatus = True
 

plugins/rmass/actions/rmassattachmentpush.py

@@ -4,6 +4,16 @@ from . import synapsenotify
 
 name = "RMASS Attachment Push"
 
+alias = "rmassattachmentpush"
+
+depends = [
+    "synapsegetroomalias",
+    "synapsegetattachment",
+    "getrmassattachmenturl",
+    "synapseattachmenttormass",
+    "getrmassattachmentid",
+]
+
 kind = "action"
 
 showstatus = True

plugins/rmass/actions/rmassmessagepush.py

@@ -4,6 +4,12 @@ from . import synapsenotify
 
 name = "RMASS Message Push"
 
+alias = "rmassmessagepush"
+
+depends = [
+    "synapsegetroomalias",
+]
+
 kind = "action"
 
 showstatus = True
@@ -86,6 +92,12 @@ filters = [
         "filtervalue": "m.image",
         "print": f"Image detected. Pushing attachment",
     },
+    {
+        "filterkey": "event_id",
+        "operator": "==",
+        "filtervalue": {"outputs": "db_event_id"},
+        "print": f"Message already delivered. Bypassing {name}",
+    },
 #    {
 #        "filterkey": "matrix_sender",
 #        "operator": "!=",

plugins/rmass/actions/synapseattachmenttormass.py

@@ -2,7 +2,12 @@ name = "Upload Attachments to RMASS"
 
 kind = "action"
 
-alias = "uploadrmassattachment"
+alias = "synapseattachmenttormass"
+
+depends = [
+    "synapsegetattachment",
+    "getrmassattachmenturl",
+]
 
 showstatus = True
 

plugins/rmass/actions/synapsegetattachment.py

@@ -4,7 +4,11 @@ kind = "action"
 
 showstatus = True
 
-alias = "getsynapseattachment"
+alias = "synapsegetattachment"
+
+depends = [
+    "synapsegetroomalias",
+]
 
 ## PRE-RUNTIME ACTIONS
 
@@ -56,4 +60,10 @@ filters = [
         "filtervalue": "m.text",
         "print": f"No image detected. Bypassing attachment push",
     },
+    {
+        "filterkey": "event_id",
+        "operator": "==",
+        "filtervalue": {"outputs": "db_event_id"},
+        "print": f"Message already delivered. Bypassing {name}",
+    },
 ]

plugins/rmass/actions/synapsegetlastmessage.py

@@ -1,7 +1,14 @@
 name = "Get Last Synapse Message"
 
+alias = "synapsegetlastmessage"
+
 kind = "action"
 
+depends = [
+    "synapsegetroomalias",
+    "rmassmessagepush",
+]
+
 showstatus = True
 
 isrequest = True
@@ -11,7 +18,7 @@ method = "GET"
 resolve = {
     "get": {
         #"sender": ["room_name"],
-        #"event_id": ["outmessageitem", "event_id"],
+        "event_id": ["outmessageitem", "event_id"],
         #"message_body": ["outmessageitem", "content", "body"],
         "room_id": ["outmessageitem", "room_id"],
         #"timestamp": ["outmessageitem", "origin_server_ts"],
@@ -53,4 +60,10 @@ filters = [
         "filtervalue": "error",
         "print": f"Failed to send message to RMASS. Bypassing save during {name}...",
     },
+    {
+        "filterkey": "event_id",
+        "operator": "==",
+        "filtervalue": {"outputs": "db_event_id"},
+        "print": f"Message already delivered. Bypassing {name}",
+    },
 ]

plugins/rmass/actions/synapsegetroomalias.py

@@ -4,6 +4,8 @@ kind = "action"
 
 showstatus = True
 
+alias = "synapsegetroomalias"
+
 ## PRE-RUNTIME ACTIONS
 
 resolve = {
@@ -11,7 +13,7 @@ resolve = {
         "room_id": ["outmessageitem", "room_id"],
 #        "matrix_sender": ["outmessageitem", "sender"],
 #        "msg_type": ["outmessageitem", "type"],
-#        "echo_tag": ["outmessageitem", "content", "com.loxthelion.rmass_message_id"],
+        "event_id": ["outmessageitem", "event_id"],
     },
 }
 
@@ -42,13 +44,28 @@ outputs = {
     "room_alias": "alias"
 }
 
-#filters = [
+readrow = {
-#    {
+    "table": "message_mapping",
-#        "filterkey": "outmessageitem",
+    "columns": [
-#        "operator": "in",
+        {"rmass_message_id": "INTEGER PRIMARY KEY"},
-#        "filtervalue": "state_key",
+        {"matrix_event_id": "TEXT NOT NULL UNIQUE"},
-#        "print": f"Excluding state.key event during {name}",
+    ],
-#    },
+    "searchcolumn": "matrix_event_id",
+    "rowsearchvalue": [
+        {"resolved": "event_id"},
+    ],
+    "export": {
+        1: "db_event_id",
+    }
+}
+
+filters = [
+    {
+        "filterkey": "event_id",
+        "operator": "==",
+        "filtervalue": {"outputs": "db_event_id"},
+        "print": f"Message already delivered. Bypassing {name}",
+    },
 #    {
 #        "filterkey": "matrix_sender",
 #        "operator": "!=",
@@ -67,4 +84,4 @@ outputs = {
 #        "filtervalue": None,
 #        "print": f"Excluding imported message event during {name}",
 #    },
-#]
+]

plugins/rmass/operations/rmassmessagepush.py

@@ -18,7 +18,7 @@ recursiveset = {
         synapseattachmenttormass,
         getrmassattachmentid,
         rmassattachmentpush,
-        synapsegetlastmessage,
+        #synapsegetlastmessage,
         getrmassconversation,
     ],
 }