Refined dispatcher and threading. Added lifecycle registry

core/__init__.py

@@ -1,3 +1,4 @@
+from .server import server
 from .apirequest import executeRequest
 from .runtime import buildSave, loadSave, buildLog, manageFailure, buildOutput
 #from .execution import buildOperation, buildAction
@@ -5,3 +6,7 @@ from .execution.action import buildAction
 from .execution.operation import buildOperation
 from .execution.executioncontract import executeContract
 #from executioncontract import executionContract
+from .execution.lifecyclemanager import manageLifecycle
+from .execution.lifecyclemanager import registerLifecycle
+from .execution.createthread import buildThread 
+from .server.dispatcher import dispatchRequest 

core/execution/createthread.py

@@ -2,11 +2,10 @@ import threading
 from lifecycleregistry import registerLifecycle
 from lifecyclemanager import manageLifecycle
 
-activethreads = []
+threadlist = []
 threadnum = 0
 
-def buildThread(module, state, lifecycles):
-#for mode in ("continuous", "oneshot")
+def buildThread(module, lifecycles):
     global threadnum
     threadnum += 1
 
@@ -14,7 +13,7 @@ def buildThread(module, state, lifecycles):
 
     thread = threading.Thread(
         target=manageLifecycle,
-        args=(module, state)
+        args=(module, lifecycles)
     )
     thread.start()
 
@@ -23,9 +22,9 @@ def buildThread(module, state, lifecycles):
 
     lifecycles = registerLifecycle (module, lifecycles, registryname, registryvalue)
 
-    activethreads.append(threadname)
+    threadlist.append(thread)
 
-    print(activethreads)
+    print(threadlist)
     print(lifecycles)
 
-    return threadnum, threadname, lifecycles, activethreads
+    return threadnum, threadname, lifecycles, threadlist

core/execution/lifecyclemanager/__init__.py

@@ -0,0 +1,2 @@
+from .lifecyclemanager import manageLifecycle
+from .lifecycleregistry import registerLifecycle

core/execution/lifecyclemanager/lifecyclemanager.py

@@ -0,0 +1,25 @@
+import time
+
+from core.execution.operation import buildOperation
+from core.execution.action import buildAction
+from core import executeContract
+
+def manageLifecycle(module, lifecycles): #lifecycles, state):
+    #print(lifecycles[module.name]["mode"])
+
+    if  lifecycles[module.name]["mode"] == "continuous":
+        while lifecycles[module.name]["mode"] == "continuous":
+            print(lifecycles)
+
+            outputs = executeContract(module)
+
+            #print(lifecycles)
+
+            time.sleep(module.sleeptime)
+
+    elif lifecycles[module.name]["mode"] == "oneshot":
+        print(lifecycles)
+        outputs = executeContract(module)
+
+        #print(lifecycles)
+    return

core/server/__init__.py

@@ -0,0 +1,2 @@
+from . import server
+

core/server/createdispatch.py

@@ -0,0 +1,17 @@
+import importlib
+from lifecyclemanager import manageLifecycle
+from .dispatchstate import buildState
+
+def buildDispatch(request):
+    dispatch = []
+
+    for key in request["module"]:
+        dispatch.append(request["module"][key])
+
+    resolveddispatch = "plugins."+ ".".join(dispatch)
+
+    module = importlib.import_module(resolveddispatch)
+
+    mode = request.get("mode", "oneshot")
+
+    return module, mode

core/server/dispatcher.py

@@ -0,0 +1,22 @@
+from .createdispatch import buildDispatch
+from .dispatchstate import buildState
+from core.execution.createthread import buildThread
+
+import threading
+
+def dispatchRequest(request, lifecycles):
+
+    module, mode = buildDispatch(request)
+
+    mode, lifecycles = buildState(module, mode, lifecycles)
+
+    if mode == "stop":
+        return lifecycles
+
+    else:
+        threadnum, threadname, lifecycles, threadlist = buildThread(module, lifecycles)
+
+    #print("Active Threads")
+    #print(threadlist)
+
+        return threadnum, threadname, lifecycles, threadlist

core/server/dispatchstate.py

@@ -0,0 +1,41 @@
+import requests
+
+from core import registerLifecycle
+
+def filterMode(mode):
+    if mode in ("continuous", "oneshot", "stop"):
+        state = mode
+        return state
+    else: raise ValueError(f"Unsupported lifecycle mode: {mode}")
+
+def buildState(module, mode, lifecycles):
+    state = filterMode(mode)
+
+    registryvalue = state
+    registryname = "mode"
+
+    if state != "stop":
+        lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
+
+    elif state == "stop":
+        if not module.name in lifecycles:
+
+            print(f"No registered lifecycle: {module.name}")
+
+            return None
+
+        elif "mode" in lifecycles[module.name] != "stop":
+            lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
+
+            print(f"Stopping lifecycle: {module.name}")
+
+    #elif mode == "oneshot":
+        #registryvalue = {"mode": "oneshot"}
+        #lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
+
+           # else:
+                #raise ValueError(f"Unsupported lifecycle mode: {mode}")
+
+    #state = registryvalue
+
+    return state, lifecycles

core/server/server.py

@@ -0,0 +1,30 @@
+import socket
+import json
+
+def startServer():
+    from .dispatcher import dispatchRequest
+
+    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+
+    server.bind(("127.0.0.1", 19002))
+
+    lifecycles = {}
+
+    server.listen()
+
+    while True:
+        connection, address = server.accept()
+
+        while True:
+            data = connection.recv(4096)
+
+            request = json.loads(data)
+
+    #        if not data:
+    #            break
+
+            dispatch = dispatchRequest(request, lifecycles)
+
+        connection.close()
+
+    #    server.close()

dispatcher.py

@@ -1,14 +0,0 @@
-from createdispatch import buildDispatch
-from createthread import buildThread
-from dispatchstate import buildState
-import threading
-
-def dispatchRequest(request, lifecycles):
-
-    module, mode = buildDispatch(request)
-
-    state, lifecycles = buildState(module, mode, lifecycles)
-
-    threadnum, threadname, lifecycles, activethreads = buildThread(module, state, lifecycles)
-
-    return threadnum, lifecycles,

dispatchstate.py

@@ -1,33 +0,0 @@
-import requests
-
-from lifecycleregistry import registerLifecycle
-
-def buildState(module, mode, lifecycles):
-    registryname = "state"
-    if mode == "continuous":
-        registryvalue = {"mode": "continuous"}
-
-        lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
-
-    elif mode == "stop":
-        if not module.name in lifecycles:
-            registryvalue ={}
-            print(f"No registered lifecycle: {module.name}")
-            return None
-        elif lifecycles[module.name]["state"]["mode"] != "stop":
-            state = lifecycles[module.name]["state"]
-            state["mode"] = "stop"
-            #lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
-
-            print(f"Stopping lifecycle: {module.name}")
-
-    elif mode == "oneshot":
-        registryvalue = {"mode": "oneshot"}
-        lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
-
-    else:
-        raise ValueError(f"Unsupported lifecycle mode: {mode}")
-
-    state = registryvalue
-
-    return state, lifecycles

initialize.py

@@ -0,0 +1,3 @@
+from core.server.server import startServer
+
+startServer()

lifecyclemanager.py

@@ -1,19 +0,0 @@
-import time
-
-from core.execution.operation import buildOperation
-from core.execution.action import buildAction
-from core import executeContract
-
-def manageLifecycle(module, state):
-    print(state)
-
-    if state["mode"] == "continuous":
-        while state["mode"] == "continuous":
-            outputs = executeContract(module)
-
-            time.sleep(module.sleeptime)
-
-    elif state["mode"] == "oneshot":
-        outputs = executeContract(module)
-
-    return

lifecyclemonitor.py

@@ -0,0 +1,12 @@
+def lifecycleMonitor(lifecycles):
+    while True
+
+        print("Active Threads:")
+            #if not exists module.name in lifecycles:
+                #print("No active threads")
+            if exists module.name in lifecyles
+                for item in lifecycles[module.name]
+                    if item == "thread".__name__:
+                        print(item)
+                    else:
+                        print("No active threads"

plugins/rmass/operations/setavailable.py

@@ -10,7 +10,7 @@ actions = [
     setavailable,
 ]
 
-sleeptime = 180
+sleeptime = 10
 
 skipstep = {
     setavailable: {

trash/createthread.py

@@ -0,0 +1,30 @@
+import threading
+from lifecycleregistry import registerLifecycle
+from lifecyclemanager import manageLifecycle
+
+threadlist = []
+threadnum = 0
+
+def buildThread(module, lifecycles):
+    global threadnum
+    threadnum += 1
+
+    threadname = f"thread.{threadnum}"
+
+    thread = threading.Thread(
+        target=manageLifecycle,
+        args=(module, lifecycles)
+    )
+    thread.start()
+
+    registryname = threadname
+    registryvalue = thread
+
+    lifecycles = registerLifecycle (module, lifecycles, registryname, registryvalue)
+
+    threadlist.append(thread)
+
+    print(threadlist)
+    print(lifecycles)
+
+    return threadnum, threadname, lifecycles, threadlist

trash/dispatcher.py

@@ -0,0 +1,22 @@
+from createdispatch import buildDispatch
+from createthread import buildThread
+from dispatchstate import buildState
+
+import threading
+
+def dispatchRequest(request, lifecycles):
+
+    module, mode = buildDispatch(request)
+
+    mode, lifecycles = buildState(module, mode, lifecycles)
+
+    if mode == "stop":
+        return lifecycles
+
+    else:
+        threadnum, threadname, lifecycles, threadlist = buildThread(module, lifecycles)
+
+    #print("Active Threads")
+    #print(threadlist)
+
+        return threadnum, threadname, lifecycles, threadlist

trash/dispatchstate.py

@@ -0,0 +1,41 @@
+import requests
+
+from lifecycleregistry import registerLifecycle
+
+def filterMode(mode):
+    if mode in ("continuous", "oneshot", "stop"):
+        state = mode
+        return state
+    else: raise ValueError(f"Unsupported lifecycle mode: {mode}")
+
+def buildState(module, mode, lifecycles):
+    state = filterMode(mode)
+
+    registryvalue = state
+    registryname = "mode"
+
+    if state != "stop":
+        lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
+
+    elif state == "stop":
+        if not module.name in lifecycles:
+
+            print(f"No registered lifecycle: {module.name}")
+
+            return None
+
+        elif "mode" in lifecycles[module.name] != "stop":
+            lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
+
+            print(f"Stopping lifecycle: {module.name}")
+
+    #elif mode == "oneshot":
+        #registryvalue = {"mode": "oneshot"}
+        #lifecycles = registerLifecycle(module, lifecycles, registryname, registryvalue)
+
+           # else:
+                #raise ValueError(f"Unsupported lifecycle mode: {mode}")
+
+    #state = registryvalue
+
+    return state, lifecycles

trash/lifecyclemanager.py

@@ -0,0 +1,25 @@
+import time
+
+from core.execution.operation import buildOperation
+from core.execution.action import buildAction
+from core import executeContract
+
+def manageLifecycle(module, lifecycles): #lifecycles, state):
+    #print(lifecycles[module.name]["mode"])
+
+    if  lifecycles[module.name]["mode"] == "continuous":
+        while lifecycles[module.name]["mode"] == "continuous":
+            print(lifecycles)
+
+            outputs = executeContract(module)
+
+            #print(lifecycles)
+
+            time.sleep(module.sleeptime)
+
+    elif lifecycles[module.name]["mode"] == "oneshot":
+        print(lifecycles)
+        outputs = executeContract(module)
+
+        #print(lifecycles)
+    return

trash/lifecycleregistry.py

@@ -0,0 +1,8 @@
+def registerLifecycle(module, lifecycles, registryname, registryvalue):
+    if module.name not in lifecycles:
+        lifecycles[module.name] = {}
+
+
+    lifecycles[module.name][registryname] = registryvalue
+
+    return lifecycles

trash/server.py

@@ -1,6 +1,7 @@
 import socket
 import json
 from dispatcher import dispatchRequest
+#from createthread import buildThread
 
 server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 
@@ -10,10 +11,9 @@ lifecycles = {}
 
 server.listen()
 
-#lifecycles = {}
-
 while True:
     connection, address = server.accept()
+
     while True:
         data = connection.recv(4096)