Coverage for zenmqtt/connection/base.py: 100%

1import asyncio 

2from ssl import SSLContext 

3from typing import Optional, Protocol, TypedDict 

4 

5 

6class ConnectionOptions(TypedDict, total=False): 

7 # context for TLS connections 

8 ssl: SSLContext 

9 

10 

11class MQTTConnectionTransport(Protocol): 

12 async def write(self, payload: bytes) -> None: 

13 """Sends the bytes; may wait while the send buffer is full.""" 

14 

15 async def read(self, size: int = -1) -> bytes: 

16 """Up to size bytes, b"" when the connection is closed.""" 

17 

18 def is_closing(self) -> bool: 

19 """True when the connection is closed or is being closed.""" 

20 

21 async def close(self) -> None: 

22 """Closes the connection; then read() returns b"".""" 

23 

24 

25class MQTTConnection: 

26 def __init__( 

27 self, 

28 transport: MQTTConnectionTransport, 

29 ) -> None: 

30 self._transport = transport 

31 

32 # loop time of the last write, used by keep alive 

33 self.last_write_at: Optional[float] = None 

34 

35 async def write(self, payload: bytes) -> None: 

36 await self._transport.write(payload) 

37 

38 self.last_write_at = asyncio.get_running_loop().time() 

39 

40 async def disconnect(self) -> None: 

41 await self._transport.close() 

42 

43 async def read(self, size: int) -> bytes: 

44 return await self._transport.read(size) 

45 

46 def is_closing(self) -> bool: 

47 return self._transport.is_closing()