[CAP] Abstraction of actor_connector to go along with runtime factory and runtime abstraction

samples/apps/cap/py/autogencap/actor_connector.py

@@ -0,0 +1,43 @@
+# Agent_Sender takes a zmq context, Topic and creates a
+# socket that can publish to that topic. It exposes this functionality
+# using send_msg method
+from abc import ABC, abstractmethod
+
+
+class IActorSender(ABC):
+    @abstractmethod
+    def send_txt_msg(self, msg):
+        pass
+
+    @abstractmethod
+    def send_bin_msg(self, msg_type: str, msg):
+        pass
+
+    @abstractmethod
+    def send_recv_msg(self, msg_type: str, msg, resp_topic: str):
+        pass
+
+    @abstractmethod
+    def close(self):
+        pass
+
+
+class IActorConnector(ABC):
+    @abstractmethod
+    def send_txt_msg(self, msg):
+        pass
+
+    def send_bin_msg(self, msg_type: str, msg):
+        pass
+
+    def send_proto_msg(self, msg):
+        pass
+
+    def send_recv_proto_msg(self, msg, num_attempts=5):
+        pass
+
+    def send_recv_msg(self, msg_type: str, msg, num_attempts=5):
+        pass
+
+    def close(self):
+        pass

samples/apps/cap/py/autogencap/actor_runtime.py

@@ -2,7 +2,7 @@
 from typing import List
 
 from .Actor import Actor
-from .ActorConnector import ActorConnector
+from .actor_connector import IActorConnector
 from .proto.CAP_pb2 import ActorInfo
 
 
@@ -20,15 +20,15 @@ def disconnect(self):
         pass
 
     @abstractmethod
-    def find_by_topic(self, topic: str) -> ActorConnector:
+    def find_by_topic(self, topic: str) -> IActorConnector:
         pass
 
     @abstractmethod
-    def find_by_name(self, name: str) -> ActorConnector:
+    def find_by_name(self, name: str) -> IActorConnector:
         pass
 
     @abstractmethod
-    def find_termination(self) -> ActorConnector:
+    def find_termination(self) -> IActorConnector:
         pass
 
     @abstractmethod

samples/apps/cap/py/autogencap/ag_adapter/AG2CAP.py

@@ -4,7 +4,7 @@
 from autogen import Agent, ConversableAgent
 
 from ..actor_runtime import IRuntime
-from .AutoGenConnector import AutoGenConnector
+from .autogen_connector import AutoGenConnector
 
 
 class AG2CAP(ConversableAgent):

samples/apps/cap/py/autogencap/ag_adapter/AutoGenConnector.py → samples/apps/cap/py/autogencap/ag_adapter/autogen_connector.py

@@ -3,7 +3,7 @@
 
 from autogen import Agent
 
-from ..ActorConnector import ActorConnector
+from ..actor_runtime import IActorConnector
 from ..proto.Autogen_pb2 import GenReplyReq, GenReplyResp, PrepChat, ReceiveReq, Terminate
 
 
@@ -13,8 +13,8 @@ class AutoGenConnector:
     to/from the CAP system.
     """
 
-    def __init__(self, cap_sender: ActorConnector):
-        self._can_channel: ActorConnector = cap_sender
+    def __init__(self, cap_sender: IActorConnector):
+        self._can_channel: IActorConnector = cap_sender
 
     def close(self):
         """

samples/apps/cap/py/autogencap/ag_adapter/CAPGroupChatManager.py → samples/apps/cap/py/autogencap/ag_adapter/cap_group_chat_manager.py

@@ -1,9 +1,9 @@
 import time
 
 from autogen import GroupChatManager
-from autogencap.ActorConnector import ActorConnector
+from autogencap.actor_runtime import IActorConnector
 from autogencap.ag_adapter.CAP2AG import CAP2AG
-from autogencap.ag_adapter.CAPGroupChat import CAPGroupChat
+from autogencap.ag_adapter.cap_group_chat import CAPGroupChat
 
 from ..actor_runtime import IRuntime
 
@@ -25,7 +25,7 @@ def __init__(self, groupchat: CAPGroupChat, llm_config: dict, network: IRuntime)
 
     def initiate_chat(self, txt_msg: str) -> None:
         self._ensemble.connect()
-        user_proxy_conn: ActorConnector = self._ensemble.find_by_name(self._cap_group_chat.chat_initiator)
+        user_proxy_conn: IActorConnector = self._ensemble.find_by_name(self._cap_group_chat.chat_initiator)
         user_proxy_conn.send_txt_msg(txt_msg)
         self._wait_for_user_exit()
 

samples/apps/cap/py/autogencap/DirectorySvc.py → samples/apps/cap/py/autogencap/zmq_directory_svc.py

@@ -5,7 +5,6 @@
 import zmq
 
 from autogencap.Actor import Actor
-from autogencap.ActorConnector import ActorConnector, ActorSender
 from autogencap.Broker import Broker
 from autogencap.Config import router_url, xpub_url, xsub_url
 from autogencap.constants import Directory_Svc_Topic
@@ -24,12 +23,13 @@
     Error as ErrorMsg,
 )
 from autogencap.utility import report_error_msg
+from autogencap.zmq_runtime import ZMQActorConnector, ZMQActorSender
 
 # TODO (Future DirectorySv PR) use actor description, network_id, other properties to make directory
 # service more generic and powerful
 
 
-class DirectoryActor(Actor):
+class ZMQDirectoryActor(Actor):
     def __init__(self, topic: str, name: str):
         super().__init__(topic, name)
         self._registered_actors = {}
@@ -50,7 +50,7 @@ def _on_ping_msg(self, topic: str, msg_type: str, msg: bytes, sender_topic: str)
         Info("DirectorySvc", f"Ping received: {sender_topic}")
         pong = Pong()
         serialized_msg = pong.SerializeToString()
-        sender_connection = ActorSender(self._context, sender_topic)
+        sender_connection = ZMQActorSender(self._context, sender_topic)
         sender_connection.send_bin_msg(Pong.__name__, serialized_msg)
 
     def _on_actor_registration_msg(self, topic: str, msg_type: str, msg: bytes, sender_topic: str):
@@ -67,7 +67,7 @@ def _on_actor_registration_msg(self, topic: str, msg_type: str, msg: bytes, send
         else:
             self._registered_actors[name] = actor_reg.actor_info
 
-        sender_connection = ActorSender(self._context, sender_topic)
+        sender_connection = ZMQActorSender(self._context, sender_topic)
         serialized_msg = err.SerializeToString()
         sender_connection.send_bin_msg(ErrorMsg.__name__, serialized_msg)
 
@@ -96,16 +96,16 @@ def _on_actor_lookup_msg(self, topic: str, msg_type: str, msg: bytes, sender_top
         else:
             Error("DirectorySvc", f"Actor not found: {actor_lookup.actor_info.name}")
 
-        sender_connection = ActorSender(self._context, sender_topic)
+        sender_connection = ZMQActorSender(self._context, sender_topic)
         serialized_msg = actor_lookup_resp.SerializeToString()
         sender_connection.send_bin_msg(ActorLookupResponse.__name__, serialized_msg)
 
 
-class DirectorySvc:
+class ZMQDirectorySvc:
     def __init__(self, context: zmq.Context = zmq.Context()):
         self._context: zmq.Context = context
-        self._directory_connector: ActorConnector = None
-        self._directory_actor: DirectoryActor = None
+        self._directory_connector: ZMQActorConnector = None
+        self._directory_actor: ZMQDirectoryActor = None
 
     def _no_other_directory(self) -> bool:
         Debug("DirectorySvc", "Pinging existing DirectorySvc")
@@ -118,9 +118,9 @@ def _no_other_directory(self) -> bool:
 
     def start(self):
         Debug("DirectorySvc", "Starting.")
-        self._directory_connector = ActorConnector(self._context, Directory_Svc_Topic)
+        self._directory_connector = ZMQActorConnector(self._context, Directory_Svc_Topic)
         if self._no_other_directory():
-            self._directory_actor = DirectoryActor(Directory_Svc_Topic, "Directory Service")
+            self._directory_actor = ZMQDirectoryActor(Directory_Svc_Topic, "Directory Service")
             self._directory_actor.on_start(self._context)
             Info("DirectorySvc", "Directory service started.")
         else:
@@ -176,7 +176,7 @@ def main():
     proxy: Broker = Broker(context)
     proxy.start()
     # Start the directory service
-    directory_svc = DirectorySvc(context)
+    directory_svc = ZMQDirectorySvc(context)
     directory_svc.start()
     # # How do you register an actor?
     # directory_svc.register_actor_by_name("my_actor")