Skip to content

Liveliness API Reference

The liveliness wrapper that powers the @on_liveliness decorator for node discovery and health monitoring.

bound_liveliness_wrapper

Bound-method proxy that injects self (the instance) into calls.

Source code in src/istos/primitives/liveliness.py 
 9
10
11
12
13
14
15
16
class bound_liveliness_wrapper:
    """Bound-method proxy that injects `self` (the instance) into calls."""
    def __init__(self, desc: "liveliness_wrapper", subj: Any):
        self.desc = desc
        self.subj = subj

    async def __call__(self, *args: Any, **kwargs: Any) -> Any:
        return await self.desc(self.subj, *args, **kwargs)

liveliness_wrapper

Descriptor that wraps a function to become a liveliness callback. It takes the payload from Zenoh, determines if a node joined or dropped, and passes the key expression and status to the function. The callback may declare Depends(...) dependencies, resolved per event.

Source code in src/istos/primitives/liveliness.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
class liveliness_wrapper:
    """
    Descriptor that wraps a function to become a liveliness callback.
    It takes the payload from Zenoh, determines if a node joined or dropped,
    and passes the key expression and status to the function. The callback may
    declare Depends(...) dependencies, resolved per event.
    """
    def __init__(self, func: Callable, prefix: str, dependency_overrides: Optional[dict] = None):
        self.func = func
        self.prefix = prefix
        self.calls = 0
        # key_expr and is_alive fill the first two positional slots.
        self._has_depends = has_dependencies(func)
        self._skip_names = tuple(positional_param_names(func)[:2])
        self._dependency_overrides = dependency_overrides if dependency_overrides is not None else {}

    async def __call__(self, *args: Any, **kwargs: Any) -> Any:
        if inspect.iscoroutinefunction(self.func):
            return await self.func(*args, **kwargs)
        return await asyncio.to_thread(self.func, *args, **kwargs)

    async def on_sample(self, sample: zenoh.Sample, instance: Optional[Any] = None) -> None:
        """Called by Zenoh when a liveliness event occurs."""
        self.calls += 1
        try:
            key_expr = str(sample.key_expr)
            is_alive = sample.kind == zenoh.SampleKind.PUT
            args = (instance, key_expr, is_alive) if instance is not None else (key_expr, is_alive)

            if self._has_depends:
                await invoke_with_dependencies(
                    self.func, args=args, skip_names=self._skip_names,
                    overrides=self._dependency_overrides,
                )
            elif inspect.iscoroutinefunction(self.func):
                await self.func(*args)
            else:
                await asyncio.to_thread(self.func, *args)
        except Exception as e:
            print(f"[Istos Liveliness] Error processing sample on {self.prefix}: {e}")

    def __get__(self, instance: Any, owner: Any) -> Any:
        if instance is None:
            return self
        return bound_liveliness_wrapper(self, instance)

on_sample(sample, instance=None) async

Called by Zenoh when a liveliness event occurs.

Source code in src/istos/primitives/liveliness.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
async def on_sample(self, sample: zenoh.Sample, instance: Optional[Any] = None) -> None:
    """Called by Zenoh when a liveliness event occurs."""
    self.calls += 1
    try:
        key_expr = str(sample.key_expr)
        is_alive = sample.kind == zenoh.SampleKind.PUT
        args = (instance, key_expr, is_alive) if instance is not None else (key_expr, is_alive)

        if self._has_depends:
            await invoke_with_dependencies(
                self.func, args=args, skip_names=self._skip_names,
                overrides=self._dependency_overrides,
            )
        elif inspect.iscoroutinefunction(self.func):
            await self.func(*args)
        else:
            await asyncio.to_thread(self.func, *args)
    except Exception as e:
        print(f"[Istos Liveliness] Error processing sample on {self.prefix}: {e}")