Skip to content

trading_agent.terminal.event_stream

Keeps the gateway's event stream open, reconnecting after a drop, as the web page's eventSocket.ts does.

Each event is a JSON object with kind, desk, payload and created_at. The stream never replays old events, so a client reads today's history from the journal routes.

Typical usage example:

stream = EventStream(gateway.events_address()) await stream.follow(on_event, on_state)

LIVE module-attribute 

LIVE = 'live'

CONNECTING module-attribute 

CONNECTING = 'connecting'

RECONNECTING module-attribute 

RECONNECTING = 'reconnecting'

REFUSED module-attribute 

REFUSED = 'refused'

AccessRefusedError

Bases: Exception

The gateway closed the event stream because the access token is wrong.

EventStream 

EventStream(address, connect=connect)

One client's view of the gateway's events.

Creates the stream without connecting.

Parameters:

Name Type Description Default
address str

The WebSocket address, with the token, from GatewayClient.events_address.

 required
connect Callable[[str], Any]

Opens a connection; replaced by a stand-in in tests.

 connect
Source code in src/trading_agent/terminal/event_stream.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    address: str,
    connect: Callable[[str], Any] = websockets.connect,
):
    """Creates the stream without connecting.

    Args:
        address (str): The WebSocket address, with the token, from GatewayClient.events_address.
        connect (Callable[[str], Any]): Opens a connection; replaced by a stand-in in tests.
    """
    self._address = address
    self._connect = connect

follow async 

follow(on_event, on_state)

Passes every event to on_event until cancelled, reconnecting after a drop.

The wait before reconnecting doubles from 1 second up to 15 seconds, and starts again at 1 second after a connection that worked.

Parameters:

Name Type Description Default
on_event Callable[[dict[str, Any]], None]

Called with each event.

 required
on_state Callable[[str], None]

Called with "connecting", "live", "reconnecting" or "refused".

 required

Raises:

Type Description
AccessRefusedError

The gateway refused the access token.
Source code in src/trading_agent/terminal/event_stream.py 
49
50
51
52
53
54
55
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
85
86
87
88
89
90
async def follow(
    self,
    on_event: Callable[[dict[str, Any]], None],
    on_state: Callable[[str], None],
) -> None:
    """Passes every event to on_event until cancelled, reconnecting after a drop.

    The wait before reconnecting doubles from 1 second up to 15 seconds, and starts again at 1 second after a connection that worked.

    Args:
        on_event (Callable[[dict[str, Any]], None]): Called with each event.
        on_state (Callable[[str], None]): Called with "connecting", "live", "reconnecting" or "refused".

    Raises:
        AccessRefusedError: The gateway refused the access token.
    """
    wait_seconds = _FIRST_WAIT_SECONDS
    on_state(CONNECTING)
    while True:
        try:
            async with self._connect(self._address) as socket:
                on_state(LIVE)
                wait_seconds = _FIRST_WAIT_SECONDS
                async for message in socket:
                    on_event(json.loads(message))
        except websockets.ConnectionClosed as closed:
            if self._was_refused(closed):
                on_state(REFUSED)
                raise AccessRefusedError(
                    'The gateway refused the access token'
                ) from closed
        except websockets.InvalidStatus as refused:
            if refused.response.status_code in (401, 403):
                on_state(REFUSED)
                raise AccessRefusedError(
                    'The gateway refused the access token'
                ) from refused
        except OSError:
            pass
        on_state(RECONNECTING)
        await asyncio.sleep(wait_seconds)
        wait_seconds = min(wait_seconds * 2, _LONGEST_WAIT_SECONDS)