Added request timeout handling

core/apirequest/execute.py

@@ -45,13 +45,15 @@ def executeRequest(action, outputs, resolved):
             )
             return response, outputs
 
-        except (requests.ConnectTimeout, requests.ConnectionError) as error:
+        except (requests.Timeout, requests.ConnectionError) as error:
             print(f"{type(error).__name__} during {action.name}: {error}")
             session = getSession(reset=True)
 
             if attempt == 0:
                 print(f"Retrying {action.name}")
 
+    return None, outputs
+
 #    try:
 #        request, outputs = buildRequest(action, outputs, resolved)
 #
@@ -87,5 +89,5 @@ def executeRequest(action, outputs, resolved):
 #    except requests.Timeout as error:
 #        print(f"Request timed out during {action.name}: {error}")
 #        session = getSession(reset=True)
-
-    return response, outputs
+#
+#    return response, outputs

core/execution/action/action.py

@@ -1,5 +1,6 @@
 from core.apirequest import executeRequest
 from core.runtime.runtimecontract import buildRuntimeAttr, buildPreRuntimeAttr
+from core.runtime.stopsignals import addStopSignal
 
 
 def buildAction (action, outputs):
@@ -8,12 +9,16 @@ def buildAction (action, outputs):
 
     if hasattr(action, "allowaction"):
         if not action.allowaction:
-            return outputs, resolved
+            return outputs, None
 
     if match:
-        return outputs, resolved
+        return outputs, None
     
     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

@@ -1,6 +1,7 @@
 import logging
 from core.runtime.managefailure import manageFailure
 from core.execution.action.action import buildAction
+from core.runtime.stopsignals import checkDependency
 #from core.runtime.skipstep import skipStep
 #from core.runtime.recursive import recursiveSet
 
@@ -9,16 +10,16 @@ def buildOperation(definition, outputs):
     from core.execution.executioncontract import executeContract
     actions = definition.actions
     responses = []
-
+    outputs["stopsignals"] = []
     for action in actions:
+        stop = checkDependency(action, outputs)
 
-        #if action == "recursiveset":
-        #if hasattr(action, "recursiveset"):
-        #    from core.runtime.recursive import recursiveSet
-        #    outputs, responses = recursiveSet(action, outputs, responses)
+        #print(f"STOP SIGNAL DEBUGGING during {action.name}")
+        #print(outputs["stopsignals"])
+
+        if stop:
+            continue
 
-        #else:
-            #outputs, response = buildAction(action, outputs)
         outputs, response = executeContract(action, outputs)
         
         #if hasattr(action, recursive)

core/execution/recursive.py

@@ -1,6 +1,8 @@
+from core.execution import action
 from core.execution.action.action import buildAction
 from core.runtime.runtimecontract import buildPreRuntimeAttr, buildRuntimeAttr
 from core.apirequest import executeRequest
+from core.runtime.stopsignals import checkDependency
 
 def buildRecursive(definition, outputs):
     responses = []
@@ -26,10 +28,23 @@ def buildRecursive(definition, outputs):
     for item in items:
         outputs[definition.recursiveset["outputlabel"]] = item
 
+        outputs["stopsignals"] = []
         for job in recursiveset:
+            stop = checkDependency(job, outputs)
+
+            #print(f"STOP SIGNAL DEBUGGING during {job.name}")
+            #print(outputs["stopsignals"])
+            #print(stop)
+
+            if stop:
+                #print("STOP SIGNAL DETECTED")
+                #print(f"Stopping {job.name} due to stop signal")
+                continue
+
             #print(f"RECURSIVE DEBUG DURING {job.name}")
             #print(item)
             #outputs, response = buildAction(job, outputs)
+            
             outputs, response = executeContract(job, outputs)
             responses.append(response)
 
@@ -38,4 +53,6 @@ def buildRecursive(definition, outputs):
     if not retain:
         outputs.pop(source, None)
 
+    outputs.pop("stopsignals", None)
+
     return outputs, responses

core/runtime/runtimecontract.py

@@ -1,9 +1,11 @@
+from core.execution.action import action
 from core.runtime import buildSelect, buildOutput, buildSave, loadSave, saveRow, readRow, buildLog
 from core.runtime.image import buildImage
 #from core.runtime.select import buildSelect3
 from core.runtime.resolver import resolveItem, extractItem
 from core.runtime.applyconditions import applyConditions
 from core.runtime.responsepolicy import responsePolicy
+from core.runtime.stopsignals import addStopSignal
 
 def buildPreRuntimeAttr(action, outputs):
     outputs = loadSave(action, outputs)
@@ -14,6 +16,9 @@ def buildPreRuntimeAttr(action, outputs):
 
     match, filter = applyConditions(action, resolved, outputs)
 
+    if match:
+        outputs = addStopSignal(action, outputs)
+    
     return outputs, resolved, match, filter
     
 
@@ -22,11 +27,15 @@ def buildRuntimeAttr(action, outputs, resolved, response):
 
     outputs = buildOutput(response, action, outputs)
 
+    if not response.ok:
+        outputs = addStopSignal(action, outputs)
+        return outputs, response
+
     resolved = extractItem(action, outputs, resolved)
 
     outputs = buildImage(response, action, outputs)
     
-    select_data, action, outputs = buildSelect(action, outputs)
+    action, outputs = buildSelect(action, outputs)
 
     buildSave(action, outputs)
 

core/runtime/select.py

@@ -30,12 +30,12 @@ def buildSelect(action, outputs):
             if select_data is None:
                 print(f"No data selected during {action.name}")           
 
-            return select_data, action, outputs
+            return action, outputs
         except KeyError as error:
             print(f"Error in selection of {error} during {action.name}")
 
         except TypeError as error:
             print(f"TypeError during select: {error} during {action.name}")
 
-    return select_data, action, outputs
+    return action, outputs
     

core/runtime/stopsignals.py

@@ -0,0 +1,17 @@
+def checkDependency(action, outputs):
+    if hasattr(action, "depends"):
+        stop = any(
+                dependency in outputs["stopsignals"]
+                for dependency in action.depends
+            )
+        
+        if stop:
+            return True
+
+    return False
+
+def addStopSignal(action, outputs):
+    if hasattr(action, "depends"):
+        outputs["stopsignals"].append(action.alias)
+
+    return outputs

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/getrmassuseravatar.py

@@ -4,7 +4,9 @@ name = "Get User Avatar"
 
 kind = "action"
 
-alias = "getavatar"
+alias = "getrmassuseravatar"
+
+depends = []
 
 allowaction = bool(os.environ.get("UPDATE_AVATARS", True))
 

plugins/rmass/actions/pullmessagelist.py

@@ -4,7 +4,7 @@ name = "Get message list"
 
 kind = "action"
 
-alias = "pullmessages"
+alias = "pullmessagelist"
 
 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
@@ -38,16 +48,16 @@ resolve = {
 }
 
 keys = [
-    #{
-    #    "keytitle": "message",
-    #    "keyvalue": {
-    #        "resolved": "message_body",
-    #    },
-    #},
     {
         "keytitle": "message",
-        "keyvalue": "Here's a photo!",
+        "keyvalue": {
+            "resolved": "message_body",
         },
+    },
+    #{
+    #    "keytitle": "message",
+    #    "keyvalue": "Here's a photo!",
+    #},
     #{
     #    "keytitle": "profileId",
     #    "keyvalue": {

plugins/rmass/actions/rmassattachmenttosynapse.py

@@ -2,7 +2,11 @@ name = "Upload Attachments to Synapse"
 
 kind = "action"
 
-alias = "uploadattachment"
+alias = "rmassuploadattachmenttosynapse"
+
+depends = [
+    "getrmassattachments",
+]
 
 showstatus = True
 

plugins/rmass/actions/rmassavatartosynapse.py

@@ -4,7 +4,11 @@ name = "Upload Avatar to Synapse"
 
 kind = "action"
 
-alias = "uploadavatar"
+alias = "rmassavatartosynapse"
+
+depends = [
+    "getrmassuseravatar",
+]
 
 allowaction = os.environ.get("UPDATE_AVATARS", True)
 

plugins/rmass/actions/rmassmessagepush.py

@@ -4,6 +4,13 @@ from . import synapsenotify
 
 name = "RMASS Message Push"
 
+alias = "rmassmessagepush"
+
+depends = [
+    "synapsegetroomalias",
+    "synapsegetlastmessage",
+]
+
 kind = "action"
 
 showstatus = True
@@ -86,6 +93,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/savermassmessageid.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/synapseassignavatar.py

@@ -2,6 +2,13 @@ import os
 
 name = "Synapse Assign Avatar"
 
+alias = "synapseassignavatar"
+
+depends = [
+    "getrmassuseravatar",
+    "rmassavatartosynapse",
+]
+
 kind = "action"
 
 showstatus = True

plugins/rmass/actions/synapseassigndmavatar.py

@@ -2,6 +2,15 @@ import os
 
 name = "Synapse Assign DM Room Avatar"
 
+alias = "synapseassigndmavatar"
+
+depends = [
+    "getrmassuseravatar",
+    "rmassavatartosynapse",
+    "synapsegetroomid",
+    "synapsecreatedmroom",
+]
+
 kind = "action"
 
 showstatus = True

plugins/rmass/actions/synapseassignusername.py

@@ -6,6 +6,10 @@ kind = "action"
 
 showstatus = True
 
+alias = "synapseassignusername"
+
+#depends = []
+
 #astoken = os.environ["AS_TOKEN"]
 
 resolve = {

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/synapsecreatedmroom.py

@@ -6,6 +6,10 @@ kind = "action"
 
 showstatus = True
 
+alias = "synapsecreatedmroom"
+
+depends = []
+
 resolve = {
     "get": {
         "sender": ["messageheaders", "userCard", "id"],

plugins/rmass/actions/synapsegetattachment.py

@@ -4,7 +4,12 @@ kind = "action"
 
 showstatus = True
 
-alias = "getsynapseattachment"
+alias = "synapsegetattachment"
+
+depends = [
+    "synapsegetroomalias",
+    "synapsegetlastmessage",
+]
 
 ## PRE-RUNTIME ACTIONS
 
@@ -56,4 +61,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,13 @@
 name = "Get Last Synapse Message"
 
+alias = "synapsegetlastmessage"
+
 kind = "action"
 
+depends = [
+    "synapsegetroomalias",
+]
+
 showstatus = True
 
 isrequest = True
@@ -11,7 +17,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"],
@@ -41,16 +47,22 @@ outputs = {
 }
 
 filters = [
+#    {
+#        "filterkey": "rmass_sent_status",
+#        "operator": "==",
+#        "filtervalue": "fail",
+#        "print": f"Failed to send message to RMASS. Bypassing {name}...",
+#    },
+#    {
+#        "filterkey": "rmass_sent_status",
+#        "operator": "==",
+#        "filtervalue": "error",
+#       "print": f"Failed to send message to RMASS. Bypassing {name}...",
+#    },
     {
-        "filterkey": "rmass_sent_status",
+        "filterkey": "event_id",
         "operator": "==",
-        "filtervalue": "fail",
-        "print": f"Failed to send message to RMASS. Bypassing save during {name}...",
-    },
-    {
-        "filterkey": "rmass_sent_status",
-        "operator": "==",
-        "filtervalue": "error",
-        "print": f"Failed to send message to RMASS. Bypassing save during {name}...",
+        "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 = [
-#    {
-#        "filterkey": "outmessageitem",
-#        "operator": "in",
-#        "filtervalue": "state_key",
-#        "print": f"Excluding state.key event during {name}",
-#    },
+readrow = {
+    "table": "message_mapping",
+    "columns": [
+        {"rmass_message_id": "INTEGER PRIMARY KEY"},
+        {"matrix_event_id": "TEXT NOT NULL UNIQUE"},
+    ],
+    "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/actions/synapsegetroomid.py

@@ -4,6 +4,12 @@ name = "Synapse Get Room ID"
 
 kind = "action"
 
+alias = "synapsegetroomid"
+
+#depends = [
+#    "synapsecreatedmroom",
+#]
+
 showstatus = True
 
 resolve = {

plugins/rmass/actions/synapsepushattachments.py

@@ -2,6 +2,13 @@ from . import synapsenotify
 
 name = "Synapse Attachment Push"
 
+alias = "synapsepushattachments"
+
+depends = [
+    "getrmassattachments",
+    "rmassattachmenttosynapse",
+]
+
 kind = "action"
 
 showstatus = True

plugins/rmass/actions/synapseregister.py

@@ -6,6 +6,10 @@ kind = "action"
 
 showstatus = True
 
+alias = "synapseregister"
+
+#depends = []
+
 resolve = {
     "get": {
         "sender": ["messageheaders", "userCard", "id"],

plugins/rmass/operations/getrmassconversations.py

@@ -3,10 +3,16 @@ from plugins.rmass.operations import synapsepushmessageitems
 
 name = "Get RMASS Conversations and Synapse Provisioning"
 
+alias = "getrmassconversations"
+
+depends = [
+    "pullmessagelist",
+]
+
 kind = "recursiveset"
 
 recursiveset = {
-    "source": "pullmessages_select",
+    "source": "pullmessagelist_select",
     "outputlabel": "messageheaders",
     "actions": [
         synapseregister, 

plugins/rmass/operations/rmassattachmentpushflow.py

@@ -0,0 +1,20 @@
+from plugins.rmass.actions import synapsegetattachment, getrmassattachmenturl, synapseattachmenttormass, getrmassattachmentid, rmassattachmentpush
+
+name = "RMASS Attachment Push"
+
+alias = "rmassattachmentpushflow"
+
+depends = [
+    "synapsegetroomalias",
+    "synapsegetlastmessage",
+]
+
+kind = "operation"
+
+actions = [
+    synapsegetattachment,
+    getrmassattachmenturl,
+    synapseattachmenttormass,
+    getrmassattachmentid,
+    rmassattachmentpush,
+]

plugins/rmass/operations/rmassmessagepush.py

@@ -1,4 +1,5 @@
-from plugins.rmass.actions import rmassmessagepush, synapsegetroomalias, refresh_tokens, synapsegetlastmessage, getrmassconversation, synapsegetattachment, getrmassattachmenturl, synapseattachmenttormass, getrmassattachmentid, rmassattachmentpush
+from plugins.rmass.actions import rmassmessagepush, savermassmessageid, synapsegetroomalias, synapsegetlastmessage
+from plugins.rmass.operations import rmassattachmentpushflow
 
 name = "RMASS Message Push"
 
@@ -9,17 +10,11 @@ recursiveset = {
     #"retain": False,
     "outputlabel": "outmessageitem",
     "actions": [
-        #refresh_tokens,
         synapsegetroomalias,
+        synapsegetlastmessage,
         rmassmessagepush,
-        synapsegetlastmessage,
-        synapsegetattachment,
-        getrmassattachmenturl,
-        synapseattachmenttormass,
-        getrmassattachmentid,
-        rmassattachmentpush,
-        synapsegetlastmessage,
-        getrmassconversation,
+        rmassattachmentpushflow,
+        savermassmessageid,
     ],
 }        
 

tokens.json

@@ -0,0 +1 @@
+{"access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzUxMiJ9.eyJleHAiOjE3OTMwNTE5ODUsIm5iZiI6MTc5MzA1MTgwNSwic3ViIjoiMjM4MzAyIiwidXNlcm5hbWUiOiJYeHhvdGlxIiwidXNlclR5cGUiOjEsInN0YWdlTmFtZSI6Ilh4eG90aXEiLCJhcHByb3ZlZCI6MSwibG9jYXRpb25JZCI6NDcyLCJzZWFyY2hDaXR5IjoiYXRsYW50YSIsImNvdW50cnlTaG9ydCI6IlVTIiwic2Vzc2lvbkNvdW50cnlDb250aW5lbnQiOiJOQSIsImlzRnJvemVuIjowLCJsb2NhdGlvbkxhYmVsIjoiQXRsYW50YSwgR0EiLCJpc3MiOiJodHRwczovL3JlbnRtYXNzZXVyLmNvbSIsImlhdCI6MTc5MzA1MTg2NX0.pMPPsvgYfsRhTj7NJFc0hj3N-8gP1Th3oX9NQce9e2-NJw7ANxFv-7sdmeDE71_B71He7cych-kW5RFuJpc2ZA", "refresh_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzUxMiJ9.eyJleHAiOjE3OTU2NDM4NjUsIm5iZiI6MTc5MzA1MTgwNSwianRpIjoiU1BJbnZpSXBmTE5WY1NVQ1ZOQ0VvRUdGUWEwOVIrN3dKOHJuNGZqYU5ZMD0iLCJpc3MiOiJodHRwczovL3JlbnRtYXNzZXVyLmNvbSIsImlhdCI6MTc5MzA1MTg2NSwic3ViIjoiMjM4MzAyIn0.2N1mzB9BYKgd5tMK10ZuBxW0zqNDDmhN8JtVoQAoeKHcm9JtNzdRa4LfsNPiJ1MOzkXBkTBuktmUI8AiDq78YQ"}