RUST-802 Support Unix Domain Sockets

Author: PureWhiteWu

src/client/executor.rs

@@ -99,7 +99,7 @@ impl Client {
         op: T,
         session: impl Into<Option<&mut ClientSession>>,
     ) -> Result<ExecutionDetails<T>> {
-        Box::pin(async {
+        async {
             // TODO RUST-9: allow unacknowledged write concerns
             if !op.is_acknowledged() {
                 return Err(ErrorKind::InvalidArgument {
@@ -130,7 +130,7 @@ impl Client {
                 }
             }
             self.execute_operation_with_retry(op, session).await
-        })
+        }
         .await
     }
 
@@ -141,7 +141,7 @@ impl Client {
     where
         Op: Operation<O = CursorSpecification>,
     {
-        Box::pin(async {
+        async {
             let mut details = self.execute_operation_with_details(op, None).await?;
             let pinned =
                 self.pin_connection_for_cursor(&details.output, &mut details.connection)?;
@@ -151,7 +151,7 @@ impl Client {
                 details.implicit_session,
                 pinned,
             ))
-        })
+        }
         .await
     }
 
@@ -212,7 +212,7 @@ impl Client {
     where
         T: DeserializeOwned + Unpin + Send + Sync,
     {
-        Box::pin(async {
+        async {
             let pipeline: Vec<_> = pipeline.into_iter().collect();
             let args = WatchArgs {
                 pipeline,
@@ -235,7 +235,7 @@ impl Client {
             let cursor = Cursor::new(self.clone(), cursor_spec, details.implicit_session, pinned);
 
             Ok(ChangeStream::new(cursor, args, cs_data))
-        })
+        }
         .await
     }
 
@@ -250,7 +250,7 @@ impl Client {
     where
         T: DeserializeOwned + Unpin + Send + Sync,
     {
-        Box::pin(async {
+        async {
             let pipeline: Vec<_> = pipeline.into_iter().collect();
             let args = WatchArgs {
                 pipeline,
@@ -268,7 +268,7 @@ impl Client {
             let cursor = SessionCursor::new(self.clone(), cursor_spec, pinned);
 
             Ok(SessionChangeStream::new(cursor, args, cs_data))
-        })
+        }
         .await
     }
 

src/client/mod.rs

@@ -485,7 +485,7 @@ impl Client {
         &self,
         criteria: Option<&SelectionCriteria>,
     ) -> Result<ServerAddress> {
-        let server = self.select_server(criteria, "Test select server").await?;
+            let server = self.select_server(criteria, "Test select server").await?;
         Ok(server.address.clone())
     }
 

src/client/options/mod.rs

@@ -96,9 +96,6 @@ lazy_static! {
 }
 
 /// An enum representing the address of a MongoDB server.
-///
-/// Currently this just supports addresses that can be connected to over TCP, but alternative
-/// address types may be supported in the future (e.g. Unix Domain Socket paths).
 #[derive(Clone, Debug, Eq, Serialize)]
 #[non_exhaustive]
 pub enum ServerAddress {
@@ -112,6 +109,12 @@ pub enum ServerAddress {
         /// The default is 27017.
         port: Option<u16>,
     },
+    /// A Unix Domain Socket path.
+    #[cfg(unix)]
+    Unix {
+        /// The path to the Unix Domain Socket.
+        path: PathBuf,
+    },
 }
 
 impl<'de> Deserialize<'de> for ServerAddress {
@@ -144,6 +147,10 @@ impl PartialEq for ServerAddress {
                     port: other_port,
                 },
             ) => host == other_host && port.unwrap_or(27017) == other_port.unwrap_or(27017),
+            #[cfg(unix)]
+            (Self::Unix { path }, Self::Unix { path: other_path }) => path == other_path,
+            #[cfg(unix)]
+            _ => false,
         }
     }
 }
@@ -158,6 +165,8 @@ impl Hash for ServerAddress {
                 host.hash(state);
                 port.unwrap_or(27017).hash(state);
             }
+            #[cfg(unix)]
+            Self::Unix { path } => path.hash(state),
         }
     }
 }
@@ -173,6 +182,15 @@ impl ServerAddress {
     /// Parses an address string into a `ServerAddress`.
     pub fn parse(address: impl AsRef<str>) -> Result<Self> {
         let address = address.as_ref();
+        // checks if the address is a unix domain socket
+        #[cfg(unix)]
+        {
+            if address.starts_with('/') {
+                return Ok(ServerAddress::Unix {
+                    path: PathBuf::from(address),
+                });
+            }
+        }
         let mut parts = address.split(':');
         let hostname = match parts.next() {
             Some(part) => {
@@ -243,18 +261,28 @@ impl ServerAddress {
                     "port": port.map(|i| Bson::Int32(i.into())).unwrap_or(Bson::Null)
                 }
             }
+            #[cfg(unix)]
+            Self::Unix { path } => {
+                doc! {
+                    "path": path.to_str().unwrap(),
+                }
+            }
         }
     }
 
     pub(crate) fn host(&self) -> &str {
         match self {
             Self::Tcp { host, .. } => host.as_str(),
+            #[cfg(unix)]
+            Self::Unix { path } => path.to_str().unwrap(),
         }
     }
 
     pub(crate) fn port(&self) -> Option<u16> {
         match self {
             Self::Tcp { port, .. } => *port,
+            #[cfg(unix)]
+            Self::Unix { .. } => None,
         }
     }
 }
@@ -265,6 +293,8 @@ impl fmt::Display for ServerAddress {
             Self::Tcp { host, port } => {
                 write!(fmt, "{}:{}", host, port.unwrap_or(DEFAULT_PORT))
             }
+            #[cfg(unix)]
+            Self::Unix { path } => write!(fmt, "{}", path.display()),
         }
     }
 }
@@ -1592,16 +1622,26 @@ impl ConnectionString {
                 }
                 .into());
             }
-            // Unwrap safety: the `len` check above guarantees this can't fail.
-            let ServerAddress::Tcp { host, port } = host_list.into_iter().next().unwrap();
 
-            if port.is_some() {
-                return Err(ErrorKind::InvalidArgument {
-                    message: "a port cannot be specified with 'mongodb+srv'".into(),
+            // Unwrap safety: the `len` check above guarantees this can't fail.
+            match host_list.into_iter().next().unwrap() {
+                ServerAddress::Tcp { host, port } => {
+                    if port.is_some() {
+                        return Err(ErrorKind::InvalidArgument {
+                            message: "a port cannot be specified with 'mongodb+srv'".into(),
+                        }
+                        .into());
+                    }
+                    HostInfo::DnsRecord(host)
+                }
+                #[cfg(unix)]
+                ServerAddress::Unix { .. } => {
+                    return Err(ErrorKind::InvalidArgument {
+                        message: "unix sockets cannot be used with 'mongodb+srv'".into(),
+                    }
+                    .into());
                 }
-                .into());
             }
-            HostInfo::DnsRecord(host)
         } else {
             HostInfo::HostIdentifiers(host_list)
         };
@@ -2299,18 +2339,39 @@ mod tests {
     #[test]
     fn test_parse_address_with_from_str() {
         let x = "localhost:27017".parse::<ServerAddress>().unwrap();
-        let ServerAddress::Tcp { host, port } = x;
-        assert_eq!(host, "localhost");
-        assert_eq!(port, Some(27017));
+        match x {
+            ServerAddress::Tcp { host, port } => {
+                assert_eq!(host, "localhost");
+                assert_eq!(port, Some(27017));
+            }
+            #[cfg(unix)]
+            _ => panic!("expected ServerAddress::Tcp"),
+        }
 
         // Port defaults to 27017 (so this doesn't fail)
         let x = "localhost".parse::<ServerAddress>().unwrap();
-        let ServerAddress::Tcp { host, port } = x;
-        assert_eq!(host, "localhost");
-        assert_eq!(port, None);
+        match x {
+            ServerAddress::Tcp { host, port } => {
+                assert_eq!(host, "localhost");
+                assert_eq!(port, Some(27017));
+            }
+            #[cfg(unix)]
+            _ => panic!("expected ServerAddress::Tcp"),
+        }
 
         let x = "localhost:not a number".parse::<ServerAddress>();
         assert!(x.is_err());
+
+        #[cfg(unix)]
+        {
+            let x = "/path/to/socket.sock".parse::<ServerAddress>().unwrap();
+            match x {
+                ServerAddress::Unix { path } => {
+                    assert_eq!(path.to_str().unwrap(), "/path/to/socket.sock");
+                }
+                _ => panic!("expected ServerAddress::Unix"),
+            }
+        }
     }
 
     #[cfg_attr(feature = "tokio-runtime", tokio::test)]