RUST-228 Document client options (#96)

Author: saghm

src/client/mod.rs

@@ -77,6 +77,8 @@ struct ClientInner {
 impl Client {
     /// Creates a new `Client` connected to the cluster specified by `uri`. `uri` must be a valid
     /// MongoDB connection string.
+    ///
+    /// See the documentation on ClientOptions::parse for more details.
     pub fn with_uri_str(uri: &str) -> Result<Self> {
         let options = ClientOptions::parse(uri)?;
 

src/client/options/mod.rs

@@ -114,10 +114,10 @@ fn document_from_client_options(mut options: ClientOptions) -> Document {
 
             doc.insert("readpreferencetags", tags);
         }
-    }
 
-    if let Some(i) = options.max_staleness.take() {
-        doc.insert("maxstalenessseconds", i.as_secs() as i64);
+        if let Some(i) = max_staleness {
+            doc.insert("maxstalenessseconds", i.as_secs() as i64);
+        }
     }
 
     if let Some(b) = options.retry_reads.take() {

src/client/options/test.rs

@@ -70,6 +70,7 @@ impl Connection {
         id: u32,
         address: StreamAddress,
         generation: u32,
+        connect_timeout: Option<Duration>,
         tls_options: Option<TlsOptions>,
         handler: Option<Arc<dyn CmapEventHandler>>,
     ) -> Result<Self> {
@@ -79,7 +80,7 @@ impl Connection {
             pool: None,
             stream_description: None,
             ready_and_available_time: None,
-            stream: Stream::connect(address.clone(), tls_options)?,
+            stream: Stream::connect(address.clone(), connect_timeout, tls_options)?,
             address,
             handler,
         };

src/cmap/conn/mod.rs

@@ -1,7 +1,8 @@
 use std::{
     io::{self, Read, Write},
-    net::TcpStream,
+    net::{TcpStream, ToSocketAddrs},
     sync::Arc,
+    time::Duration,
 };
 
 use derivative::Derivative;
@@ -12,6 +13,8 @@ use crate::{
     options::{StreamAddress, TlsOptions},
 };
 
+const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
+
 /// Stream encapsulates the different socket types that can be used and adds a thin wrapper for I/O.
 #[derive(Derivative)]
 #[derivative(Debug)]
@@ -33,8 +36,22 @@ pub(super) enum Stream {
 
 impl Stream {
     /// Creates a new stream connected to `address`.
-    pub(super) fn connect(host: StreamAddress, tls_options: Option<TlsOptions>) -> Result<Self> {
-        let inner = TcpStream::connect(host.to_string())?;
+    pub(super) fn connect(
+        host: StreamAddress,
+        connect_timeout: Option<Duration>,
+        tls_options: Option<TlsOptions>,
+    ) -> Result<Self> {
+        let timeout = connect_timeout.unwrap_or(DEFAULT_CONNECT_TIMEOUT);
+
+        // The URI options spec requires that the default is 10 seconds, but that 0 should indicate
+        // no timeout.
+        let inner = if timeout == Duration::from_secs(0) {
+            TcpStream::connect(&host)?
+        } else {
+            let socket_addrs: Vec<_> = host.to_socket_addrs()?.collect();
+
+            TcpStream::connect_timeout(&socket_addrs[0], timeout)?
+        };
         inner.set_nodelay(true)?;
 
         match tls_options {

src/cmap/conn/stream.rs

@@ -64,9 +64,30 @@ pub(crate) struct ConnectionPoolInner {
     /// The address the pool's connections will connect to.
     address: StreamAddress,
 
-    /// If a checkout operation takes longer than `wait_queue_timeout`, the pool will return an
-    /// error. If `wait_queue_timeout` is `None`, then the checkout operation will not time out.
-    wait_queue_timeout: Option<Duration>,
+    /// The set of available connections in the pool. Because the CMAP spec requires that
+    /// connections are checked out in a FIFO manner, connections are pushed/popped from the back
+    /// of the Vec.
+    connections: Arc<RwLock<Vec<Connection>>>,
+
+    /// The connect timeout passed to each underlying TcpStream when attemtping to connect to the
+    /// server.
+    connect_timeout: Option<Duration>,
+
+    /// The credential to use for authenticating connections in this pool.
+    credential: Option<Credential>,
+
+    /// Contains the logic for "establishing" a connection. This includes handshaking and
+    /// authenticating a connection when it's first created.
+    establisher: ConnectionEstablisher,
+
+    /// The event handler specified by the user to process CMAP events.
+    #[derivative(Debug = "ignore")]
+    event_handler: Option<Arc<dyn CmapEventHandler>>,
+
+    /// The current generation of the pool. The generation is incremented whenever the pool is
+    /// cleared. Connections belonging to a previous generation are considered stale and will be
+    /// closed when checked back in or when popped off of the set of available connections.
+    generation: AtomicU32,
 
     /// Connections that have been ready for usage in the pool for longer than `max_idle_time` will
     /// be closed either by the background thread or when popped off of the set of available
@@ -84,75 +105,66 @@ pub(crate) struct ConnectionPoolInner {
     /// them to the pool.
     min_pool_size: Option<u32>,
 
+    /// The ID of the next connection created by the pool.
+    next_connection_id: AtomicU32,
+
+    /// If a checkout operation takes longer than `wait_queue_timeout`, the pool will return an
+    /// error. If `wait_queue_timeout` is `None`, then the checkout operation will not time out.
+    wait_queue_timeout: Option<Duration>,
+
     /// The TLS options to use for the connections. If `tls_options` is None, then TLS will not be
     /// used to connect to the server.
     tls_options: Option<TlsOptions>,
 
-    /// The credential to use for authenticating connections in this pool.
-    credential: Option<Credential>,
-
-    /// The current generation of the pool. The generation is incremented whenever the pool is
-    /// cleared. Connections belonging to a previous generation are considered stale and will be
-    /// closed when checked back in or when popped off of the set of available connections.
-    generation: AtomicU32,
-
     /// The total number of connections currently in the pool. This includes connections which are
     /// currently checked out of the pool.
     total_connection_count: AtomicU32,
 
-    /// The ID of the next connection created by the pool.
-    next_connection_id: AtomicU32,
-
     /// Connections are checked out by concurrent threads on a first-come, first-server basis. This
     /// is enforced by threads entering the wait queue when they first try to check out a
     /// connection and then blocking until they are at the front of the queue.
     wait_queue: WaitQueue,
-
-    /// The set of available connections in the pool. Because the CMAP spec requires that
-    /// connections are checked out in a FIFO manner, connections are pushed/popped from the back
-    /// of the Vec.
-    connections: Arc<RwLock<Vec<Connection>>>,
-
-    /// Contains the logic for "establishing" a connection. This includes handshaking and
-    /// authenticating a connection when it's first created.
-    establisher: ConnectionEstablisher,
-
-    /// The event handler specified by the user to process CMAP events.
-    #[derivative(Debug = "ignore")]
-    event_handler: Option<Arc<dyn CmapEventHandler>>,
 }
 
 impl ConnectionPool {
     pub(crate) fn new(address: StreamAddress, mut options: Option<ConnectionPoolOptions>) -> Self {
         // Get the individual options from `options`.
-        let max_idle_time = options.as_ref().and_then(|opts| opts.max_idle_time);
-        let wait_queue_timeout = options.as_ref().and_then(|opts| opts.wait_queue_timeout);
+        let connect_timeout = options.as_ref().and_then(|opts| opts.connect_timeout);
+        let credential = options.as_mut().and_then(|opts| opts.credential.clone());
+        let establisher = ConnectionEstablisher::new(options.as_ref());
+        let event_handler = options.as_mut().and_then(|opts| opts.event_handler.take());
+
+        // The CMAP spec indicates that a max idle time of zero means that connections should not be
+        // closed due to idleness.
+        let mut max_idle_time = options.as_ref().and_then(|opts| opts.max_idle_time);
+        if max_idle_time == Some(Duration::from_millis(0)) {
+            max_idle_time = None;
+        }
+
         let max_pool_size = options
             .as_ref()
             .and_then(|opts| opts.max_pool_size)
             .unwrap_or(DEFAULT_MAX_POOL_SIZE);
         let min_pool_size = options.as_ref().and_then(|opts| opts.min_pool_size);
         let tls_options = options.as_mut().and_then(|opts| opts.tls_options.take());
-        let event_handler = options.as_mut().and_then(|opts| opts.event_handler.take());
-
-        let credential = options.as_mut().and_then(|opts| opts.credential.clone());
-        let establisher = ConnectionEstablisher::new(options.as_ref());
+        let wait_queue_timeout = options.as_ref().and_then(|opts| opts.wait_queue_timeout);
 
         let inner = ConnectionPoolInner {
             address: address.clone(),
-            wait_queue_timeout,
+            connect_timeout,
+            credential,
+            establisher,
+            event_handler,
+            generation: AtomicU32::new(0),
             max_idle_time,
             max_pool_size,
             min_pool_size,
+            next_connection_id: AtomicU32::new(1),
             tls_options,
-            credential,
-            generation: AtomicU32::new(0),
             total_connection_count: AtomicU32::new(0),
-            next_connection_id: AtomicU32::new(1),
             wait_queue: WaitQueue::new(address.clone(), wait_queue_timeout),
+            wait_queue_timeout,
             connections: Default::default(),
-            establisher,
-            event_handler,
         };
 
         let pool = Self {
@@ -329,6 +341,7 @@ impl ConnectionPool {
             self.inner.next_connection_id.fetch_add(1, Ordering::SeqCst),
             self.inner.address.clone(),
             self.inner.generation.load(Ordering::SeqCst),
+            self.inner.connect_timeout,
             self.inner.tls_options.clone(),
             self.inner.event_handler.clone(),
         )?;

src/cmap/mod.rs

@@ -21,6 +21,33 @@ pub struct ConnectionPoolOptions {
     #[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>,
+
+    /// 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 = "self::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.
     ///
@@ -36,15 +63,13 @@ pub struct ConnectionPoolOptions {
     #[builder(default)]
     pub min_pool_size: Option<u32>,
 
-    /// Connections that have been ready for usage in the pool for longer than `max_idle_time` will
-    /// not be used.
+    /// 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 that connections will not be closed due to being idle.
+    /// The default is not to use TLS for connections.
     #[builder(default)]
-    #[serde(rename = "maxIdleTimeMS")]
-    #[serde(default)]
-    #[serde(deserialize_with = "self::deserialize_duration_from_u64_millis")]
-    pub max_idle_time: Option<Duration>,
+    #[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.
@@ -55,24 +80,6 @@ pub struct ConnectionPoolOptions {
     #[serde(default)]
     #[serde(deserialize_with = "self::deserialize_duration_from_u64_millis")]
     pub wait_queue_timeout: Option<Duration>,
-
-    /// 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>,
-
-    #[derivative(Debug = "ignore", PartialEq = "ignore")]
-    #[builder(default)]
-    #[serde(skip)]
-    pub event_handler: Option<Arc<dyn CmapEventHandler>>,
-
-    /// The credential to use for authenticating connections in this pool.
-    #[builder(default)]
-    #[serde(skip)]
-    pub credential: Option<Credential>,
 }
 
 fn deserialize_duration_from_u64_millis<'de, D>(
@@ -88,13 +95,15 @@ where
 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())
+            .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)
-            .max_idle_time(options.max_idle_time)
-            .wait_queue_timeout(options.wait_queue_timeout)
             .tls_options(options.tls_options())
-            .event_handler(options.cmap_event_handler.clone())
+            .wait_queue_timeout(options.wait_queue_timeout)
             .build()
     }
 }

src/cmap/options.rs

@@ -62,6 +62,9 @@ impl std::ops::Deref for Error {
 
 #[derive(Debug, Error)]
 pub enum ErrorKind {
+    #[error(display = "{}", _0)]
+    AddrParse(#[error(source)] std::net::AddrParseError),
+
     #[error(
         display = "An invalid argument was provided to a database operation: {}",
         message

src/error.rs

@@ -177,20 +177,35 @@ pub struct ConnectionCheckedInEvent {
 /// by the driver.
 ///
 /// ```rust
-/// # use mongodb::{Client, event::cmap::{CmapEventHandler, ConnectionCheckoutFailedEvent}};
-///
-/// struct FailedCheckoutLogger {}
+/// # use std::sync::Arc;
+/// #
+/// # use mongodb::{
+/// #     error::Result,
+/// #     event::cmap::{
+/// #         CmapEventHandler,
+/// #         ConnectionCheckoutFailedEvent
+/// #     },
+/// #     options::ClientOptions,
+/// #     Client,
+/// # };
+/// #
+/// struct FailedCheckoutLogger;
 ///
 /// impl CmapEventHandler for FailedCheckoutLogger {
 ///     fn handle_connection_checkout_failed_event(&self, event: ConnectionCheckoutFailedEvent) {
 ///         eprintln!("Failed connection checkout: {:?}", event);
 ///     }
 /// }
 ///
-/// # fn main() {
-/// // TODO: Construct client with event handler by using `ClientOptions`.
+/// # fn do_stuff() -> Result<()> {
+/// let handler: Arc<dyn CmapEventHandler> = Arc::new(FailedCheckoutLogger);
+/// let options = ClientOptions::builder()
+///                   .cmap_event_handler(handler)
+///                   .build();
+/// let client = Client::with_options(options)?;
 ///
-/// // Do things with the client, and failed checkout events will be logged to stderr.
+/// // Do things with the client, and failed connection pool checkouts will be logged to stderr.
+/// # Ok(())
 /// # }
 /// ```
 pub trait CmapEventHandler: Send + Sync {