NatsBroker

def __init__(
    self,
    servers: str | Iterable[str] = ("nats://localhost:4222",),
    *,
    error_cb: Optional["ErrorCallback"] = None,
    disconnected_cb: Optional["Callback"] = None,
    closed_cb: Optional["Callback"] = None,
    discovered_server_cb: Optional["Callback"] = None,
    reconnected_cb: Optional["Callback"] = None,
    name: str | None = SERVICE_NAME,
    pedantic: bool = False,
    verbose: bool = False,
    allow_reconnect: bool = True,
    connect_timeout: int = DEFAULT_CONNECT_TIMEOUT,
    reconnect_time_wait: int = DEFAULT_RECONNECT_TIME_WAIT,
    max_reconnect_attempts: int = DEFAULT_MAX_RECONNECT_ATTEMPTS,
    ping_interval: int = DEFAULT_PING_INTERVAL,
    max_outstanding_pings: int = DEFAULT_MAX_OUTSTANDING_PINGS,
    dont_randomize: bool = False,
    flusher_queue_size: int = DEFAULT_MAX_FLUSHER_QUEUE_SIZE,
    no_echo: bool = False,
    tls_hostname: Annotated[
        str | None,
        deprecated(
            "Use `security=NatsSecurity(...)` instead. "
            "This argument will be removed in 1.0.0."
        ),
    ] = EMPTY,
    token: Annotated[
        str | None,
        deprecated(
            "Use `security=NatsToken(...)` instead. "
            "This argument will be removed in 1.0.0."
        ),
    ] = EMPTY,
    drain_timeout: int = DEFAULT_DRAIN_TIMEOUT,
    signature_cb: Annotated[
        Optional["SignatureCallback"],
        deprecated(
            "Use `security=NatsJWT(...)` instead. "
            "This argument will be removed in 1.0.0."
        ),
    ] = EMPTY,
    user_jwt_cb: Annotated[
        Optional["JWTCallback"],
        deprecated(
            "Use `security=NatsJWT(...)` instead. "
            "This argument will be removed in 1.0.0."
        ),
    ] = EMPTY,
    user_credentials: Annotated[
        Optional["Credentials"],
        deprecated(
            "Use `security=NatsCredentials(...)` instead. "
            "This argument will be removed in 1.0.0."
        ),
    ] = EMPTY,
    nkeys_seed: Annotated[
        str | None,
        deprecated(
            "Use `security=NatsNKey.from_file(...)` instead. "
            "This argument will be removed in 1.0.0."
        ),
    ] = EMPTY,
    nkeys_seed_str: Annotated[
        str | None,
        deprecated(
            "Use `security=NatsNKey.from_seed(...)` instead. "
            "This argument will be removed in 1.0.0."
        ),
    ] = EMPTY,
    inbox_prefix: str | bytes = DEFAULT_INBOX_PREFIX,
    pending_size: int = DEFAULT_PENDING_SIZE,
    flush_timeout: float | None = None,
    ws_connection_headers: dict[str, list[str]] | None = None,
    reconnect_to_server_handler: ReconnectToServerHandler | None = None,
    js_options: Union["JsInitOptions", dict[str, Any], None] = None,
    graceful_timeout: float | None = 15.0,
    ack_policy: AckPolicy = EMPTY,
    id_generator: IdGenerator = gen_cor_id,
    decoder: Optional["CustomCallable"] = None,
    codec: Optional["CodecProto"] = None,
    parser: Optional["CustomCallable"] = None,
    dependencies: Sequence["Dependant"] = (),
    middlewares: Sequence["BrokerMiddleware[Any, Any]"] = (),
    routers: Iterable[NatsRegistrator] = (),
    security: Optional["BaseSecurity"] = None,
    specification_url: str | Iterable[str] | None = None,
    protocol: str | None = "nats",
    protocol_version: str | None = "custom",
    description: str | None = None,
    tags: Iterable[Union["Tag", "TagDict"]] = (),
    logger: Optional["LoggerProto"] = EMPTY,
    log_level: int = logging.INFO,
    apply_types: bool = True,
    serializer: Optional["SerializerProto"] = EMPTY,
    provider: Optional["Provider"] = None,
    context: Optional["ContextRepo"] = None,
    underlying_driver_annotations: "UnderlyingDriverAnnotations | None" = None,
) -> None:
    """Initialize the NatsBroker object.

    Args:
        servers:
            NATS cluster addresses to connect.
        error_cb:
            Callback to report errors.
        disconnected_cb:
            Callback to report disconnection from NATS.
        closed_cb:
            Callback to report when client stops reconnection to NATS.
        discovered_server_cb:
            A callback to report when a new server joins the cluster.
        reconnected_cb:
            Callback to report success reconnection.
        name:
            Label the connection with name (shown in NATS monitoring).
        pedantic:
            Turn on NATS server pedantic mode that performs extra checks on the protocol.
            https://docs.nats.io/using-nats/developer/connecting/misc#turn-on-pedantic-mode
        verbose:
            Verbose mode produce more feedback about code execution.
        allow_reconnect:
            Whether recover connection automatically or not.
        connect_timeout:
            Timeout in seconds to establish connection with NATS server.
        reconnect_time_wait:
            Time in seconds to wait for reestablish connection to NATS server
        max_reconnect_attempts:
            Maximum attempts number to reconnect to NATS server.
        ping_interval:
            Interval in seconds to ping.
        max_outstanding_pings:
            Maximum number of failed pings
        dont_randomize:
            Boolean indicating should client randomly shuffle servers list for reconnection randomness.
        flusher_queue_size:
            Max count of commands awaiting to be flushed to the socket
        no_echo:
            Boolean indicating should commands be echoed.
        tls_hostname:
            Hostname for TLS.
        token:
            Auth token for NATS auth.
        drain_timeout:
            Timeout in seconds to drain subscriptions.
        signature_cb:
            A callback used to sign a nonce from the server while authenticating with nkeys.
            The user should sign the nonce and return the base64 encoded signature.
        user_jwt_cb:
            A callback used to fetch and return the account signed JWT for this user.
        user_credentials:
            A user credentials file or tuple of files.
        nkeys_seed:
            Path-like object containing nkeys seed that will be used.
        nkeys_seed_str:
            Nkeys seed to be used.
        inbox_prefix:
            Prefix for generating unique inboxes, subjects with that prefix and NUID.ß
        pending_size:
            Max size of the pending buffer for publishing commands.
        flush_timeout:
            Max duration to wait for a forced flush to occur.
        ws_connection_headers:
            WebSockets connection headers.
        reconnect_to_server_handler:
            Reconnect to server handler.
        js_options:
            JetStream initialization options.
        graceful_timeout:
            Graceful shutdown timeout. Broker waits for all running subscribers completion before shut down.
        ack_policy:
            Default acknowledgement policy for all subscribers. Individual subscribers can override.
        id_generator:
            Factory used to generate `correlation_id` when a publish/request call doesn't set one explicitly.
            Defaults to `gen_cor_id` (uuid4-based).
        decoder:
            Custom decoder object.
        codec:
            Custom codec object.
        parser:
            Custom parser object.
        dependencies:
            Dependencies to apply to all broker subscribers.
        middlewares:
            "Middlewares to apply to all broker publishers/subscribers.
        routers:
            "Routers to apply to broker.
        security:
            Security options to connect broker and generate AsyncAPI server security information.
        specification_url:
            AsyncAPI hardcoded server addresses. Use `servers` if not specified.
        protocol:
            AsyncAPI server protocol.
        protocol_version:
            AsyncAPI server protocol version.
        description:
            AsyncAPI server description.
        tags:
            AsyncAPI server tags.
        logger:
            User specified logger to pass into Context and log service messages.
        log_level:
            Service messages log level.
        apply_types:
            Whether to use FastDepends or not.
        serializer:
            FastDepends-compatible serializer to validate incoming messages.
        provider:
            Provider for FastDepends.
        context:
            Context for FastDepends.
        underlying_driver_annotations: Extra driver type hints that
            FastStream cannot inject, mapped to the annotation to use
            instead. Merged over the broker's own rows. Wrap a value in
            `UnderlyingDriverAnnotation` to name the import to suggest.
    """
    servers = [servers] if isinstance(servers, str) else list(servers)

    security = _adapt_deprecated_security(
        security,
        tls_hostname=tls_hostname,
        token=token,
        signature_cb=signature_cb,
        user_jwt_cb=user_jwt_cb,
        user_credentials=user_credentials,
        nkeys_seed=nkeys_seed,
        nkeys_seed_str=nkeys_seed_str,
    )

    parsed_servers = [
        urlparse(server if "://" in server else f"//{server}") for server in servers
    ]

    if (
        security is not None
        and type(security)
        not in {
            BaseSecurity,
            NatsSecurity,
        }
        and any(parsed.username is not None for parsed in parsed_servers)
    ):
        msg = "URL credentials conflict with `security`."
        raise SetupError(msg)

    secure_kwargs = parse_security(security)
    if tls_hostname is not EMPTY and tls_hostname is not None:
        secure_kwargs["tls_hostname"] = tls_hostname

    if specification_url is not None:
        if isinstance(specification_url, str):
            specification_url = [specification_url]
        else:
            specification_url = list(specification_url)
    else:
        specification_url = [
            server.replace(parsed.netloc, parsed.netloc.rpartition("@")[-1], 1)
            for server, parsed in zip(servers, parsed_servers, strict=True)
        ]

    js_producer = NatsJSFastProducer(
        parser=parser,
        decoder=decoder,
    )

    producer = NatsFastProducerImpl(
        parser=parser,
        decoder=decoder,
    )

    super().__init__(
        # NATS options
        servers=servers,
        name=name,
        verbose=verbose,
        allow_reconnect=allow_reconnect,
        reconnect_time_wait=reconnect_time_wait,
        max_reconnect_attempts=max_reconnect_attempts,
        no_echo=no_echo,
        pedantic=pedantic,
        inbox_prefix=inbox_prefix,
        pending_size=pending_size,
        connect_timeout=connect_timeout,
        drain_timeout=drain_timeout,
        flush_timeout=flush_timeout,
        ping_interval=ping_interval,
        max_outstanding_pings=max_outstanding_pings,
        dont_randomize=dont_randomize,
        flusher_queue_size=flusher_queue_size,
        ws_connection_headers=ws_connection_headers,
        # security
        **secure_kwargs,
        # callbacks
        error_cb=self._log_connection_broken(error_cb),
        reconnected_cb=self._log_reconnected(reconnected_cb),
        disconnected_cb=disconnected_cb,
        closed_cb=closed_cb,
        discovered_server_cb=discovered_server_cb,
        reconnect_to_server_handler=reconnect_to_server_handler,
        # Basic args
        routers=routers,
        config=NatsBrokerConfig(
            producer=producer,
            js_producer=js_producer,
            js_options=js_options or {},
            # both args
            broker_middlewares=middlewares,
            broker_parser=parser,
            broker_decoder=decoder,
            broker_codec=codec,
            logger=make_nats_logger_state(
                logger=logger,
                log_level=log_level,
            ),
            fd_config=FastDependsConfig(
                use_fastdepends=apply_types,
                serializer=serializer,
                provider=provider or dependency_provider,
                context=context or ContextRepo(),
            ),
            # subscriber args
            broker_dependencies=dependencies,
            graceful_timeout=graceful_timeout,
            ack_policy=ack_policy,
            id_generator=id_generator,
            extra_context={
                "broker": self,
            },
            underlying_driver_annotations=underlying_driver_annotations or {},
        ),
        specification=BrokerSpec(
            description=description,
            url=specification_url,
            protocol=protocol,
            protocol_version=protocol_version,
            security=security,
            tags=tags,
        ),
    )