Coverage for faststream / nats / subscriber / usecases / stream_push_subscriber.py: 58%

1from typing import TYPE_CHECKING, Optional (empty)

2 

3from nats.aio.msg import Msg (empty)

4from typing_extensions import override (empty)

5 

6from faststream._internal.endpoint.subscriber.mixins import ConcurrentMixin (empty)

7 

8from .stream_basic import StreamSubscriber (empty)

9 

10if TYPE_CHECKING: (empty)

11 from nats.js import JetStreamContext 

12 

13 

14class PushStreamSubscriber(StreamSubscriber): (empty)

15 subscription: Optional["JetStreamContext.PushSubscription"] (empty)

16 

17 @override (empty)

18 async def _create_subscription(self) -> None: (empty)

19 """Create NATS subscription and start consume task.""" 

20 if self.subscription: 20 ↛ 21line 20 didn't jump to line 21 because the condition on line 20 was never true(empty)

21 return 

22 

23 self.subscription = await self.jetstream.subscribe( (empty)

24 subject=self.clear_subject, 

25 queue=self.queue, 

26 cb=self.consume, 

27 config=self.config, 

28 **self.extra_options, 

29 ) 

30 

31 

32class ConcurrentPushStreamSubscriber(ConcurrentMixin[Msg], StreamSubscriber): (empty)

33 subscription: Optional["JetStreamContext.PushSubscription"] (empty)

34 

35 @override (empty)

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

37 """Create NATS subscription and start consume task.""" 

38 if self.subscription: 

39 return 

40 

41 self.start_consume_task() 

42 

43 self.subscription = await self.jetstream.subscribe( 

44 subject=self.clear_subject, 

45 queue=self.queue, 

46 cb=self._put_msg, 

47 config=self.config, 

48 **self.extra_options, 

49 )