Skip to content

Subscription

Synchronous context manager for a long-lived Herdr event stream. It is normally created with HerdrClient.subscribe.

Constructor

Prepare a synchronous subscription without opening the socket.

Parameters:

Name Type Description Default
socket_path Path

Unix socket path used for the subscription connection.

 required
timeout float

Timeout in seconds for connect, write and read operations.

 required
subscriptions Iterable[EventSubscription]

Event filters sent to events.subscribe when the context is entered.

 required
Example 
subscription = Subscription(
    Path("/tmp/herdr.sock"),
    timeout=5.0,
    subscriptions=[{"event": "pane.output"}],
)
Source code in src/herdr_client/sync/client.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
def __init__(
    self,
    socket_path: Path,
    timeout: float,
    subscriptions: Iterable[EventSubscription],
) -> None:
    """Prepare a synchronous subscription without opening the socket.

    Args:
        socket_path: Unix socket path used for the subscription connection.
        timeout: Timeout in seconds for connect, write and read operations.
        subscriptions: Event filters sent to ``events.subscribe`` when the
            context is entered.

    Example:
        ```python
        subscription = Subscription(
            Path("/tmp/herdr.sock"),
            timeout=5.0,
            subscriptions=[{"event": "pane.output"}],
        )
        ```
    """
    self._socket_path = socket_path
    self._timeout = timeout
    self._subscriptions = list(subscriptions)
    self._socket: socket.socket | None = None
    self._file: BinaryIO | None = None
    self._ack: SubscriptionAck | None = None
Member Purpose
ack Read the server acknowledgement
__enter__ Open the subscription context
__exit__ Close the context
close Close the connection explicitly
events Iterate over pushed event envelopes