BatchPullStreamSubscriber(
config, specification, calls, *, stream, pull_sub
)
Bases: TasksMixin, DefaultSubscriber[list['Msg']]
Batch-message consumer class.
Source code in faststream/nats/subscriber/usecases/stream_pull_subscriber.py
| def __init__(
self,
config: "NatsSubscriberConfig",
specification: "SubscriberSpecification[Any, Any]",
calls: "CallsCollection[list[Msg]]",
*,
stream: "JStream",
pull_sub: "PullSub",
) -> None:
parser = BatchParser(pattern=config.subject)
config.decoder = parser.decode_batch
config.parser = parser.parse_batch
super().__init__(config, specification, calls)
self.stream = stream
self.pull_sub = pull_sub
|
instance-attribute
instance-attribute
instance-attribute
extra_watcher_options = {}
instance-attribute
instance-attribute
specification = specification
instance-attribute
instance-attribute
instance-attribute
extra_options = extra_options or {}
property
Compile test.{name} to test.* subject.
async
Source code in faststream/nats/subscriber/usecases/stream_pull_subscriber.py
| @override
async def get_one(
self,
*,
timeout: float = 5,
) -> Optional["NatsMessage"]:
assert not self.calls, (
"You can't use `get_one` method if subscriber has registered handlers."
)
if not self._fetch_sub:
fetch_sub = self._fetch_sub = await self.jetstream.pull_subscribe(
subject=self.clear_subject,
config=self.config,
**self.extra_options,
)
else:
fetch_sub = self._fetch_sub
try:
raw_message = await fetch_sub.fetch(
batch=1,
timeout=timeout,
)
except TimeoutError:
return None
context = self._outer_config.fd_config.context
async_parser, async_decoder = self._get_parser_and_decoder()
return cast(
"NatsMessage",
await process_msg(
msg=raw_message,
middlewares=(
m(raw_message, context=context) for m in self._broker_middlewares
),
parser=async_parser,
decoder=async_decoder,
),
)
|
async
Create NATS subscription and start consume tasks.
Source code in faststream/nats/subscriber/usecases/basic.py
| async def start(self) -> None:
"""Create NATS subscription and start consume tasks."""
await super().start()
if self.calls:
await self._create_subscription()
self._post_start()
|
async
Clean up handler subscription, cancel consume task in graceful mode.
Source code in faststream/_internal/endpoint/subscriber/mixins.py
| async def stop(self) -> None:
"""Clean up handler subscription, cancel consume task in graceful mode."""
await super().stop()
for task in self.tasks:
if not task.done():
task.cancel()
self.tasks.clear()
|
add_call(*, parser_, decoder_, middlewares_, dependencies_)
Source code in faststream/_internal/endpoint/subscriber/usecase.py
| def add_call(
self,
*,
parser_: Optional["CustomCallable"],
decoder_: Optional["CustomCallable"],
middlewares_: Sequence["SubscriberMiddleware[Any]"],
dependencies_: Iterable["Dependant"],
) -> Self:
self._call_options = _CallOptions(
parser=parser_,
decoder=decoder_,
middlewares=middlewares_,
dependencies=dependencies_,
)
return self
|
async
Consume a message asynchronously.
Source code in faststream/_internal/endpoint/subscriber/usecase.py
| async def consume(self, msg: MsgType) -> Any:
"""Consume a message asynchronously."""
if not self.running:
return None
try:
return await self.process_message(msg)
except StopConsume:
# Stop handler at StopConsume exception
await self.stop()
except SystemExit:
# Stop handler at `exit()` call
await self.stop()
if app := self._outer_config.fd_config.context.get("app"):
app.exit()
except Exception: # nosec B110
# All other exceptions were logged by CriticalLogMiddleware
pass
|
async
Execute all message processing stages.
Source code in faststream/_internal/endpoint/subscriber/usecase.py
| async def process_message(self, msg: MsgType) -> "Response":
"""Execute all message processing stages."""
context = self._outer_config.fd_config.context
logger_state = self._outer_config.logger
async with AsyncExitStack() as stack:
stack.enter_context(self.lock)
# Enter context before middlewares
stack.enter_context(context.scope("handler_", self))
stack.enter_context(context.scope("logger", logger_state.logger.logger))
for k, v in self._outer_config.extra_context.items():
stack.enter_context(context.scope(k, v))
# enter all middlewares
middlewares: list[BaseMiddleware] = []
for base_m in self.__build__middlewares_stack():
middleware = base_m(msg, context=context)
middlewares.append(middleware)
await middleware.__aenter__()
cache: dict[Any, Any] = {}
parsing_error: Exception | None = None
for h in self.calls:
try:
message = await h.is_suitable(msg, cache)
except Exception as e:
parsing_error = e
break
if message is not None:
stack.enter_context(
context.scope("log_context", self.get_log_context(message)),
)
stack.enter_context(context.scope("message", message))
# Middlewares should be exited before scope release
for m in middlewares:
stack.push_async_exit(m.__aexit__)
result_msg = ensure_response(
await h.call(
message=message,
# consumer middlewares
_extra_middlewares=(
m.consume_scope for m in middlewares[::-1]
),
),
)
if not result_msg.correlation_id:
result_msg.correlation_id = message.correlation_id
for p in chain(
self.__get_response_publisher(message),
h.handler._publishers,
):
await p._publish(
result_msg.as_publish_command(),
_extra_middlewares=(
m.publish_scope for m in middlewares[::-1]
),
)
# Return data for tests
return result_msg
# Suitable handler was not found or
# parsing/decoding exception occurred
for m in middlewares:
stack.push_async_exit(m.__aexit__)
# Reraise it to catch in tests
if parsing_error:
raise parsing_error
error_msg = f"There is no suitable handler for {msg=}"
raise SubscriberNotFound(error_msg)
# An error was raised and processed by some middleware
return ensure_response(None)
|
Log context factory using in self.consume scope.
Source code in faststream/nats/subscriber/usecases/basic.py
| def get_log_context(
self,
message: Optional["StreamMessage[MsgType]"],
) -> dict[str, str]:
"""Log context factory using in `self.consume` scope."""
return self.build_log_context(
message=message,
subject=self.subject,
)
|
Source code in faststream/_internal/endpoint/subscriber/usecase.py
| def schema(self) -> dict[str, "SubscriberSpec"]:
self._build_fastdepends_model()
return self.specification.get_schema()
|
staticmethod
build_log_context(message, subject, *, queue='', stream='')
Static method to build log context out of self.consume scope.
Source code in faststream/nats/subscriber/usecases/basic.py
| @staticmethod
def build_log_context(
message: Optional["StreamMessage[MsgType]"],
subject: str,
*,
queue: str = "",
stream: str = "",
) -> dict[str, str]:
"""Static method to build log context out of `self.consume` scope."""
return {
"subject": subject,
"queue": queue,
"stream": stream,
"message_id": getattr(message, "message_id", ""),
}
|
add_task(func, func_args=None, func_kwargs=None)
Source code in faststream/_internal/endpoint/subscriber/mixins.py
| def add_task(
self,
func: Callable[..., Coroutine[Any, Any, Any]],
func_args: tuple[Any, ...] | None = None,
func_kwargs: dict[str, Any] | None = None,
) -> asyncio.Task[Any]:
args = func_args or ()
kwargs = func_kwargs or {}
task = asyncio.create_task(func(*args, **kwargs))
callback = TaskCallbackSupervisor(func, func_args, func_kwargs, self)
task.add_done_callback(callback)
self.tasks.append(task)
return task
|