Coverage for faststream / confluent / publisher / state.py: 80%

1from typing import TYPE_CHECKING, Protocol (empty)

2 

3from faststream.exceptions import IncorrectState (empty)

4 

5if TYPE_CHECKING: (empty)

6 from faststream.confluent.helpers.client import AsyncConfluentProducer 

7 

8 

9class ProducerState(Protocol): (empty)

10 @property (empty)

11 def producer(self) -> "AsyncConfluentProducer": ... (empty)

12 

13 def __bool__(self) -> bool: ... (empty)

14 

15 async def ping(self, timeout: float) -> bool: ... (empty)

16 

17 async def stop(self) -> None: ... (empty)

18 

19 async def flush(self) -> None: ... (empty)

20 

21 

22class EmptyProducerState: (empty)

23 __slots__ = () (empty)

24 

25 @property (empty)

26 def producer(self) -> "AsyncConfluentProducer": (empty)

27 msg = "You can't use producer here, please connect broker first." 

28 raise IncorrectState(msg) 

29 

30 async def ping(self, timeout: float) -> bool: (empty)

31 return False 

32 

33 def __bool__(self) -> bool: (empty)

34 return False 

35 

36 async def stop(self) -> None: (empty)

37 pass (empty)

38 

39 async def flush(self) -> None: (empty)

40 pass 

41 

42 

43class RealProducer: (empty)

44 __slots__ = ("producer",) (empty)

45 

46 def __init__(self, producer: "AsyncConfluentProducer") -> None: (empty)

47 self.producer = producer (empty)

48 

49 def __bool__(self) -> bool: (empty)

50 return True (empty)

51 

52 async def stop(self) -> None: (empty)

53 await self.producer.stop() (empty)

54 

55 async def ping(self, timeout: float) -> bool: (empty)

56 return await self.producer.ping(timeout=timeout) (empty)

57 

58 async def flush(self) -> None: (empty)

59 await self.producer.flush()