google_cloud_gax/

options.rs

// Copyright 2024 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Client configuration and per request options.
//!
//! While the client library  defaults are intended to work for most
//! applications, it is sometimes necessary to change the configuration. Notably
//! the default endpoint, and the default authentication credentials do not work
//! for some applications.
//!
//! Likewise, applications may need to customize the behavior of some calls made
//! via a client, even a customized one. Applications sometimes change the
//! timeout for an specific call, or change the retry configuration. The
//! `*Builder` returned by each client method implements the
//! [RequestOptionsBuilder] trait where applications can override some defaults.

use crate::backoff_policy::{BackoffPolicy, BackoffPolicyArg};
use crate::polling_backoff_policy::{PollingBackoffPolicy, PollingBackoffPolicyArg};
use crate::polling_error_policy::{PollingErrorPolicy, PollingErrorPolicyArg};
use crate::retry_policy::{RetryPolicy, RetryPolicyArg};
use crate::retry_throttler::{RetryThrottlerArg, SharedRetryThrottler};
use std::sync::Arc;

/// A set of options configuring a single request.
///
/// Application only use this class directly in mocks, where they may want to
/// verify their application has configured all the right request parameters and
/// options.
///
/// All other code uses this type indirectly, via the per-request builders.
#[derive(Clone, Debug, Default)]
pub struct RequestOptions {
    idempotent: Option<bool>,
    user_agent: Option<String>,
    quota_project: Option<String>,
    attempt_timeout: Option<std::time::Duration>,
    retry_policy: Option<Arc<dyn RetryPolicy>>,
    backoff_policy: Option<Arc<dyn BackoffPolicy>>,
    retry_throttler: Option<SharedRetryThrottler>,
    polling_error_policy: Option<Arc<dyn PollingErrorPolicy>>,
    polling_backoff_policy: Option<Arc<dyn PollingBackoffPolicy>>,
    request_stream_channel_capacity: Option<usize>,
    extensions: http::Extensions,
}

impl RequestOptions {
    /// Gets the idempotency
    pub fn idempotent(&self) -> Option<bool> {
        self.idempotent
    }

    /// Treat the RPC underlying RPC in this method as idempotent.
    ///
    /// If a retry policy is configured, the policy may examine the idempotency
    /// and the error details to decide if the error is retryable. Typically
    /// [idempotent] RPCs are safe to retry under more error conditions
    /// than non-idempotent RPCs.
    ///
    /// The client libraries provide a default for RPC idempotency, based on the
    /// HTTP method (`GET`, `POST`, `DELETE`, etc.).
    ///
    /// [idempotent]: https://en.wikipedia.org/wiki/Idempotence
    pub fn set_idempotency(&mut self, value: bool) {
        self.idempotent = Some(value);
    }

    /// Set the idempotency for the underlying RPC unless it is already set.
    ///
    /// If [set_idempotency][Self::set_idempotency] was already called this
    /// method has no effect. Otherwise it sets the idempotency. The client
    /// libraries use this to provide a default idempotency value.
    pub(crate) fn set_default_idempotency(&mut self, default: bool) {
        self.idempotent.get_or_insert(default);
    }

    /// Prepends this prefix to the user agent header value.
    pub fn set_user_agent<T: Into<String>>(&mut self, v: T) {
        self.user_agent = Some(v.into());
    }

    /// Gets the current user-agent prefix
    pub fn user_agent(&self) -> &Option<String> {
        &self.user_agent
    }

    /// Sets the [quota project] for the request.
    ///
    /// This adds the `x-goog-user-project` header to the request. Note that
    /// setting this option overrides a credentials' quota project.
    ///
    /// [quota project]: https://docs.cloud.google.com/docs/quotas/quota-project
    pub fn set_quota_project<T: Into<String>>(&mut self, v: T) {
        self.quota_project = Some(v.into());
    }

    /// Gets the current quota project.
    pub fn quota_project(&self) -> &Option<String> {
        &self.quota_project
    }

    /// Sets the per-attempt timeout.
    ///
    /// When using a retry loop, this affects the timeout for each attempt. The
    /// overall timeout for a request is set by the retry policy.
    pub fn set_attempt_timeout<T: Into<std::time::Duration>>(&mut self, v: T) {
        self.attempt_timeout = Some(v.into());
    }

    /// Gets the current per-attempt timeout.
    pub fn attempt_timeout(&self) -> &Option<std::time::Duration> {
        &self.attempt_timeout
    }

    /// Get the current retry policy override, if any.
    pub fn retry_policy(&self) -> &Option<Arc<dyn RetryPolicy>> {
        &self.retry_policy
    }

    /// Sets the retry policy configuration.
    pub fn set_retry_policy<V: Into<RetryPolicyArg>>(&mut self, v: V) {
        self.retry_policy = Some(v.into().into());
    }

    /// Get the current backoff policy override, if any.
    pub fn backoff_policy(&self) -> &Option<Arc<dyn BackoffPolicy>> {
        &self.backoff_policy
    }

    /// Sets the backoff policy configuration.
    pub fn set_backoff_policy<V: Into<BackoffPolicyArg>>(&mut self, v: V) {
        self.backoff_policy = Some(v.into().into());
    }

    /// Get the current retry throttler override, if any.
    pub fn retry_throttler(&self) -> &Option<SharedRetryThrottler> {
        &self.retry_throttler
    }

    /// Sets the retry throttling configuration.
    pub fn set_retry_throttler<V: Into<RetryThrottlerArg>>(&mut self, v: V) {
        self.retry_throttler = Some(v.into().into());
    }

    /// Get the current polling policy override, if any.
    pub fn polling_error_policy(&self) -> &Option<Arc<dyn PollingErrorPolicy>> {
        &self.polling_error_policy
    }

    /// Sets the polling policy configuration.
    pub fn set_polling_error_policy<V: Into<PollingErrorPolicyArg>>(&mut self, v: V) {
        self.polling_error_policy = Some(v.into().0);
    }

    /// Get the current polling backoff policy override, if any.
    pub fn polling_backoff_policy(&self) -> &Option<Arc<dyn PollingBackoffPolicy>> {
        &self.polling_backoff_policy
    }

    /// Sets the backoff policy configuration.
    pub fn set_polling_backoff_policy<V: Into<PollingBackoffPolicyArg>>(&mut self, v: V) {
        self.polling_backoff_policy = Some(v.into().0);
    }

    /// Gets the current request stream channel capacity, if set.
    pub fn request_stream_channel_capacity(&self) -> Option<usize> {
        self.request_stream_channel_capacity
    }

    /// Sets the buffer capacity of the internal request channel for streaming RPCs.
    ///
    /// Valid values are between `1` and `usize::MAX >> 3`. Values outside this range will be clamped.
    pub fn set_request_stream_channel_capacity(&mut self, capacity: usize) {
        self.request_stream_channel_capacity =
            Some(capacity.clamp(1, MAX_REQUEST_CHANNEL_CAPACITY));
    }
}

/// The maximum allowed capacity for a bidirectional stream request channel.
///
/// This upper limit matches Tokio's internal `mpsc::channel` capacity limit
/// (`usize::MAX >> 3`).
const MAX_REQUEST_CHANNEL_CAPACITY: usize = usize::MAX >> 3;

/// Implementations of this trait provide setters to configure request options.
///
/// The Google Cloud Client Libraries for Rust provide a builder for each RPC.
/// These builders can be used to set the request parameters, e.g., the name of
/// the resource targeted by the RPC, as well as any options affecting the
/// request, such as additional headers or timeouts.
pub trait RequestOptionsBuilder: internal::RequestBuilder {
    /// If `v` is `true`, treat the RPC underlying this method as idempotent.
    fn with_idempotency(self, v: bool) -> Self;

    /// Set the user agent header.
    fn with_user_agent<V: Into<String>>(self, v: V) -> Self;

    /// Sets the per-attempt timeout.
    ///
    /// When using a retry loop, this affects the timeout for each attempt. The
    /// overall timeout for a request is set by the retry policy.
    fn with_attempt_timeout<V: Into<std::time::Duration>>(self, v: V) -> Self;

    /// Sets the retry policy configuration.
    fn with_retry_policy<V: Into<RetryPolicyArg>>(self, v: V) -> Self;

    /// Sets the backoff policy configuration.
    fn with_backoff_policy<V: Into<BackoffPolicyArg>>(self, v: V) -> Self;

    /// Sets the retry throttler configuration.
    fn with_retry_throttler<V: Into<RetryThrottlerArg>>(self, v: V) -> Self;

    /// Sets the polling error policy configuration.
    fn with_polling_error_policy<V: Into<PollingErrorPolicyArg>>(self, v: V) -> Self;

    /// Sets the polling backoff policy configuration.
    fn with_polling_backoff_policy<V: Into<PollingBackoffPolicyArg>>(self, v: V) -> Self;

    // Methods with a default implementation.
    // See https://github.com/googleapis/google-cloud-rust/pull/5490 for context.

    /// Sets the [quota project] for the request.
    ///
    /// This adds the `x-goog-user-project` header to the request. Note that
    /// setting this option overrides a credentials' quota project.
    ///
    /// [quota project]: https://docs.cloud.google.com/docs/quotas/quota-project
    fn with_quota_project<V: Into<String>>(self, _v: V) -> Self
    where
        Self: Sized,
    {
        unimplemented!();
    }

    /// Injects a custom HTTP header (or gRPC metadata) into this specific request.
    ///
    /// This method will ignore attempts to set system headers used by the client
    /// library. These include: `user-agent`, `authorization`,
    /// `x-goog-api-key`, `x-goog-api-client`, `x-goog-user-project`,
    /// and `x-goog-request-params`.
    ///
    /// Callers who want to configure respective system headers should use the dedicated methods instead:
    /// * `user-agent`: Use [`with_user_agent`](Self::with_user_agent) (or globally via `ClientBuilder`).
    /// * `x-goog-user-project`: Use [`with_quota_project`](Self::with_quota_project).
    /// * `authorization` and `x-goog-api-key`: Configure globally via `ClientBuilder::with_credentials`.
    ///
    /// Note that `x-goog-api-client` and `x-goog-request-params` are dynamically auto-generated by the SDK
    /// for telemetry and routing, and cannot be customized by callers.
    ///
    /// ### Repeated Headers
    ///
    /// Under HTTP RFC 9110 (Section 5.3), standard repeated headers (e.g. `Cache-Control`, `Accept`, or `Allow`)
    /// are defined as comma-separated lists. Callers can supply repeated values natively as a single
    /// comma-separated [`HeaderValue`]:
    ///
    /// ```
    /// # use google_cloud_gax::options::RequestOptionsBuilder;
    /// use http::header::{HeaderName, HeaderValue};
    /// fn sample<T: RequestOptionsBuilder>(builder: T) -> T {
    ///     builder.with_custom_header(
    ///         HeaderName::from_static("cache-control"),
    ///         HeaderValue::from_static("no-cache, no-store"),
    ///     )
    /// }
    /// ```
    ///
    /// [`HeaderValue`]: http::header::HeaderValue
    fn with_custom_header(
        mut self,
        name: http::header::HeaderName,
        value: http::header::HeaderValue,
    ) -> Self
    where
        Self: Sized,
    {
        use internal::RequestOptionsExt;
        let mut headers = self
            .request_options()
            .get_extension::<http::HeaderMap>()
            .cloned()
            .unwrap_or_default();
        headers.insert(name, value);
        let mut options = std::mem::take(self.request_options());
        options = options.insert_extension(headers);
        *self.request_options() = options;
        self
    }

    /// Sets the buffer capacity of the internal request channel for streaming RPCs.
    ///
    /// Valid values are between `1` and `usize::MAX >> 3`. The default
    /// capacity is `16`. Values outside this range will be clamped.
    fn with_request_stream_channel_capacity(self, _capacity: usize) -> Self
    where
        Self: Sized,
    {
        unimplemented!();
    }
}

#[cfg_attr(not(feature = "_internal-semver"), doc(hidden))]
#[allow(missing_docs)]
pub mod internal {
    //! This module contains implementation details. It is not part of the
    //! public API. Types and functions in this module may be changed or removed
    //! without warnings. Applications should not use any types contained
    //! within.
    use super::RequestOptions;

    /// Simplify implementation of the [super::RequestOptionsBuilder] trait in
    /// generated code.
    ///
    /// This is an implementation detail, most applications have little need to
    /// worry about or use this trait.
    pub trait RequestBuilder {
        fn request_options(&mut self) -> &mut RequestOptions;
    }

    pub fn set_default_idempotency(mut options: RequestOptions, default: bool) -> RequestOptions {
        options.set_default_idempotency(default);
        options
    }

    mod sealed {
        pub trait OptionsExt {}
    }

    /// Access the `RequestOption` extensions.
    ///
    /// The client library internals can use this trait to attach extension
    /// values to the request options. Possibly passing (nearly) arbitrary types
    /// between layers.
    ///
    /// This is useful when (for example) the tracing layer wants to pass
    /// information to the HTTP or gRPC client, without having to change all the
    /// intermediate types, which may include public interfaces.
    pub trait RequestOptionsExt: sealed::OptionsExt {
        /// Gets an extension value.
        fn get_extension<T>(&self) -> Option<&T>
        where
            T: Send + Sync + 'static;

        /// Gets a mutable reference to an extension value.
        fn get_extension_mut<T>(&mut self) -> Option<&mut T>
        where
            T: Send + Sync + 'static;

        /// Gets a mutable reference to an extension value, inserting the default if it does not exist.
        fn get_extension_or_default_mut<T>(&mut self) -> &mut T
        where
            T: Default + Clone + Send + Sync + 'static;

        /// Sets an extension value.
        fn insert_extension<T>(self, value: T) -> Self
        where
            T: Clone + Send + Sync + 'static;
    }

    impl sealed::OptionsExt for RequestOptions {}
    impl RequestOptionsExt for RequestOptions {
        fn get_extension<T>(&self) -> Option<&T>
        where
            T: Send + Sync + 'static,
        {
            self.extensions.get::<T>()
        }

        fn get_extension_mut<T>(&mut self) -> Option<&mut T>
        where
            T: Send + Sync + 'static,
        {
            self.extensions.get_mut::<T>()
        }

        fn get_extension_or_default_mut<T>(&mut self) -> &mut T
        where
            T: Default + Clone + Send + Sync + 'static,
        {
            if self.extensions.get::<T>().is_none() {
                let _ = self.extensions.insert(T::default());
            }
            self.extensions
                .get_mut::<T>()
                .expect("value was just inserted if missing")
        }

        fn insert_extension<T>(mut self, value: T) -> Self
        where
            T: Clone + Send + Sync + 'static,
        {
            let _ = self.extensions.insert(value);
            self
        }
    }

    #[derive(Debug, Clone, Default, PartialEq)]
    pub struct PathTemplate(pub &'static str);

    #[derive(Debug, Clone, Default, PartialEq)]
    pub struct ResourceName(pub String);

    // Cannot remove this function, as that would break any client libraries
    // that are released and use this function.
    #[deprecated]
    pub fn set_path_template(
        options: RequestOptions,
        path_template: &'static str,
    ) -> RequestOptions {
        options.insert_extension(PathTemplate(path_template))
    }

    // Cannot remove this function, as that would break any client libraries
    // that are released and use this function.
    #[deprecated]
    pub fn get_path_template(options: &RequestOptions) -> Option<&'static str> {
        options.get_extension::<PathTemplate>().map(|e| e.0)
    }
}

/// Implements the sealed [RequestOptionsBuilder] trait.
impl<T> RequestOptionsBuilder for T
where
    T: internal::RequestBuilder,
{
    fn with_idempotency(mut self, v: bool) -> Self {
        self.request_options().set_idempotency(v);
        self
    }

    fn with_user_agent<V: Into<String>>(mut self, v: V) -> Self {
        self.request_options().set_user_agent(v);
        self
    }

    fn with_quota_project<V: Into<String>>(mut self, v: V) -> Self {
        self.request_options().set_quota_project(v);
        self
    }

    fn with_attempt_timeout<V: Into<std::time::Duration>>(mut self, v: V) -> Self {
        self.request_options().set_attempt_timeout(v);
        self
    }

    fn with_retry_policy<V: Into<RetryPolicyArg>>(mut self, v: V) -> Self {
        self.request_options().set_retry_policy(v);
        self
    }

    fn with_backoff_policy<V: Into<BackoffPolicyArg>>(mut self, v: V) -> Self {
        self.request_options().set_backoff_policy(v);
        self
    }

    fn with_retry_throttler<V: Into<RetryThrottlerArg>>(mut self, v: V) -> Self {
        self.request_options().set_retry_throttler(v);
        self
    }

    fn with_polling_error_policy<V: Into<PollingErrorPolicyArg>>(mut self, v: V) -> Self {
        self.request_options().set_polling_error_policy(v);
        self
    }

    fn with_polling_backoff_policy<V: Into<PollingBackoffPolicyArg>>(mut self, v: V) -> Self {
        self.request_options().set_polling_backoff_policy(v);
        self
    }

    fn with_request_stream_channel_capacity(mut self, capacity: usize) -> Self {
        self.request_options()
            .set_request_stream_channel_capacity(capacity);
        self
    }
}

#[cfg(test)]
mod tests {
    use super::internal::*;
    use super::*;
    use crate::exponential_backoff::ExponentialBackoffBuilder;
    use crate::polling_error_policy;
    use crate::retry_policy::LimitedAttemptCount;
    use crate::retry_throttler::AdaptiveThrottler;
    use static_assertions::{assert_impl_all, assert_not_impl_all};
    use std::panic::{RefUnwindSafe, UnwindSafe};
    use std::time::Duration;

    #[derive(Debug, Default)]
    struct TestBuilder {
        request_options: RequestOptions,
    }
    impl RequestBuilder for TestBuilder {
        fn request_options(&mut self) -> &mut RequestOptions {
            &mut self.request_options
        }
    }

    #[test]
    fn traits() {
        assert_impl_all!(RequestOptions: Clone, Send, Sync, Unpin, std::fmt::Debug);
        assert_not_impl_all!(RequestOptions: RefUnwindSafe, UnwindSafe);
    }

    #[test]
    fn request_options() {
        const USER_AGENT: &str = "test-only";
        const USER_PROJECT: &str = "test-project";

        let mut opts = RequestOptions::default();

        assert_eq!(opts.idempotent, None);
        opts.set_idempotency(true);
        assert_eq!(opts.idempotent(), Some(true));
        opts.set_idempotency(false);
        assert_eq!(opts.idempotent(), Some(false));

        opts.set_user_agent(USER_AGENT);
        assert_eq!(opts.user_agent().as_deref(), Some(USER_AGENT));
        assert_eq!(opts.attempt_timeout(), &None);

        opts.set_quota_project(USER_PROJECT);
        assert_eq!(opts.quota_project().as_deref(), Some(USER_PROJECT));

        let d = Duration::from_secs(123);
        opts.set_attempt_timeout(d);
        assert_eq!(opts.user_agent().as_deref(), Some(USER_AGENT));
        assert_eq!(opts.attempt_timeout(), &Some(d));

        opts.set_retry_policy(LimitedAttemptCount::new(3));
        assert!(opts.retry_policy().is_some(), "{opts:?}");

        opts.set_backoff_policy(ExponentialBackoffBuilder::new().clamp());
        assert!(opts.backoff_policy().is_some(), "{opts:?}");

        opts.set_retry_throttler(AdaptiveThrottler::default());
        assert!(opts.retry_throttler().is_some(), "{opts:?}");

        opts.set_polling_error_policy(polling_error_policy::Aip194Strict);
        assert!(opts.polling_error_policy().is_some(), "{opts:?}");

        opts.set_polling_backoff_policy(ExponentialBackoffBuilder::new().clamp());
        assert!(opts.polling_backoff_policy().is_some(), "{opts:?}");
    }

    #[test]
    fn request_options_idempotency() {
        let opts = set_default_idempotency(RequestOptions::default(), true);
        assert_eq!(opts.idempotent(), Some(true));
        let opts = set_default_idempotency(opts, false);
        assert_eq!(opts.idempotent(), Some(true));

        let opts = set_default_idempotency(RequestOptions::default(), false);
        assert_eq!(opts.idempotent(), Some(false));
        let opts = set_default_idempotency(opts, true);
        assert_eq!(opts.idempotent(), Some(false));
    }

    #[test]
    fn request_options_ext() {
        #[derive(Debug, Clone, PartialEq)]
        struct TestA(&'static str);
        #[derive(Debug, Clone, PartialEq)]
        struct TestB(u32);

        let opts = RequestOptions::default();
        assert!(opts.get_extension::<TestA>().is_none(), "{opts:?}");
        assert!(opts.get_extension::<TestB>().is_none(), "{opts:?}");
        let opts = opts.insert_extension(TestA("1"));
        assert_eq!(opts.get_extension::<TestA>(), Some(&TestA("1")), "{opts:?}");
        assert!(opts.get_extension::<TestB>().is_none(), "{opts:?}");
        let opts = opts
            .insert_extension(TestA("2"))
            .insert_extension(TestB(42));
        assert_eq!(opts.get_extension::<TestA>(), Some(&TestA("2")), "{opts:?}");
        assert_eq!(opts.get_extension::<TestB>(), Some(&TestB(42)), "{opts:?}");
    }

    #[test]
    fn request_options_ext_mut() {
        #[derive(Debug, Clone, Default, PartialEq)]
        struct TestCounter(u32);

        let mut opts = RequestOptions::default();

        // 1. get_extension_mut returns None when not present.
        assert!(opts.get_extension_mut::<TestCounter>().is_none());

        // 2. get_extension_or_default_mut inserts default TestCounter(0) and returns mutable reference.
        let counter = opts.get_extension_or_default_mut::<TestCounter>();
        assert_eq!(counter, &mut TestCounter(0));
        counter.0 += 10;

        // 3. get_extension_mut returns Some(&mut TestCounter(10)) when present.
        let counter = opts
            .get_extension_mut::<TestCounter>()
            .expect("counter extension should be present after insertion");
        assert_eq!(counter, &mut TestCounter(10));
        counter.0 += 10;

        // 4. Second call to get_extension_or_default_mut returns existing reference without resetting.
        let counter2 = opts.get_extension_or_default_mut::<TestCounter>();
        assert_eq!(counter2, &mut TestCounter(20));
        counter2.0 += 5;

        assert_eq!(opts.get_extension::<TestCounter>(), Some(&TestCounter(25)));
    }

    #[test]
    fn request_options_builder() -> anyhow::Result<()> {
        const USER_AGENT: &str = "test-only";
        const USER_PROJECT: &str = "test-project";

        let mut builder = TestBuilder::default();
        assert_eq!(builder.request_options().user_agent(), &None);
        assert_eq!(builder.request_options().quota_project(), &None);
        assert_eq!(builder.request_options().attempt_timeout(), &None);

        let mut builder = TestBuilder::default().with_idempotency(true);
        assert_eq!(builder.request_options().idempotent(), Some(true));
        let mut builder = TestBuilder::default().with_idempotency(false);
        assert_eq!(builder.request_options().idempotent(), Some(false));

        let mut builder = TestBuilder::default().with_user_agent(USER_AGENT);
        assert_eq!(
            builder.request_options().user_agent().as_deref(),
            Some(USER_AGENT)
        );
        assert_eq!(builder.request_options().attempt_timeout(), &None);

        let mut builder = TestBuilder::default().with_quota_project(USER_PROJECT);
        assert_eq!(
            builder.request_options().quota_project().as_deref(),
            Some(USER_PROJECT)
        );

        let d = Duration::from_secs(123);
        let mut builder = TestBuilder::default().with_attempt_timeout(d);
        assert_eq!(builder.request_options().user_agent(), &None);
        assert_eq!(builder.request_options().attempt_timeout(), &Some(d));

        let mut builder = TestBuilder::default().with_retry_policy(LimitedAttemptCount::new(3));
        assert!(
            builder.request_options().retry_policy().is_some(),
            "{builder:?}"
        );

        let mut builder =
            TestBuilder::default().with_backoff_policy(ExponentialBackoffBuilder::new().build()?);
        assert!(
            builder.request_options().backoff_policy().is_some(),
            "{builder:?}"
        );

        let mut builder = TestBuilder::default().with_retry_throttler(AdaptiveThrottler::default());
        assert!(
            builder.request_options().retry_throttler().is_some(),
            "{builder:?}"
        );

        let mut builder =
            TestBuilder::default().with_polling_error_policy(polling_error_policy::Aip194Strict);
        assert!(
            builder.request_options().polling_error_policy().is_some(),
            "{builder:?}"
        );

        let mut builder = TestBuilder::default()
            .with_polling_backoff_policy(ExponentialBackoffBuilder::new().build()?);
        assert!(
            builder.request_options().polling_backoff_policy().is_some(),
            "{builder:?}"
        );

        Ok(())
    }

    #[test]
    fn request_options_builder_custom_headers() {
        let mut builder = TestBuilder::default()
            .with_custom_header(
                http::header::HeaderName::from_static("x-custom-1"),
                http::header::HeaderValue::from_static("value1"),
            )
            .with_custom_header(
                http::header::HeaderName::from_static("x-custom-2"),
                http::header::HeaderValue::from_static("value2"),
            );

        let headers = builder
            .request_options()
            .get_extension::<http::HeaderMap>()
            .expect("headers extension should be present");

        assert_eq!(
            headers.get("x-custom-1").and_then(|v| v.to_str().ok()),
            Some("value1")
        );
        assert_eq!(
            headers.get("x-custom-2").and_then(|v| v.to_str().ok()),
            Some("value2")
        );
    }

    #[test]
    fn request_options_builder_request_stream_channel_capacity() {
        let builder = TestBuilder::default();
        assert_eq!(
            builder.request_options.request_stream_channel_capacity(),
            None
        );

        let builder = TestBuilder::default().with_request_stream_channel_capacity(32);
        assert_eq!(
            builder.request_options.request_stream_channel_capacity(),
            Some(32)
        );

        // Clamping tests
        let builder = TestBuilder::default().with_request_stream_channel_capacity(0);
        assert_eq!(
            builder.request_options.request_stream_channel_capacity(),
            Some(1)
        );

        let builder = TestBuilder::default().with_request_stream_channel_capacity(usize::MAX);
        assert_eq!(
            builder.request_options.request_stream_channel_capacity(),
            Some(MAX_REQUEST_CHANNEL_CAPACITY)
        );
    }
}