ibapi 5.4.0

//! Fluent builder for constructing a [`Client`](crate::Client).
//!
//! Replaces the v2-era `Client::connect_with_options` / `connect_with_callback`
//! entry points and folds the handshake-time notice surface into a single
//! linear chain. Pick one of two terminals based on whether handshake notices
//! matter:
//!
//! ```no_run
//! # #[cfg(feature = "async")]
//! # async fn run() -> Result<(), ibapi::Error> {
//! use ibapi::Client;
//!
//! // Connect, no handshake-notice stream
//! let client = Client::builder()
//!     .address("127.0.0.1:4002")
//!     .client_id(100)
//!     .connect()
//!     .await?;
//! drop(client);
//!
//! // Connect AND get a stream that captures handshake notices too
//! let (client, mut notices) = Client::builder()
//!     .address("127.0.0.1:4002")
//!     .client_id(101)
//!     .connect_with_notice_stream()
//!     .await?;
//! while let Some(n) = notices.next().await {
//!     println!("{n}");
//! }
//! # Ok(())
//! # }
//! ```
//!
//! The sync builder lives at `client::blocking::ClientBuilder` and mirrors the
//! shape exactly — no `.await`, [`crate::client::blocking::NoticeStream`]
//! instead of the async one.

use std::time::Duration;

use crate::connection::common::{ConnectionConfig, StartupCallback};
use crate::errors::Error;
use crate::transport::common::MAX_RECONNECT_ATTEMPTS;
use crate::transport::rate_limiter::RateLimiter;

/// Configuration state shared by [`sync_impl::ClientBuilder`] and
/// [`async_impl::ClientBuilder`]. Centralizes the field set and the
/// `InvalidArgument` validation messages so future configurators only need
/// to be added in one place. Terminals call [`BuilderState::validate`] to
/// extract the checked [`ClientConfig`].
pub(super) struct BuilderState {
    pub(super) address: Option<String>,
    pub(super) client_id: Option<i32>,
    pub(super) tcp_no_delay: bool,
    pub(super) startup_callback: Option<StartupCallback>,
    /// `None` means retry forever; the default is `Some(MAX_RECONNECT_ATTEMPTS)`.
    pub(super) max_reconnect_attempts: Option<u32>,
    /// `None` (the default) sends without limit.
    pub(super) rate_limiter: Option<RateLimiter>,
    /// `None` (the default) leaves request-id one-shots unbounded.
    pub(super) one_shot_timeout: Option<Duration>,
}

impl Default for BuilderState {
    fn default() -> Self {
        Self {
            address: None,
            client_id: None,
            tcp_no_delay: true,
            startup_callback: None,
            max_reconnect_attempts: Some(MAX_RECONNECT_ATTEMPTS),
            rate_limiter: None,
            one_shot_timeout: None,
        }
    }
}

/// Output of [`BuilderState::validate`]: the connection settings with
/// `address` and `client_id` unwrapped, plus the client-side settings.
pub(crate) struct ClientConfig {
    pub(crate) connection: ConnectionConfig,
    pub(crate) rate_limiter: Option<RateLimiter>,
    pub(crate) one_shot_timeout: Option<Duration>,
}

impl BuilderState {
    pub(super) fn validate(self) -> Result<ClientConfig, Error> {
        if self.rate_limiter.as_ref().is_some_and(|limiter| limiter.messages_per_second() == 0) {
            return Err(Error::InvalidArgument(
                "ClientBuilder: rate_limiter must allow at least 1 message per second".into(),
            ));
        }
        if self.one_shot_timeout == Some(Duration::ZERO) {
            return Err(Error::InvalidArgument("ClientBuilder: one_shot_timeout must be longer than zero".into()));
        }
        Ok(ClientConfig {
            connection: ConnectionConfig {
                address: self
                    .address
                    .ok_or_else(|| Error::InvalidArgument("ClientBuilder: address is required".into()))?,
                client_id: self
                    .client_id
                    .ok_or_else(|| Error::InvalidArgument("ClientBuilder: client_id is required".into()))?,
                tcp_no_delay: self.tcp_no_delay,
                startup_callback: self.startup_callback,
                max_reconnect_attempts: self.max_reconnect_attempts,
            },
            rate_limiter: self.rate_limiter,
            one_shot_timeout: self.one_shot_timeout,
        })
    }
}

#[cfg(feature = "sync")]
pub mod sync_impl {
    //! Sync `ClientBuilder` for the blocking transport.

    use std::sync::Arc;
    use std::time::Duration;

    use super::BuilderState;
    use crate::client::sync::Client;
    use crate::connection::common::StartupMessage;
    use crate::errors::Error;
    use crate::subscriptions::notice_stream::sync_impl::NoticeStream;
    use crate::transport::rate_limiter::RateLimiter;
    use crate::transport::sync::NoticeBroadcaster;

    /// Builder for a synchronous [`Client`]. Acquire via
    /// [`Client::builder`](crate::client::blocking::Client::builder).
    ///
    /// Configurators (`address`, `client_id`, `tcp_no_delay`, `startup_callback`)
    /// chain on `self`. Terminate with [`connect`](Self::connect) or
    /// [`connect_with_notice_stream`](Self::connect_with_notice_stream).
    #[derive(Default)]
    #[must_use = "ClientBuilder does nothing until you call connect() or connect_with_notice_stream()"]
    pub struct ClientBuilder {
        pub(super) state: BuilderState,
    }

    impl ClientBuilder {
        /// TWS / IB Gateway address, e.g. `"127.0.0.1:4002"`. Required.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # use ibapi::client::blocking::Client;
        /// let _ = Client::builder().address("127.0.0.1:4002").client_id(100).connect();
        /// ```
        pub fn address(mut self, addr: impl Into<String>) -> Self {
            self.state.address = Some(addr.into());
            self
        }

        /// Client id, e.g. `100`. Required.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # use ibapi::client::blocking::Client;
        /// let _ = Client::builder().address("127.0.0.1:4002").client_id(100).connect();
        /// ```
        pub fn client_id(mut self, id: i32) -> Self {
            self.state.client_id = Some(id);
            self
        }

        /// Set `TCP_NODELAY` on the socket. Default: `true`, matching the
        /// official IB clients.
        ///
        /// With Nagle's algorithm on, a small write issued while the previous
        /// segment is still unacknowledged is held back until that ACK arrives —
        /// up to ~40 ms (Linux) or ~200 ms (macOS) against TWS's delayed ACK.
        /// A single request never notices; a burst of orders does, because each
        /// one queues behind the ACK of the one before. Pass `false` only if you
        /// want Nagle's coalescing back.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # use ibapi::client::blocking::Client;
        /// let _ = Client::builder().address("127.0.0.1:4002").client_id(100).tcp_no_delay(false).connect();
        /// ```
        pub fn tcp_no_delay(mut self, enabled: bool) -> Self {
            self.state.tcp_no_delay = enabled;
            self
        }

        /// Overrides the reconnection attempt count. Default: 20 attempts
        /// (~7.5 minutes with the capped Fibonacci backoff). Useful when
        /// TWS/IB Gateway may be unreachable for longer, such as when a
        /// scheduled nightly restart needs a manual re-login. `0` disables
        /// reconnection entirely.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # use ibapi::client::blocking::Client;
        /// let _ = Client::builder().address("127.0.0.1:4002").client_id(100).max_reconnect_attempts(50).connect();
        /// ```
        pub fn max_reconnect_attempts(mut self, attempts: u32) -> Self {
            self.state.max_reconnect_attempts = Some(attempts);
            self
        }

        /// Remove the reconnection attempt limit: keep retrying (30 s apart
        /// once the backoff caps) until the connection is re-established.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # use ibapi::client::blocking::Client;
        /// let _ = Client::builder().address("127.0.0.1:4002").client_id(100).reconnect_forever().connect();
        /// ```
        pub fn reconnect_forever(mut self) -> Self {
            self.state.max_reconnect_attempts = None;
            self
        }

        /// Cap the messages sent to TWS with a [`RateLimiter`]. Default: no
        /// limit. Pass clones of one limiter to every client connected to the
        /// same gateway; see [`RateLimiter`] for what is counted and how.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # use ibapi::client::blocking::Client;
        /// use ibapi::RateLimiter;
        ///
        /// let _ = Client::builder()
        ///     .address("127.0.0.1:4002")
        ///     .client_id(100)
        ///     .rate_limiter(RateLimiter::default())
        ///     .connect();
        /// ```
        pub fn rate_limiter(mut self, limiter: RateLimiter) -> Self {
            self.state.rate_limiter = Some(limiter);
            self
        }

        /// Fail a one-shot request, one that returns a value rather than a
        /// subscription, with [`Error::Timeout`] when TWS hasn't answered
        /// within `timeout`. Default: requests answered on their own request id,
        /// such as `historical_data(..).fetch()`, `head_timestamp` and the
        /// option calculations, wait without a limit; the rest fail after 30 s.
        ///
        /// The clock starts when the call does and keeps running across the
        /// resends after a connection reset; a reconnect in progress is waited
        /// out, not cut short. A request TWS can cancel is cancelled on timeout.
        /// Subscriptions are not affected: bound those with
        /// `collect_to_end_within`. `timeout` must be longer than zero, or the
        /// connect fails with [`Error::InvalidArgument`].
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # use ibapi::client::blocking::Client;
        /// use std::time::Duration;
        ///
        /// let _ = Client::builder()
        ///     .address("127.0.0.1:4002")
        ///     .client_id(100)
        ///     .one_shot_timeout(Duration::from_secs(60))
        ///     .connect();
        /// ```
        pub fn one_shot_timeout(mut self, timeout: Duration) -> Self {
            self.state.one_shot_timeout = Some(timeout);
            self
        }

        /// Set a callback for unsolicited typed messages during the handshake.
        ///
        /// Fires for `OpenOrder`, `OrderStatus`, account updates, and other
        /// frames TWS emits before `next_valid_id` lands. Callback fires on the
        /// initial connect *and* every auto-reconnect handshake.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # use ibapi::client::blocking::Client;
        /// # use ibapi::StartupMessage;
        /// let _ = Client::builder()
        ///     .address("127.0.0.1:4002")
        ///     .client_id(100)
        ///     .startup_callback(|msg| if let StartupMessage::OpenOrder(o) = msg {
        ///         println!("startup open order: {}", o.order_id);
        ///     })
        ///     .connect();
        /// ```
        pub fn startup_callback(mut self, callback: impl Fn(StartupMessage) + Send + Sync + 'static) -> Self {
            self.state.startup_callback = Some(Arc::new(callback));
            self
        }

        /// Establish the connection and return a [`Client`].
        ///
        /// Handshake-time notices are not surfaced to the caller — see
        /// [`connect_with_notice_stream`](Self::connect_with_notice_stream) if
        /// you need them. Post-connect, `client.notice_stream()` still works
        /// for runtime-only unrouted notices.
        ///
        /// # Errors
        ///
        /// [`Error::ClientIdInUse`] when another connection holds the client
        /// id; connect with a different id. Not retried here.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # use ibapi::client::blocking::Client;
        /// let client = Client::builder()
        ///     .address("127.0.0.1:4002")
        ///     .client_id(100)
        ///     .connect()
        ///     .expect("connection failed");
        /// drop(client);
        /// ```
        pub fn connect(self) -> Result<Client, Error> {
            let broadcaster = Arc::new(NoticeBroadcaster::new());
            self.connect_with_broadcaster(broadcaster)
        }

        /// Establish the connection AND a pre-bound [`NoticeStream`] that
        /// captures handshake-time notices (farm-status 2104/2106/2158,
        /// connectivity 1100/1101/1102, etc.) plus every unrouted notice for
        /// the lifetime of the connection. Survives auto-reconnects.
        ///
        /// # Errors
        ///
        /// [`Error::ClientIdInUse`] when another connection holds the client
        /// id; connect with a different id. Not retried here.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # use ibapi::client::blocking::Client;
        /// let (client, notices) = Client::builder()
        ///     .address("127.0.0.1:4002")
        ///     .client_id(100)
        ///     .connect_with_notice_stream()
        ///     .expect("connection failed");
        /// for n in notices.iter() {
        ///     println!("{n}");
        /// }
        /// drop(client);
        /// ```
        pub fn connect_with_notice_stream(self) -> Result<(Client, NoticeStream), Error> {
            let broadcaster = Arc::new(NoticeBroadcaster::new());
            let stream = NoticeStream::new(broadcaster.subscribe());
            let client = self.connect_with_broadcaster(broadcaster)?;
            Ok((client, stream))
        }

        fn connect_with_broadcaster(self, broadcaster: Arc<NoticeBroadcaster>) -> Result<Client, Error> {
            Client::connect_with_config(self.state.validate()?, broadcaster)
        }
    }
}

#[cfg(feature = "async")]
pub mod async_impl {
    //! Async `ClientBuilder` for the tokio-backed transport.

    use std::sync::Arc;
    use std::time::Duration;

    use tokio::sync::broadcast;

    use super::BuilderState;
    use crate::client::r#async::Client;
    use crate::connection::common::StartupMessage;
    use crate::errors::Error;
    use crate::messages::Notice;
    use crate::subscriptions::notice_stream::async_impl::NoticeStream;
    use crate::transport::r#async::BROADCAST_CHANNEL_CAPACITY;
    use crate::transport::rate_limiter::RateLimiter;

    /// Builder for an async [`Client`]. Acquire via
    /// [`Client::builder`](crate::Client::builder).
    ///
    /// Configurators (`address`, `client_id`, `tcp_no_delay`, `startup_callback`)
    /// chain on `self`. Terminate with [`connect`](Self::connect) or
    /// [`connect_with_notice_stream`](Self::connect_with_notice_stream).
    #[derive(Default)]
    #[must_use = "ClientBuilder does nothing until you call connect() or connect_with_notice_stream()"]
    pub struct ClientBuilder {
        pub(super) state: BuilderState,
        channel_capacity: Option<usize>,
    }

    impl ClientBuilder {
        /// TWS / IB Gateway address, e.g. `"127.0.0.1:4002"`. Required.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # async fn run() -> Result<(), ibapi::Error> {
        /// use ibapi::Client;
        /// let _client = Client::builder().address("127.0.0.1:4002").client_id(100).connect().await?;
        /// # Ok(()) }
        /// ```
        pub fn address(mut self, addr: impl Into<String>) -> Self {
            self.state.address = Some(addr.into());
            self
        }

        /// Client id, e.g. `100`. Required.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # async fn run() -> Result<(), ibapi::Error> {
        /// use ibapi::Client;
        /// let _client = Client::builder().address("127.0.0.1:4002").client_id(100).connect().await?;
        /// # Ok(()) }
        /// ```
        pub fn client_id(mut self, id: i32) -> Self {
            self.state.client_id = Some(id);
            self
        }

        /// Set `TCP_NODELAY` on the socket. Default: `true`, matching the
        /// official IB clients.
        ///
        /// With Nagle's algorithm on, a small write issued while the previous
        /// segment is still unacknowledged is held back until that ACK arrives —
        /// up to ~40 ms (Linux) or ~200 ms (macOS) against TWS's delayed ACK.
        /// A single request never notices; a burst of orders does, because each
        /// one queues behind the ACK of the one before. Pass `false` only if you
        /// want Nagle's coalescing back.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # async fn run() -> Result<(), ibapi::Error> {
        /// use ibapi::Client;
        /// let _client = Client::builder().address("127.0.0.1:4002").client_id(100).tcp_no_delay(false).connect().await?;
        /// # Ok(()) }
        /// ```
        pub fn tcp_no_delay(mut self, enabled: bool) -> Self {
            self.state.tcp_no_delay = enabled;
            self
        }

        /// Overrides the reconnection attempt count. Default: 20 attempts
        /// (~7.5 minutes with the capped Fibonacci backoff). Useful when
        /// TWS/IB Gateway may be unreachable for longer, such as when a
        /// scheduled nightly restart needs a manual re-login. `0` disables
        /// reconnection entirely.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # async fn run() -> Result<(), ibapi::Error> {
        /// use ibapi::Client;
        /// let _client = Client::builder().address("127.0.0.1:4002").client_id(100).max_reconnect_attempts(50).connect().await?;
        /// # Ok(()) }
        /// ```
        pub fn max_reconnect_attempts(mut self, attempts: u32) -> Self {
            self.state.max_reconnect_attempts = Some(attempts);
            self
        }

        /// Remove the reconnection attempt limit: keep retrying (30 s apart
        /// once the backoff caps) until the connection is re-established.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # async fn run() -> Result<(), ibapi::Error> {
        /// use ibapi::Client;
        /// let _client = Client::builder().address("127.0.0.1:4002").client_id(100).reconnect_forever().connect().await?;
        /// # Ok(()) }
        /// ```
        pub fn reconnect_forever(mut self) -> Self {
            self.state.max_reconnect_attempts = None;
            self
        }

        /// Set the per-subscription broadcast channel capacity for market data
        /// and other non-order streams (default 1024).
        ///
        /// Each of these subscriptions reads from a bounded broadcast channel;
        /// a consumer that falls more than `capacity` frames behind has the
        /// oldest frames evicted and receives a
        /// [`SUBSCRIPTION_LAG_CODE`](crate::SUBSCRIPTION_LAG_CODE)
        /// notice naming the dropped count. Raise the capacity if your
        /// consumers legitimately fall behind during bursts; the cost is
        /// memory per in-flight subscription.
        ///
        /// Order channels have their own floors, which this setting can raise
        /// but not lower: 1024 per order (`place_order`, `cancel_order`,
        /// `exercise_options`) and per `executions` request, 8192 for
        /// `order_update_stream` and the open/completed-order streams.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// use ibapi::Client;
        /// # async fn run() -> Result<(), ibapi::Error> {
        /// let client = Client::builder()
        ///     .address("127.0.0.1:4002")
        ///     .client_id(100)
        ///     .channel_capacity(8192)
        ///     .connect()
        ///     .await?;
        /// # Ok(()) }
        /// ```
        pub fn channel_capacity(mut self, capacity: usize) -> Self {
            self.channel_capacity = Some(capacity);
            self
        }

        /// Cap the messages sent to TWS with a [`RateLimiter`]. Default: no
        /// limit. Pass clones of one limiter to every client connected to the
        /// same gateway; see [`RateLimiter`] for what is counted and how.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # async fn run() -> Result<(), ibapi::Error> {
        /// use ibapi::{Client, RateLimiter};
        ///
        /// let _client = Client::builder()
        ///     .address("127.0.0.1:4002")
        ///     .client_id(100)
        ///     .rate_limiter(RateLimiter::default())
        ///     .connect()
        ///     .await?;
        /// # Ok(()) }
        /// ```
        pub fn rate_limiter(mut self, limiter: RateLimiter) -> Self {
            self.state.rate_limiter = Some(limiter);
            self
        }

        /// Fail a one-shot request, one that returns a value rather than a
        /// subscription, with [`Error::Timeout`] when TWS hasn't answered
        /// within `timeout`. Default: requests answered on their own request id,
        /// such as `historical_data(..).fetch()`, `head_timestamp` and the
        /// option calculations, wait without a limit; the rest fail after 30 s.
        ///
        /// The clock starts when the call does and keeps running across the
        /// resends after a connection reset; a reconnect in progress is waited
        /// out, not cut short. A request TWS can cancel is cancelled on timeout.
        /// Subscriptions are not affected: bound those with
        /// `collect_to_end_within`. `timeout` must be longer than zero, or the
        /// connect fails with [`Error::InvalidArgument`].
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # async fn run() -> Result<(), ibapi::Error> {
        /// use ibapi::Client;
        /// use std::time::Duration;
        ///
        /// let _client = Client::builder()
        ///     .address("127.0.0.1:4002")
        ///     .client_id(100)
        ///     .one_shot_timeout(Duration::from_secs(60))
        ///     .connect()
        ///     .await?;
        /// # Ok(()) }
        /// ```
        pub fn one_shot_timeout(mut self, timeout: Duration) -> Self {
            self.state.one_shot_timeout = Some(timeout);
            self
        }

        /// Set a callback for unsolicited typed messages during the handshake.
        ///
        /// Fires for `OpenOrder`, `OrderStatus`, account updates, and other
        /// frames TWS emits before `next_valid_id` lands. Callback fires on the
        /// initial connect *and* every auto-reconnect handshake.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # async fn run() -> Result<(), ibapi::Error> {
        /// use ibapi::{Client, StartupMessage};
        /// let _client = Client::builder()
        ///     .address("127.0.0.1:4002")
        ///     .client_id(100)
        ///     .startup_callback(|msg| if let StartupMessage::OpenOrder(o) = msg {
        ///         println!("startup open order: {}", o.order_id);
        ///     })
        ///     .connect()
        ///     .await?;
        /// # Ok(()) }
        /// ```
        pub fn startup_callback(mut self, callback: impl Fn(StartupMessage) + Send + Sync + 'static) -> Self {
            self.state.startup_callback = Some(Arc::new(callback));
            self
        }

        /// Establish the connection and return a [`Client`].
        ///
        /// Handshake-time notices are not surfaced to the caller — see
        /// [`connect_with_notice_stream`](Self::connect_with_notice_stream) if
        /// you need them. Post-connect, `client.notice_stream()` still works
        /// for runtime-only unrouted notices.
        ///
        /// # Errors
        ///
        /// [`Error::ClientIdInUse`] when another connection holds the client
        /// id; connect with a different id. Not retried here.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # async fn run() -> Result<(), ibapi::Error> {
        /// use ibapi::Client;
        /// let client = Client::builder()
        ///     .address("127.0.0.1:4002")
        ///     .client_id(100)
        ///     .connect()
        ///     .await?;
        /// drop(client);
        /// # Ok(()) }
        /// ```
        pub async fn connect(self) -> Result<Client, Error> {
            let (sender, _rx) = broadcast::channel::<Notice>(BROADCAST_CHANNEL_CAPACITY);
            self.connect_with_sender(sender).await
        }

        /// Establish the connection AND a pre-bound [`NoticeStream`] that
        /// captures handshake-time notices (farm-status 2104/2106/2158,
        /// connectivity 1100/1101/1102, etc.) plus every unrouted notice for
        /// the lifetime of the connection. Survives auto-reconnects.
        ///
        /// # Errors
        ///
        /// [`Error::ClientIdInUse`] when another connection holds the client
        /// id; connect with a different id. Not retried here.
        ///
        /// # Examples
        ///
        /// ```no_run
        /// # async fn run() -> Result<(), ibapi::Error> {
        /// use ibapi::Client;
        /// let (client, mut notices) = Client::builder()
        ///     .address("127.0.0.1:4002")
        ///     .client_id(100)
        ///     .connect_with_notice_stream()
        ///     .await?;
        /// while let Some(n) = notices.next().await {
        ///     println!("{n}");
        /// }
        /// drop(client);
        /// # Ok(()) }
        /// ```
        pub async fn connect_with_notice_stream(self) -> Result<(Client, NoticeStream), Error> {
            let (sender, receiver) = broadcast::channel::<Notice>(BROADCAST_CHANNEL_CAPACITY);
            let stream = NoticeStream::new(receiver);
            let client = self.connect_with_sender(sender).await?;
            Ok((client, stream))
        }

        async fn connect_with_sender(self, sender: broadcast::Sender<Notice>) -> Result<Client, Error> {
            if self.channel_capacity == Some(0) {
                // tokio's broadcast::channel panics on capacity 0; fail the
                // build instead, at the seam every other invalid input uses.
                return Err(Error::InvalidArgument("ClientBuilder: channel_capacity must be at least 1".into()));
            }
            let channel_capacity = self.channel_capacity.unwrap_or(BROADCAST_CHANNEL_CAPACITY);
            Client::connect_with_config(self.state.validate()?, sender, channel_capacity).await
        }
    }
}

#[cfg(test)]
#[path = "client_builder_tests.rs"]
mod tests;