add first commit

.gitignore

@@ -0,0 +1,11 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+build_nuitka/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv

pyproject.toml

@@ -0,0 +1,17 @@
+[build-system]
+build-backend = "hatchling.build"
+requires = ["hatchling"]
+
+[project]
+name = "chongming-request"
+version = "0.1.0"
+description = "worker direct communication"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "eclipse-zenoh>=1.9.0",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/chongming_request"]
+

src/chongming_request/__init__.py

@@ -0,0 +1 @@
+from .client import Client

src/chongming_request/client.py

@@ -0,0 +1,110 @@
+import asyncio
+import logging
+import random
+import time
+from typing import Optional, Dict, Any, List
+
+import zenoh
+import orjson
+
+
+_SENTINEL = object()
+
+class Client:
+    def __init__(self, 
+                 session: Optional[zenoh.Session] = None,
+                 logger: Optional[logging.Logger] = None):
+        if session is None:
+            self._session = zenoh.open(zenoh.Config())
+            self._owns_session = True
+        else:
+            self._session = session
+            self._owns_session = False
+        self._logger = logger or logging.getLogger(__name__)
+        self._worker_cache = {}
+        self._cache_ttl = 15
+        self._cache_timestamps = {}
+
+    async def _get_workers(self, worker_type: str) -> List[str]:
+        now = time.monotonic()
+        if worker_type in self._worker_cache:
+            if now - self._cache_timestamps[worker_type] < self._cache_ttl:
+                return self._worker_cache[worker_type]
+        try:
+            replies = self._session.liveliness().get(f"worker/{worker_type}/*")
+            workers = [str(r.ok.key_expr) for r in replies if r.ok]
+            self._worker_cache[worker_type] = workers
+            self._cache_timestamps[worker_type] = now
+            return workers
+        except Exception as e:
+            if worker_type in self._worker_cache:
+                self._logger.warning("Using expired cache")
+                return self._worker_cache[worker_type]
+            self._logger.error(str(e))
+            raise ValueError(str(e))
+
+    async def call(
+        self,
+        worker_type: str,
+        subject: str,
+        payload: Dict[str, Any],
+        method: str = "GET",
+        timeout: float = 3.0,
+        retries: int = 1,
+    ) -> Dict[str, Any]:
+        payload.setdefault("_method", method.upper())
+        full_key = ""
+        for attempt in range(retries + 1):
+            workers = await self._get_workers(worker_type)
+            if not workers:
+                raise RuntimeError(...)
+            target = random.choice(workers)
+            full_key = f"{target}/{subject}"
+            try:
+                reply_iter = self._session.get(
+                    full_key,
+                    payload=orjson.dumps(payload),
+                    consolidation=zenoh.ConsolidationMode.NONE,
+                )
+                def _get_first(iterator):
+                    for item in iterator:
+                        return item
+                    return _SENTINEL
+                first_reply = await asyncio.wait_for(
+                    asyncio.to_thread(_get_first, reply_iter),
+                    timeout=timeout,
+                )
+                if first_reply is _SENTINEL:
+                    self._logger.warning(f"Worker {target} did not respond to '{full_key}', retrying {attempt+1}/{retries}...")
+                    continue
+
+                reply = first_reply
+                if reply.ok: # type: ignore
+                    payload_bytes = bytes(reply.ok.payload) if reply.ok.payload else b"" # type: ignore
+                    if not payload_bytes:
+                        result = {}
+                    else:
+                        result = orjson.loads(payload_bytes)
+                    return result
+                else:
+                    error_payload = reply.err.payload  # type: ignore
+                    error_msg = str(error_payload) if error_payload else "unknown error"
+                    self._logger.warning(f"Worker Error: {error_msg} retrying {attempt+1}/{retries}...")
+                    continue
+            except asyncio.TimeoutError:
+                raise RuntimeError(f"Request to '{full_key}' timed out after {timeout} seconds")
+        raise RuntimeError(f"Failed to get a valid response from '{full_key}' after {retries} retries")
+    
+    async def close(self):
+        if self._owns_session and self._session:
+            self._session.close()
+
+    async def __aenter__(self):
+        return self
+    
+    async def __aexit__(self, *args):
+        await self.close()
+
+    def call_sync(self, *args, **kwargs):
+        return asyncio.run(self.call(*args, **kwargs))
+