Merge pull request #215 from ezmsg-org/feature/sync-lowlevel-api

Feature: Synchronous Low‑Level API for ezmsg (ROS2‑style ergonomics)

Author: griffinmilsap

examples/lowlevel_api.py

@@ -0,0 +1,33 @@
+import asyncio
+
+import ezmsg.core as ez
+
+TOPIC = "/TEST"
+
+
+async def main(host: str = "127.0.0.1", port: int = 12345) -> None:
+    async with ez.GraphContext((host, port), auto_start=True) as ctx:
+        pub = await ctx.publisher(TOPIC)
+        try:
+            print("Publisher Task Launched")
+            count = 0
+            while True:
+                await pub.broadcast(f"{count=}")
+                await asyncio.sleep(0.1)
+                count += 1
+        except asyncio.CancelledError:
+            pass
+        finally:
+            print("Publisher Task Concluded")
+
+
+if __name__ == "__main__":
+    from argparse import ArgumentParser
+
+    parser = ArgumentParser()
+    parser.add_argument("--host", default="127.0.0.1", help="hostname for graphserver")
+    parser.add_argument("--port", default=12345, type=int, help="port for graphserver")
+
+    args = parser.parse_args()
+
+    asyncio.run(main(host=args.host, port=args.port))

examples/simple_async_publisher.py

@@ -0,0 +1,34 @@
+import asyncio
+
+import ezmsg.core as ez
+
+TOPIC = "/TEST"
+
+
+async def main(host: str = "127.0.0.1", port: int = 12345) -> None:
+    async with ez.GraphContext((host, port), auto_start=True) as ctx:
+        sub = await ctx.subscriber(TOPIC)
+        try:
+            print("Subscriber Task Launched")
+            while True:
+                async with sub.recv_zero_copy() as msg:
+                    # Uncomment if you want to witness backpressure!
+                    # await asyncio.sleep(1.0)
+                    print(msg)
+        except asyncio.CancelledError:
+            pass
+        finally:
+            print("Subscriber Task Concluded")
+            print("Detached")
+
+
+if __name__ == "__main__":
+    from argparse import ArgumentParser
+
+    parser = ArgumentParser()
+    parser.add_argument("--host", default="127.0.0.1", help="hostname for graphserver")
+    parser.add_argument("--port", default=12345, type=int, help="port for graphserver")
+
+    args = parser.parse_args()
+
+    asyncio.run(main(host=args.host, port=args.port))

examples/simple_async_subscriber.py

@@ -0,0 +1,35 @@
+import time
+
+import ezmsg.core as ez
+
+TOPIC = "/TEST"
+
+def main(host: str = "127.0.0.1", port: int = 12345) -> None:
+    with ez.sync.init((host, port), auto_start=True) as ctx:
+        pub = ctx.create_publisher(TOPIC, force_tcp=True)
+
+        print("Publisher Task Launched")
+        count = 0
+        try:
+            while True:
+                output = f"{count=}"
+                pub.publish(output)
+                print(output)
+                time.sleep(0.1)
+                count += 1
+        except KeyboardInterrupt:
+            pass
+        print("Publisher Task Concluded")
+
+    print("Done")
+
+
+if __name__ == "__main__":
+    from argparse import ArgumentParser
+
+    parser = ArgumentParser()
+    parser.add_argument("--host", default="127.0.0.1", help="hostname for graphserver")
+    parser.add_argument("--port", default=12345, type=int, help="port for graphserver")
+    args = parser.parse_args()
+
+    main(host=args.host, port=args.port)

examples/simple_publisher.py

@@ -0,0 +1,30 @@
+import ezmsg.core as ez
+
+TOPIC = "/TEST"
+
+
+def main(host: str = "127.0.0.1", port: int = 12345) -> None:
+    with ez.sync.init((host, port), auto_start=True) as ctx:
+        print("Subscriber Task Launched")
+
+        def on_message(msg: str) -> None:
+            # Uncomment if you want to witness backpressure!
+            # import time
+            # time.sleep(1.0)
+            print(msg)
+
+        ctx.create_subscription(TOPIC, callback=on_message)
+        ez.sync.spin(ctx)
+
+    print("Subscriber Task Concluded")
+
+
+if __name__ == "__main__":
+    from argparse import ArgumentParser
+
+    parser = ArgumentParser()
+    parser.add_argument("--host", default="127.0.0.1", help="hostname for graphserver")
+    parser.add_argument("--port", default=12345, type=int, help="port for graphserver")
+    args = parser.parse_args()
+
+    main(host=args.host, port=args.port)

examples/simple_subscriber.py

@@ -25,6 +25,10 @@
     "NormalTermination",
     "GraphServer",
     "GraphContext",
+    "sync",
+    "SyncContext",
+    "SyncPublisher",
+    "SyncSubscriber",
     "run_command",
     "Publisher",
     "Subscriber",
@@ -45,6 +49,8 @@
 from .backendprocess import Complete, NormalTermination
 from .graphserver import GraphServer
 from .graphcontext import GraphContext
+from . import sync
+from .sync import SyncContext, SyncPublisher, SyncSubscriber
 from .command import run_command
 from .pubclient import Publisher
 from .subclient import Subscriber

src/ezmsg/core/__init__.py

@@ -222,6 +222,15 @@ def graph_address(self) -> AddressType | None:
             return self._graph_context.graph_address
         return self._graph_address
 
+    @property
+    def strict_shutdown(self) -> bool:
+        value = os.environ.get("EZMSG_STRICT_SHUTDOWN", "")
+        return value.lower() in ("1", "true", "yes", "on")
+
+    @strict_shutdown.setter
+    def strict_shutdown(self, value: bool) -> None:
+        os.environ["EZMSG_STRICT_SHUTDOWN"] = "1" if value else "0"
+
     @property
     def graph_server_spawned(self) -> bool:
         return self._graph_server_spawned