1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

use std::{sync::Arc, time::Duration};

use derivative::Derivative;
use serde::Deserialize;
use typed_builder::TypedBuilder;

use crate::{
    bson_util,
    client::auth::Credential,
    event::cmap::CmapEventHandler,
    options::{ClientOptions, DriverInfo, StreamAddress, TlsOptions},
};

/// Contains the options for creating a connection pool. While these options are specified at the
/// client-level, `ConnectionPoolOptions` is exposed for the purpose of CMAP event handling.
#[derive(Clone, Default, Deserialize, TypedBuilder, Derivative)]
#[derivative(Debug, PartialEq)]
#[serde(rename_all = "camelCase")]
pub struct ConnectionPoolOptions {
    /// The application name specified by the user. This is sent to the server as part of the
    /// handshake that each connection makes when it's created.
    #[builder(default)]
    pub app_name: Option<String>,

    /// The connect timeout passed to each underlying TcpStream when attemtping to connect to the
    /// server.
    #[builder(default)]
    #[serde(skip)]
    pub connect_timeout: Option<Duration>,

    /// The credential to use for authenticating connections in this pool.
    #[builder(default)]
    #[serde(skip)]
    pub credential: Option<Credential>,

    /// Extra information to append to the driver version in the metadata of the handshake with the
    /// server. This should be used by libraries wrapping the driver, e.g. ODMs.
    #[builder(default)]
    #[serde(skip)]
    pub driver_info: Option<DriverInfo>,

    /// Processes all events generated by the pool.
    #[derivative(Debug = "ignore", PartialEq = "ignore")]
    #[builder(default)]
    #[serde(skip)]
    pub event_handler: Option<Arc<dyn CmapEventHandler>>,

    /// Connections that have been ready for usage in the pool for longer than `max_idle_time` will
    /// not be used.
    ///
    /// The default is that connections will not be closed due to being idle.
    #[builder(default)]
    #[serde(rename = "maxIdleTimeMS")]
    #[serde(default)]
    #[serde(deserialize_with = "bson_util::deserialize_duration_from_u64_millis")]
    pub max_idle_time: Option<Duration>,

    /// The maximum number of connections that the pool can have at a given time. This includes
    /// connections which are currently checked out of the pool.
    ///
    /// The default is 100.
    #[builder(default)]
    pub max_pool_size: Option<u32>,

    /// The minimum number of connections that the pool can have at a given time. This includes
    /// connections which are currently checked out of the pool. If fewer than `min_pool_size`
    /// connections are in the pool, connections will be added to the pool in the background.
    ///
    /// The default is that no minimum is enforced
    #[builder(default)]
    pub min_pool_size: Option<u32>,

    /// The options specifying how a TLS connection should be configured. If `tls_options` is
    /// `None`, then TLS will not be used for the connections.
    ///
    /// The default is not to use TLS for connections.
    #[builder(default)]
    #[serde(skip)]
    pub tls_options: Option<TlsOptions>,

    /// Rather than wait indefinitely for a connection to become available, instead return an error
    /// after the given duration.
    ///
    /// The default is to block indefinitely until a connection becomes available.
    #[builder(default)]
    #[serde(rename = "waitQueueTimeoutMS")]
    #[serde(default)]
    #[serde(deserialize_with = "bson_util::deserialize_duration_from_u64_millis")]
    pub wait_queue_timeout: Option<Duration>,
}

impl ConnectionPoolOptions {
    pub(crate) fn from_client_options(options: &ClientOptions) -> Self {
        Self::builder()
            .app_name(options.app_name.clone())
            .connect_timeout(options.connect_timeout)
            .credential(options.credential.clone())
            .driver_info(options.driver_info.clone())
            .event_handler(options.cmap_event_handler.clone())
            .max_idle_time(options.max_idle_time)
            .max_pool_size(options.max_pool_size)
            .min_pool_size(options.min_pool_size)
            .tls_options(options.tls_options())
            .wait_queue_timeout(options.wait_queue_timeout)
            .build()
    }
}

/// Options used for constructing a `Connection`.
#[derive(Derivative)]
#[derivative(Debug)]
#[derive(Clone)]
pub(crate) struct ConnectionOptions {
    pub(crate) connect_timeout: Option<Duration>,

    pub(crate) tls_options: Option<TlsOptions>,

    #[derivative(Debug = "ignore")]
    pub(crate) event_handler: Option<Arc<dyn CmapEventHandler>>,
}

impl From<ConnectionPoolOptions> for ConnectionOptions {
    fn from(pool_options: ConnectionPoolOptions) -> Self {
        Self {
            connect_timeout: pool_options.connect_timeout,
            tls_options: pool_options.tls_options,
            event_handler: pool_options.event_handler,
        }
    }
}

#[derive(Clone, Debug, TypedBuilder)]
pub(crate) struct StreamOptions {
    pub(crate) address: StreamAddress,

    #[builder(default)]
    pub(crate) connect_timeout: Option<Duration>,

    #[builder(default)]
    pub(crate) tls_options: Option<TlsOptions>,
}