Add support for Zenoh transport

.github/workflows/build.yml

@@ -31,9 +31,10 @@ jobs:
     if: "!contains(github.event.pull_request.labels.*.name, 'docs-only')"
     runs-on: 'ubuntu-latest'
     steps:
-      - name: Set up docker container
+      - name: Set up docker containers
         run: |
           docker run -d --name nanomq -p 1883:1883 -p 8083:8083 -p 8883:8883 emqx/nanomq:latest
+          docker run -d --name zenoh --init -p 7447:7447/tcp -p 8000:8000/tcp eclipse/zenoh
           docker ps -a
       - uses: compas-dev/compas-actions.build@v4
         with:
@@ -49,6 +50,7 @@ jobs:
       - name: Tear down docker container
         run: |
           docker rm -f nanomq
+          docker rm -f zenoh
 
   build-cpython-components:
     runs-on: windows-latest

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+* Add support for Zenoh transport protocol.
+
 ### Changed
 
 ### Removed

README.md

@@ -44,6 +44,7 @@ Or using `conda`:
 * Publisher/subscriber communication model (N-to-N communication)
 * In-process events
 * MQTT support
+* Zenoh support
 * Extensible codec system for message serialization (JSON, Protocol Buffers)
 
 ## Examples
@@ -87,6 +88,25 @@ for i in range(10):
 This example shows how to send and receive from a single script, but
 running publishers and subscribers on different scripts, different processes, or even different computers will work the exact same way.
 
+### Zenoh
+
+Apache Zenoh is a pub/sub/query protocol. In many ways, it is similar to MQTT but with some additional features and optimizations. COMPAS EVE also supports Zenoh as a transport protocol with an identical API to MQTT:
+
+```python
+import compas_eve as eve
+from compas_eve.zenoh import ZenohTransport
+
+tx = ZenohTransport()
+eve.set_default_transport(tx)
+
+pub = eve.Publisher("/hello_world")
+sub = eve.EchoSubscriber("/hello_world")
+sub.subscribe()
+
+for i in range(10):
+    pub.publish(dict(text=f"Hello World {i}"))
+```
+
 ### Using different codecs
 
 By default, COMPAS EVE uses JSON for message serialization. However, you can use different codecs for more efficient serialization:

docs/api.rst

@@ -9,5 +9,6 @@ API Reference
     api/compas_eve.codecs
     api/compas_eve.memory
     api/compas_eve.mqtt
+    api/compas_eve.zenoh
     api/compas_eve.ghpython
 

docs/api/compas_eve.zenoh.rst

@@ -0,0 +1,2 @@
+
+.. automodule:: compas_eve.zenoh

docs/examples/05_zenoh_distributed_world_pub.py

@@ -0,0 +1,16 @@
+import time
+
+from compas_eve import Publisher
+from compas_eve import Topic
+from compas_eve.zenoh import ZenohTransport
+
+topic = Topic("/hello/zenoh")
+tx = ZenohTransport()
+
+publisher = Publisher(topic, transport=tx)
+
+for i in range(20):
+    msg = dict(text=f"Hello world #{i} over Zenoh")
+    print(f"Publishing message: {msg}")
+    publisher.publish(msg)
+    time.sleep(1)

docs/examples/05_zenoh_distributed_world_sub.py

@@ -0,0 +1,18 @@
+import time
+
+from compas_eve import Subscriber
+from compas_eve import Topic
+from compas_eve.zenoh import ZenohTransport
+
+topic = Topic("/hello/zenoh")
+tx = ZenohTransport()
+
+subcriber = Subscriber(topic, callback=lambda msg: print(f"Received message: {msg}"), transport=tx)
+subcriber.subscribe()
+
+print("Waiting for messages, press CTRL+C to cancel")
+try:
+    while True:
+        time.sleep(1)
+finally:
+    subcriber.unsubscribe()

pyproject.toml

@@ -36,7 +36,7 @@ classifiers = [
 [tool.setuptools.dynamic]
 version = { attr = "compas_eve.__version__" }
 dependencies = { file = "requirements.txt" }
-optional-dependencies = { dev = { file = "requirements-dev.txt" } }
+optional-dependencies = { dev = { file = "requirements-dev.txt" }, zenoh = { file = "requirements-zenoh.txt" } }
 
 [project.entry-points.'compas_pb.plugins']
 serializers = 'compas_eve.codecs.conversions'

requirements-zenoh.txt

@@ -0,0 +1 @@
+eclipse-zenoh

src/compas_eve/ghpython/components/Ce_ZenohConnect/code.py

@@ -0,0 +1,37 @@
+# r: compas_eve>=2.1.1
+"""
+Initialize a Zenoh transport.
+"""
+
+from threading import Event
+
+import Grasshopper
+from compas_ghpython import create_id
+from scriptcontext import sticky as st
+
+from compas_eve.zenoh import ZenohTransport
+
+
+class ZenohConnectComponent(Grasshopper.Kernel.GH_ScriptInstance):
+    def RunScript(self, connect: bool):
+        zenoh_transport = None
+
+        key = create_id(ghenv.Component, "zenoh_transport")  # noqa: F821
+        zenoh_transport = st.get(key, None)
+
+        if zenoh_transport:
+            st[key].close()
+
+        if connect:
+            event = Event()
+            transport = ZenohTransport()
+            transport.on_ready(event.set)
+
+            if not event.wait(5):
+                raise Exception("Failed to initialize Zenoh router transport.")
+
+            st[key] = transport
+
+        zenoh_transport = st.get(key, None)
+        is_connected = zenoh_transport._is_connected if zenoh_transport else False
+        return (zenoh_transport, is_connected)

src/compas_eve/ghpython/components/Ce_ZenohConnect/metadata.json

@@ -0,0 +1,30 @@
+{
+    "name": "Zenoh Transport",
+    "nickname": "Zenoh",
+    "category": "COMPAS EVE",
+    "subcategory": "Events",
+    "description": "Initialize a Zenoh transport.",
+    "exposure": 2,
+
+    "ghpython": {
+        "isAdvancedMode": true,
+        "iconDisplay": 2,
+        "inputParameters": [
+            {
+                "name": "connect",
+                "description": "If True, initializes a Zenoh transport. If False, removes it. Defaults to False.",
+                "typeHintID": "bool"
+            }
+        ],
+        "outputParameters": [
+            {
+                "name": "zenoh_transport",
+                "description": "The Zenoh transport instance."
+            },
+            {
+                "name": "is_connected",
+                "description": "True if initialization has been processed."
+            }
+        ]
+    }
+}

src/compas_eve/zenoh/__init__.py

@@ -0,0 +1,22 @@
+"""
+********************************************************************************
+compas_eve.zenoh
+********************************************************************************
+
+.. currentmodule:: compas_eve.zenoh
+
+
+Classes
+=======
+
+.. autosummary::
+    :toctree: generated/
+    :nosignatures:
+
+    ZenohTransport
+
+"""
+
+from .zenoh_transport import ZenohTransport
+
+__all__ = ["ZenohTransport"]