Adding metadaa and work status

Author: matburt

receptor/connection.py

@@ -12,8 +12,9 @@
 
 
 class Connection:
-    def __init__(self, id_, protocol_obj, buffer_mgr, receptor):
+    def __init__(self, id_, meta, protocol_obj, buffer_mgr, receptor):
         self.id_ = id_
+        self.meta = meta
         self.protocol_obj = protocol_obj
         self.buffer_mgr = buffer_mgr
         self.receptor = receptor

receptor/protocol.py

@@ -100,7 +100,7 @@ async def wait_greeting(self):
 
     def handle_handshake(self, data):
         self.greeted = True
-        self.connection = self.receptor.add_connection(data["id"], self)
+        self.connection = self.receptor.add_connection(data["id"], data.get("meta", {}), self)
         self.loop.create_task(self.watch_queue(data["id"], self.transport))
         self.loop.create_task(self.connection.message_handler(self.incoming_buffer))
 

receptor/receptor.py

@@ -93,9 +93,9 @@ def update_connections(self, connection):
             self.connections[connection.id_] = [connection]
         self.update_connection_manifest(connection.id_)
 
-    def add_connection(self, id_, protocol_obj):
+    def add_connection(self, id_, meta, protocol_obj):
         buffer_mgr = self.config.components_buffer_manager
-        conn = Connection(id_, protocol_obj, buffer_mgr, self)
+        conn = Connection(id_, meta, protocol_obj, buffer_mgr, self)
         self.update_connections(conn)
         return conn
 

receptor/work.py

@@ -10,6 +10,7 @@
 class WorkManager:
     def __init__(self, receptor):
         self.receptor = receptor
+        self.active_work = []
 
     def load_receptor_worker(self, name):
         entry_points = [x for x in filter(lambda x: x.name == name,
@@ -21,6 +22,16 @@ def load_receptor_worker(self, name):
     def get_capabilities(self):
         return [x.name for x in pkg_resources.iter_entry_points('receptor.worker')]
 
+    def add_work(self, env):
+        self.active_work.append(dict(id=env.message_id,
+                                     directive=env.directive,
+                                     sender=env.sender))
+
+    def remove_work(self, env):
+        for work in self.active_work:
+            if env.message_id == work["id"]:
+                self.active_work.remove(work)
+
     async def handle(self, inner_env):
         logger.info(f'Handling work for {inner_env.message_id} as {inner_env.directive}')
         namespace, action = inner_env.directive.split(':', 1)
@@ -32,7 +43,7 @@ async def handle(self, inner_env):
             except AttributeError:
                 logger.exception(f'Could not load action {action} from {namespace}')
                 raise exceptions.InvalidDirectiveAction(f'Invalid action {action} for {namespace}')
-
+            self.add_work(inner_env)
             responses = action_method(inner_env)
             async for response in responses:
                 serial += 1
@@ -57,5 +68,6 @@ async def handle(self, inner_env):
                 serial=serial,
                 code=1,
             )
+            self.remove_work(inner_env)
             await self.receptor.router.send(enveloped_response)