Coverage for object_streams/transports/base.py: 100%

1"""Transport protocol used by concrete WebSocket or message adapters.""" 

2 

3from __future__ import annotations 

4 

5from typing import Any 

6from typing import Protocol 

7 

8from object_streams.events import StreamEvent 

9from object_streams.subscriptions import ResyncRequired 

10from object_streams.subscriptions import SubscriptionRequest 

11 

12 

13__all__ = ("Transport",) 

14 

15 

16class Transport(Protocol): 

17 async def prepare_subscription(self, subscription: SubscriptionRequest) -> None: 

18 """Prepare transport routing before the subscription is acknowledged.""" 

19 

20 async def send_subscribed(self, subscription: SubscriptionRequest) -> None: 

21 """Send a subscription acknowledgement.""" 

22 

23 async def send_unsubscribed(self, subscription_id: str) -> None: 

24 """Send an unsubscribe acknowledgement.""" 

25 

26 async def send_event(self, event: StreamEvent) -> None: 

27 """Send a stream event.""" 

28 

29 async def send_resync(self, resync: ResyncRequired) -> None: 

30 """Send a subscription-level resync instruction.""" 

31 

32 async def send_error(self, code: str, message: str, *, details: Any = None) -> None: 

33 """Send a protocol error."""