Skip to content

Commit a626beb

Browse files
authored
refactor: sleep with listen
2 parents 2be2c32 + 8320512 commit a626beb

File tree

2 files changed

+19
-18
lines changed

2 files changed

+19
-18
lines changed

taskiq_faststream/__about__.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,2 @@
11
"""FastStream - taskiq integration to schedule FastStream tasks."""
2-
__version__ = "0.1.1"
2+
__version__ = "0.1.2"

taskiq_faststream/broker.py

Lines changed: 18 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import typing
22
import warnings
33

4+
import anyio
45
from faststream._compat import TypeAlias, override
56
from faststream.app import FastStream
67
from faststream.broker.core.asyncronous import BrokerAsyncUsecase
@@ -23,8 +24,8 @@ class BrokerWrapper(AsyncBroker):
2324
2425
Methods:
2526
__init__ : Initializes the object.
26-
startup : Startup wrapper FastStream broker.
27-
shutdown : Shutdown wrapper FastStream broker.
27+
startup : Startup wrapped FastStream broker.
28+
shutdown : Shutdown wrapped FastStream broker.
2829
kick : Call wrapped FastStream broker `publish` method.
2930
task : Register FastStream scheduled task.
3031
"""
@@ -34,12 +35,12 @@ def __init__(self, broker: BrokerAsyncUsecase[typing.Any, typing.Any]) -> None:
3435
self.broker = broker
3536

3637
async def startup(self) -> None:
37-
"""Startup wrapper FastStream broker."""
38+
"""Startup wrapped FastStream broker."""
3839
await super().startup()
3940
await self.broker.start()
4041

4142
async def shutdown(self) -> None:
42-
"""Shutdown wrapper FastStream broker."""
43+
"""Shutdown wrapped FastStream broker."""
4344
await self.broker.close()
4445
await super().shutdown()
4546

@@ -51,17 +52,17 @@ async def listen(
5152
self,
5253
) -> typing.AsyncGenerator[typing.Union[bytes, AckableMessage], None]:
5354
"""Not supported method."""
54-
warnings.warn(
55-
message=(
56-
f"{self.__class__.__name__} doesn't support `listen` method. "
57-
"Please, use it only to register a task."
58-
),
59-
category=RuntimeWarning,
60-
stacklevel=1,
61-
)
62-
6355
while True:
56+
warnings.warn(
57+
message=(
58+
f"{self.__class__.__name__} doesn't support `listen` method. "
59+
"Please, use it only to register a task."
60+
),
61+
category=RuntimeWarning,
62+
stacklevel=1,
63+
)
6464
yield b""
65+
await anyio.sleep(60)
6566

6667
@override
6768
def task( # type: ignore[override]
@@ -98,8 +99,8 @@ class AppWrapper(BrokerWrapper):
9899
99100
Methods:
100101
__init__ : Initializes the object.
101-
startup : Startup wrapper FastStream.
102-
shutdown : Shutdown wrapper FastStream.
102+
startup : Startup wrapped FastStream.
103+
shutdown : Shutdown wrapped FastStream.
103104
kick : Call wrapped FastStream broker `publish` method.
104105
task : Register FastStream scheduled task.
105106
"""
@@ -109,12 +110,12 @@ def __init__(self, app: FastStream) -> None:
109110
self.app = app
110111

111112
async def startup(self) -> None:
112-
"""Startup wrapper FastStream broker."""
113+
"""Startup wrapped FastStream."""
113114
await super(BrokerWrapper, self).startup()
114115
await self.app._startup() # noqa: SLF001
115116

116117
async def shutdown(self) -> None:
117-
"""Shutdown wrapper FastStream broker."""
118+
"""Shutdown wrapped FastStream."""
118119
await self.app._shutdown() # noqa: SLF001
119120
await super(BrokerWrapper, self).shutdown()
120121

0 commit comments

Comments
 (0)