WIP changes from 6/30/25 pair coding

Author: mrakitin

src/bluesky/callbacks/nohup.out

@@ -0,0 +1,3 @@
+Connecting...
+Receiving on port ipc:///tmp/input.sock:5555; publishing to port ipc:///tmp/output.sock:5555.
+Use Ctrl+C to exit.

src/bluesky/callbacks/zmq.py

@@ -28,11 +28,6 @@ class Bluesky0MQDecodeError(Exception):
     ...
 
 
-class Protocols(Enum):
-    TCP = "tcp"
-    IPC = "ipc"
-
-
 class Publisher:
     """
     A callback that publishes documents to a 0MQ proxy.
@@ -65,7 +60,7 @@ class Publisher:
     >>> RE.subscribe(publisher)
     """
 
-    def __init__(self, address, *, prefix=b"", RE=None, zmq=None, serializer=pickle.dumps, protocol=Protocols.TCP):
+    def __init__(self, address, *, prefix=b"", RE=None, zmq=None, serializer=pickle.dumps):
         if RE is not None:
             warnings.warn(  # noqa: B028
                 "The RE argument to Publisher is deprecated and "
@@ -81,18 +76,13 @@ def __init__(self, address, *, prefix=b"", RE=None, zmq=None, serializer=pickle.
         if zmq is None:
             import zmq
 
-        if isinstance(address, tuple):
-            url = f"{protocol.value}://{address[0]}:{address[1]}"
-        else:
-            url = f"{protocol.value}://address"
-
-        self.address = url
+        self.address = address
         self.RE = RE
 
         self._prefix = bytes(prefix)
         self._context = zmq.Context()
         self._socket = self._context.socket(zmq.PUB)
-        self._socket.connect(url)
+        self._socket.connect(address)
         if RE:
             self._subscription_token = RE.subscribe(self)
         self._serializer = serializer
@@ -157,40 +147,28 @@ class Proxy:
     >>> proxy.start()  # runs until interrupted
     """
 
-    def __init__(self, in_port=None, out_port=None, *, zmq=None, protocol=Protocols.TCP):
+    def __init__(self, in_address=None, out_address=None, *, zmq=None):
         if zmq is None:
             import zmq
         self.zmq = zmq
         self.closed = False
-        self.input_path = None
-        self.output_path = None
         try:
             context = zmq.Context(1)
             # Socket facing clients
             frontend = context.socket(zmq.SUB)
-            if protocol is Protocols.TCP:
-                if in_port is None:
-                    in_port = frontend.bind_to_random_port(f"{protocol.value}://*")
-                else:
-                    frontend.bind(f"{protocol.value}://*:{in_port}")
-            elif protocol is Protocols.IPC:
-                if in_port is None:
-                    in_port = f"/tmp/{uuid.uuid4()}"
-                frontend.bind(f"{protocol.value}://{in_port}")
+            if in_address is None:
+                in_port = frontend.bind_to_random_port(f"tcp://*")
+            else:
+                in_port = frontend.bind(in_address)
 
             frontend.setsockopt_string(zmq.SUBSCRIBE, "")
 
             # Socket facing services
             backend = context.socket(zmq.PUB)
-            if protocol is Protocols.TCP:
-                if out_port is None:
-                    out_port = backend.bind_to_random_port(f"{protocol.value}://*")
-                else:
-                    backend.bind(f"{protocol.value}://*:{out_port}")
-            elif protocol is Protocols.IPC:
-                if out_port is None:
-                    out_port = f"/tmp/{uuid.uuid4()}"
-                frontend.bind(f"{protocol.value}://{out_port}")
+            if out_address is None:
+                out_port = backend.bind_to_random_port(f"tcp://*")
+            else:
+                out_port = backend.bind(out_address)
 
         except BaseException:
             # Clean up whichever components we have defined so far.
@@ -271,7 +249,6 @@ def __init__(
         zmq_asyncio=None,
         deserializer=pickle.loads,
         strict=False,
-        protocol=Protocols.TCP,
     ):
         if isinstance(prefix, str):
             raise ValueError("prefix must be bytes, not string")
@@ -282,12 +259,8 @@ def __init__(
             import zmq
         if zmq_asyncio is None:
             import zmq.asyncio as zmq_asyncio
-        if isinstance(address, tuple):
-            url = f"{protocol.value}://{address[0]}:{address[1]}"
-        else:
-            url = f"{protocol.value}://address"
         self._deserializer = deserializer
-        self.address = url
+        self.address = address
 
         if loop is None:
             loop = asyncio.new_event_loop()

src/bluesky/commandline/zmq_proxy.py

@@ -7,15 +7,15 @@
 logger = logging.getLogger("bluesky")
 
 
-def start_dispatcher(host, port, logfile=None):
+def start_dispatcher(out_address, logfile=None):
     """The dispatcher function
     Parameters
     ----------
     logfile : string
         string come from user command. ex --logfile=temp.log
         logfile will be "temp.log". logfile could be empty.
     """
-    dispatcher = RemoteDispatcher((host, port))
+    dispatcher = RemoteDispatcher(out_address)
     if logfile is not None:
         raise ValueError(
             "Parameter 'logfile' is deprecated and will be removed in future releases. "
@@ -41,8 +41,8 @@ def log_writer(name, doc):
 def main():
     DESC = "Start a 0MQ proxy for publishing bluesky documents over a network."
     parser = argparse.ArgumentParser(description=DESC)
-    parser.add_argument("in_port", type=int, nargs=1, help="port that RunEngines should broadcast to")
-    parser.add_argument("out_port", type=int, nargs=1, help="port that subscribers should subscribe to")
+    parser.add_argument("--in-address", help="port that RunEngines should broadcast to")
+    parser.add_argument("--out-address", help="port that subscribers should subscribe to")
     parser.add_argument(
         "--verbose",
         "-v",
@@ -51,8 +51,6 @@ def main():
     )
     parser.add_argument("--logfile", type=str, help="Redirect logging output to a file on disk.")
     args = parser.parse_args()
-    in_port = args.in_port[0]
-    out_port = args.out_port[0]
 
     if args.verbose:
         from bluesky.log import config_bluesky_logging
@@ -64,11 +62,11 @@ def main():
         else:
             config_bluesky_logging(level=level)
         # Set daemon to kill all threads upon IPython exit
-        threading.Thread(target=start_dispatcher, args=("localhost", out_port), daemon=True).start()
+        threading.Thread(target=start_dispatcher, args=(args.out_address), daemon=True).start()
 
     print("Connecting...")
-    proxy = Proxy(in_port, out_port)
-    print("Receiving on port %d; publishing to port %d." % (in_port, out_port))
+    proxy = Proxy(args.in_address, args.out_address)
+    print("Receiving on address %s; publishing to address %s." % (args.in_address, args.out_address))
     print("Use Ctrl+C to exit.")
     try:
         proxy.start()