__enter__
Open the Unix socket, register the requested event filters and return the subscription
itself for use inside a with block.
Open the subscription and return it for a with block.
Example
with client.subscribe([{"event": "pane.output"}]) as stream:
for event in stream.events():
print(event)
Source code in src/herdr_client/sync/client.py
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129 | def __enter__(self) -> Self:
"""Open the subscription and return it for a ``with`` block.
Example:
```python
with client.subscribe([{"event": "pane.output"}]) as stream:
for event in stream.events():
print(event)
```
"""
if self._socket is not None:
raise RuntimeError("subscription is already open")
self._socket = connect_socket(self._socket_path, self._timeout)
try:
send_envelope(
self._socket,
{
"id": new_id(),
"method": "events.subscribe",
"params": {"subscriptions": cast("JSONValue", self._subscriptions)},
},
)
self._file = self._socket.makefile("rb")
response = read_json_line(self._file)
self._ack = subscription_ack(response)
return self
except BaseException:
self.close()
raise
|