Skip to content

Commit dfec397

Browse files
authored
fix: correct Nats ObjectStorage get file behavior inside watch subscriber (#1523)
* tests: fix flacking nats-real test * fix: correct NATS ObjectStorage watch processing * tests: fix NATS OS tes
1 parent 2355603 commit dfec397

File tree

3 files changed

+14
-29
lines changed

3 files changed

+14
-29
lines changed

faststream/nats/subscriber/usecase.py

+9-10
Original file line numberDiff line numberDiff line change
@@ -1014,20 +1014,19 @@ async def _create_subscription( # type: ignore[override]
10141014
declare=self.obj_watch.declare,
10151015
)
10161016

1017-
self.subscription = UnsubscribeAdapter["ObjectStore.ObjectWatcher"](
1018-
await self.bucket.watch(
1019-
ignore_deletes=self.obj_watch.ignore_deletes,
1020-
include_history=self.obj_watch.include_history,
1021-
meta_only=self.obj_watch.meta_only,
1022-
)
1023-
)
1024-
10251017
self.add_task(self._consume_watch())
10261018

10271019
async def _consume_watch(self) -> None:
1028-
assert self.subscription, "You should call `create_subscription` at first." # nosec B101
1020+
assert self.bucket, "You should call `create_subscription` at first." # nosec B101
1021+
1022+
# Should be created inside task to avoid nats-py lock
1023+
obj_watch = await self.bucket.watch(
1024+
ignore_deletes=self.obj_watch.ignore_deletes,
1025+
include_history=self.obj_watch.include_history,
1026+
meta_only=self.obj_watch.meta_only,
1027+
)
10291028

1030-
obj_watch = self.subscription.obj
1029+
self.subscription = UnsubscribeAdapter["ObjectStore.ObjectWatcher"](obj_watch)
10311030

10321031
while self.running:
10331032
with suppress(TimeoutError):

tests/docs/nats/js/test_object.py

+4-18
Original file line numberDiff line numberDiff line change
@@ -9,23 +9,9 @@
99
async def test_basic():
1010
from docs.docs_src.nats.js.object import app, broker, handler
1111

12-
async with TestNatsBroker(broker, with_real=True):
13-
await broker.start()
14-
15-
os = await broker.object_storage("example-bucket")
16-
try:
17-
existed_files = await os.list()
18-
except Exception:
19-
existed_files = ()
20-
21-
call = True
22-
for file in existed_files:
23-
if file.name == "file.txt":
24-
call = False
25-
26-
if call:
27-
async with TestApp(app):
28-
pass
29-
12+
async with (
13+
TestNatsBroker(broker, with_real=True, connect_only=True),
14+
TestApp(app),
15+
):
3016
await handler.wait_call(3.0)
3117
handler.mock.assert_called_once_with("file.txt")

tests/opentelemetry/nats/test_nats.py

+1-1
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ async def test_batch(
4848
@broker.subscriber(
4949
queue,
5050
stream=stream,
51-
pull_sub=PullSub(3, batch=True),
51+
pull_sub=PullSub(3, batch=True, timeout=30.0),
5252
**self.subscriber_kwargs,
5353
)
5454
async def handler(m):

0 commit comments

Comments
 (0)