"""
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.
+ def fetch(self, entity, invocation):
+ # Create the 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
+ "entity": entity,
+ "service": self.service,
+ "invocation": invocation
+ }
+ intent = typs.Intent(name="fetch", params=params)
- # 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):
+ self.send_agent_intent(intent)
+ intent = self.recv_agent_intent()
- print(f"Notifying: {text} for {self.service}")
+ response = intent.params.get("response")
+ return response
+ def notify(self, text, end=False):
+ # Create the intent.
params = {
- "service": self.service,
- "text": text,
+ "text": text,
+ "service": "message",
"end": end
}
-
- # Build the intent
- intent = typs.Intent("notify", params)
-
- # Send intent to agent.
- await self.send_agent_intent(intent)
-
- # Get user info.
- async def sync(self):
-
- intent = typs.Intent("sync", params={})
+ intent = typs.Intent(name="notify", params=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
+ self.send_agent_intent(intent)
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__)
"""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):
# 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:
@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)
return intent
-
\ No newline at end of file
+
\ No newline at end of file
# Independent loop that updates mailboxes.
-async def messenger():
-
- print("Running Messenger")
+async def messenger():
while True:
# Grab a mailbox.
mailbox = mailroom.que.get()
- print(f"Handling mailbox: {mailbox.name}.")
# Handle every message in outbox.
while not mailbox.outbox.empty():
# 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
class Schemes:
- def __init__(self):
- self.service = None
+ def handle_intent(self, intent):
+ print(f"Handling {intent.name}.")
- async def handle_intent(self, intent):
- print(f"Handling intent {intent.name}.")
-
- # Set the service
+ # 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
+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(),
+ s.do_agent_intents(),
s.deliver_mail()
)
+ # 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)
\ No newline at end of file
+ await websockets.serve(main, HOST, PORT)
\ No newline at end of file
import time
import asyncio
+import websockets
import mail
import agent
"""
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.
- intent = await self.recv_agent_intent()
- print(f"Intent: {intent.name}")
+ try:
+ intent = await asyncio.wait_for(
+ 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)
-
- await asyncio.sleep(0.1)
+ self.handle_intent(intent)
# 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()
- await self.send_agent_intent(intent)
-
- await asyncio.sleep(0.1)
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- # 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)
+ # Try recv.
+ try:
+ intent = await asyncio.wait_for(
+ self.mailbox_recv(), timeout=0.2)
+ except asyncio.TimeoutError:
+ continue
- # print("Shutting down in 10s")
+ # Send message to Agent.
+ print("Sending message")
+ await self.send_agent_intent(intent)
- # await asyncio.sleep(10)
+ # Shutdown protocol.
+ async def shutdown(self):
+ print("Shutting down session.")
+ self.on = False
\ No newline at end of file
projects / abc-messenger / commitdiff