messenger v0.9

agent.py

@@ -13,74 +13,44 @@ class Agent:
 
     """
     def __init__(self, connection):
-        self.connection = connection
+        self.service = None
-
+        self.client_connection = connection
-    # Submit a request for info, return response.
-    # Must disclose what service it's for.
-    async def fetch(self, entities, service):
-
-        # Assert list.
-        if type(entities) == str:
-            entities = [entities]
-
-        print(f"Fetching {[e for e in entities]} for {service}")
-
-        # Build intent.
-        params = {
-            "service": service,
-            "entities": entities
-        }; intent = typs.Intent("fetch", params)
-
-        # Send intent to agent, wait for response.
-        await self.send_agent_intent(intent)
-        response = await self.recv_agent_intent()
-
-        # Parse response and return entities.
-        entities = response.params.get("found")
-        print(f"Fetch response: {entities}")
-        return entities
-
-    # Text the client something. Also, let agent know
-    #   when your service is complete via the complete
-    #   attribute.
-    # Must disclose what service it's for.
-    async def notify(self, text, end=False):
-
-        print(f"Notifying: {text} for {self.service}")
 
+    def fetch(self, entity, invocation):
+        # Create the intent.
         params = {
+            "entity": entity, 
             "service": self.service,
+            "invocation": invocation
+        }
+        intent = typs.Intent(name="fetch", params=params)
+
+        self.send_agent_intent(intent)
+        intent = self.recv_agent_intent()
+
+        response = intent.params.get("response")
+        return response
+
+    def notify(self, text, end=False):
+        # Create the intent.
+        params = {
             "text": text, 
+            "service": "message", 
             "end": end
         }
-
+        intent = typs.Intent(name="notify", params=params)
-        # Build the intent
-        intent = typs.Intent("notify", params)
 
         # Send intent to agent.
-        await self.send_agent_intent(intent)
+        self.send_agent_intent(intent)
-
-    # Get user info.
-    async def sync(self):
-
-        intent = typs.Intent("sync", params={})
-
-        # Send intent to agent.
-        await self.send_agent_intent(intent)
-
-        # Recv and return profile.
-        intent = await self.recv_agent_intent()
-        profile = intent.params.get("profile")
-        return profile
 
     async def recv_agent_intent(self):
-        package = await self.connection.recv()
+        package = await self.client_connection.recv()
         intent = self.__decode(package)
         return intent
 
     async def send_agent_intent(self, intent):
         package = self.__encode(intent)
-        await self.connection.send(package)
+        await self.client_connection.send(package)
 
     def __encode(self, intent: typs.Intent):
         package = json.dumps(intent.__dict__)

mail.py

@@ -8,9 +8,10 @@ class Mail:
 	"""Session interface for mailroom.
 
 	Two main IO methods. Will translate intents to mail and vice versa.
+	Should be a parent class of Session.
 	"""
 	def __init__(self):
-		pass
+		self.crimata_id = False
 
 	# Receive intent.
 	async def mailbox_recv(self):
@@ -21,6 +22,7 @@ class Mail:
 	# Send intent.
 	def mailbox_send(self, intent):
 		mail = self.__encode(intent)
+		print(f"Putting message into outbox {self.crimata_id}")
 		mailroom.put_mail(self.crimata_id, mail)
 
 	def __encode(self, intent: typs.Intent) -> typs.Mail:
@@ -33,8 +35,9 @@ class Mail:
 
 	@staticmethod
 	def __decode(mail: typs.Mail) -> typs.Intent:
-		name = "notify"
+		name = "message"
 		params = {
+			"owner": mail.owner,
 			"text": mail.text #dev only
 		}
 		intent = typs.Intent(name, params)

message.py

@@ -8,8 +8,6 @@ import mailroom
 # Independent loop that updates mailboxes.
 async def messenger():    
 
-    print("Running Messenger")   
-
     while True:
 
         # Yield if no mailboxes.
@@ -19,7 +17,6 @@ async def messenger():
 
         # Grab a mailbox.
         mailbox = mailroom.que.get()
-        print(f"Handling mailbox: {mailbox.name}.")
 
         # Handle every message in outbox.
         while not mailbox.outbox.empty():
@@ -27,6 +24,7 @@ async def messenger():
 
             # Get mailbox of target and put mail there.
             target_mailbox = mailroom.get_mailbox(message.target)
+            print(f"{mailbox.name} -> {target_mailbox.name}: {message.text}")
             await target_mailbox.inbox.put(message)
 
         # Put the mailbox back

schemes.py

@@ -3,27 +3,29 @@
 
 class Schemes:
 
-	def __init__(self):
+	def handle_intent(self, intent):
-		self.service = None
+		print(f"Handling {intent.name}.")  
 
-	async def handle_intent(self, intent):
+		# Set the service.
-		print(f"Handling intent {intent.name}.")  
-
-		# Set the service
 		self.service = intent.name
 		 
+		# Run intent endpoint.
 		if intent.name == "init":
-			await self.init(intent)
+			self.init(intent)
+
 		if intent.name == "message":
 			self.message(intent)
 
-	async def init(self, intent):
+	# Init session by setting Crimata ID.
+	def init(self, intent):
 		self.crimata_id = intent.params.get("crimata_id")
-		await self.notify("Messaging is live.")
+		print(f"Initalized for {self.crimata_id}")
 
+	# Push message to anoter Crimata ID.
 	def message(self, intent):
 		text = intent.params.get("text")
 		target = intent.params.get("target") #crimata_id
+
 		print(f"Messaging {target}: '{text}'")
 
 		# Uses mail.Mail interface.

server.py

@@ -1,18 +1,36 @@
 # server.py
 
+import json
+import uuid
 import asyncio
 import websockets
 
+import typs
 import session
 
 HOST = "localhost"
 PORT = 8764
 
+token_refs = {}
 
 # Makes instance of a Messenger session for every client that connects.
-async def launch_session(connection, path):
+async def launch_session(token, connection):
+
+    # Create the session.
     s = session.Session(connection)
 
+    # Login if token.
+    if token in token_refs.keys():
+        crimata_id = token_refs.get(token)
+        s.crimata_id = crimata_id
+
+        print(f"Reconnected: {crimata_id}")
+
+    # Update client's login token.
+    intent = typs.Intent("token", params={"text": token})
+    await s.send_agent_intent(intent)
+
+    # Run core loops.
     await asyncio.gather(
 
         s.do_agent_intents(),
@@ -21,7 +39,22 @@ async def launch_session(connection, path):
 
     )
 
+    # Update state before end.
+    token_refs.update({token: s.crimata_id})
+
+    print("Ended session.")
+
+async def main(connection, path):
+    token = await connection.recv()
+
+    # Create Session ID if None.
+    if token not in token_refs.keys():
+        token = str(uuid.uuid4())
+
+    # Start session with token.
+    await launch_session(token, connection) #! Not stopping on disconnect
+
 # Run run for every new connection.
 async def lift():
     print(f"Listening for connections on {HOST}:{PORT}")
-    await websockets.serve(launch_session, HOST, PORT)
+    await websockets.serve(main, HOST, PORT)

session.py

@@ -2,6 +2,7 @@
 
 import time
 import asyncio
+import websockets
 
 import mail
 import agent
@@ -17,65 +18,48 @@ class Session(schemes.Schemes, agent.Agent, mail.Mail):
     """
     def __init__(self, connection):
         agent.Agent.__init__(self, connection)
+        mail.Mail.__init__(self)
 
-        self.crimata_id = False
+        self.on = True
 
-        print("Launched new session")
+        print("Launched sesssion.")
 
     # Recv intents run desired endpoint.
     async def do_agent_intents(self):
-        while True:
+        while self.on:
 
-            # Recv intent from agent.
+            try:
-            intent = await self.recv_agent_intent()
+                intent = await asyncio.wait_for(
-            print(f"Intent: {intent.name}")
+                    self.recv_agent_intent(), timeout=0.2)
+            except asyncio.TimeoutError:
+                continue
+            except websockets.exceptions.ConnectionClosed:
+                await self.shutdown()
+                continue
 
             # Call corresponding endpoint.
-            await self.handle_intent(intent)
+            self.handle_intent(intent)
-
-            await asyncio.sleep(0.1)
 
     # Send mail back to agent.
     async def deliver_mail(self):
-        while True:
+        while self.on:
 
-            # Sleep until Crimata ID.
             if not self.crimata_id:
-                await asyncio.sleep(1)
+                await asyncio.sleep(0.1)
                 continue
 
-            intent = await self.mailbox_recv()
+            # Try recv.
+            try:
+                intent = await asyncio.wait_for(
+                    self.mailbox_recv(), timeout=0.2)
+            except asyncio.TimeoutError:
+                continue
+
+            # Send message to Agent.
+            print("Sending message")
             await self.send_agent_intent(intent)
 
-            await asyncio.sleep(0.1)
+    # Shutdown protocol.
-
+    async def shutdown(self):
-
+        print("Shutting down session.")
-
+        self.on = False
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-    # async def hello(self):
-
-    #     intent = await self.recv_agent_intent()
-
-    #     response = await self.fetch("crimata_id", service="message")
-    #     self.profile = response.get("profile")
-
-    #     target = intent.params.get("target")
-    #     await self.notify(f"Messenger APP: Received Message Intent", 
-    #         service="message", complete=True)
-
-    #     print("Shutting down in 10s")
-
-    #     await asyncio.sleep(10)