__aenter__
Open the Unix socket, register the requested event filters and return the subscription
itself for use inside an async with block.
Open the subscription and return it for an async with block.
Example
async with client.subscribe([{"event": "pane.output"}]) as stream:
async for event in stream.events():
print(event)
Source code in src/herdr_client/async_client/client.py
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138 | async def __aenter__(self) -> Self:
"""Open the subscription and return it for an ``async with`` block.
Example:
```python
async with client.subscribe([{"event": "pane.output"}]) as stream:
async for event in stream.events():
print(event)
```
"""
if self._writer is not None:
raise RuntimeError("subscription is already open")
self._reader, self._writer = await connect_socket(
self._socket_path, self._timeout
)
try:
await send_envelope(
self._writer,
self._timeout,
{
"id": new_id(),
"method": "events.subscribe",
"params": {"subscriptions": cast("JSONValue", self._subscriptions)},
},
)
response = await read_json_line(self._reader, self._timeout)
self._ack = subscription_ack(response)
return self
except BaseException:
await self.aclose()
raise
|