RUST-995 Add tokio-sync feature flag (#578)

Author: isabelatkinson

.evergreen/check-clippy.sh

@@ -10,3 +10,4 @@ cargo clippy --all-targets -p mongodb -- -D warnings
 cargo clippy --all-targets -p mongodb --features zstd-compression,snappy-compression,zlib-compression -- -D warnings
 cargo clippy --all-targets --no-default-features --features async-std-runtime -p mongodb -- -D warnings
 cargo clippy --all-targets --no-default-features --features sync -p mongodb -- -D warnings
+cargo clippy --all-targets --features tokio-sync -p mongodb -- -D warnings

.evergreen/check-rustdoc.sh

@@ -6,3 +6,4 @@ set -o errexit
 cargo +nightly rustdoc -- -D warnings --cfg docsrs
 cargo +nightly rustdoc --no-default-features --features async-std-runtime -- -D warnings --cfg docsrs
 cargo +nightly rustdoc --no-default-features --features sync -- -D warnings --cfg docsrs
+cargo +nightly rustdoc --features tokio-sync -- -D warnings --cfg docsrs

.evergreen/compile-only-tokio.sh

@@ -13,3 +13,4 @@ if [[ $RUST_VERSION == "nightly" ]]; then
 fi
 
 rustup run $RUST_VERSION cargo build --features $FEATURE_FLAGS
+rustup run $RUST_VERSION cargo build --features tokio-sync,$FEATURE_FLAGS

.evergreen/run-tokio-tests.sh

@@ -15,4 +15,10 @@ FEATURE_FLAGS="zstd-compression,snappy-compression,zlib-compression"
 echo "cargo test options: --features $FEATURE_FLAGS ${OPTIONS}"
 
 RUST_BACKTRACE=1 cargo test --features $FEATURE_FLAGS $OPTIONS | tee results.json
-cat results.json | cargo2junit > results.xml
+cat results.json | cargo2junit > async-tests.xml
+RUST_BACKTRACE=1 cargo test sync --features tokio-sync,$FEATURE_FLAGS $OPTIONS | tee sync-tests.json
+cat sync-tests.json | cargo2junit > sync-tests.xml
+RUST_BACKTRACE=1 cargo test --doc sync --features tokio-sync,$FEATURE_FLAGS $OPTIONS | tee sync-doc-tests.json
+cat sync-doc-tests.json | cargo2junit > sync-doc-tests.xml
+
+junit-report-merger results.xml async-tests.xml sync-tests.xml sync-doc-tests.xml

Cargo.toml

@@ -24,6 +24,7 @@ default = ["tokio-runtime"]
 tokio-runtime = ["tokio/macros", "tokio/net", "tokio/rt", "tokio/time", "serde_bytes"]
 async-std-runtime = ["async-std", "async-std/attributes", "async-std-resolver", "tokio-util/compat"]
 sync = ["async-std-runtime"]
+tokio-sync = ["tokio-runtime"]
 
 # Enable support for v0.4 of the chrono crate in the public API of the BSON library.
 bson-chrono-0_4 = ["bson/chrono-0_4"]

src/change_stream/mod.rs

@@ -45,7 +45,7 @@ use crate::{
 /// A `ChangeStream` can be iterated like any other [`Stream`]:
 ///
 /// ```
-/// # #[cfg(not(feature = "sync"))]
+/// # #[cfg(all(not(feature = "sync"), not(feature = "tokio-sync")))]
 /// # use futures::stream::StreamExt;
 /// # use mongodb::{Client, error::Result, bson::doc,
 /// # change_stream::event::ChangeStreamEvent};

src/client/auth/mod.rs

@@ -352,7 +352,7 @@ pub struct Credential {
 }
 
 impl Credential {
-    #[cfg(all(test, not(feature = "sync")))]
+    #[cfg(all(test, not(feature = "sync"), not(feature = "tokio-sync")))]
     pub(crate) fn into_document(mut self) -> Document {
         use crate::bson::Bson;
 

src/client/mod.rs

@@ -52,14 +52,14 @@ const DEFAULT_SERVER_SELECTION_TIMEOUT: Duration = Duration::from_secs(30);
 /// so it can safely be shared across threads or async tasks. For example:
 ///
 /// ```rust
-/// # #[cfg(not(feature = "sync"))]
+/// # #[cfg(all(not(feature = "sync"), not(feature = "tokio-sync")))]
 /// # use mongodb::{bson::Document, Client, error::Result};
 /// # #[cfg(feature = "async-std-runtime")]
 /// # use async_std::task;
 /// # #[cfg(feature = "tokio-runtime")]
 /// # use tokio::task;
 /// #
-/// # #[cfg(not(feature = "sync"))]
+/// # #[cfg(all(not(feature = "sync"), not(feature = "tokio-sync")))]
 /// # async fn start_workers() -> Result<()> {
 /// let client = Client::with_uri_str("mongodb://example.com").await?;
 ///
@@ -431,7 +431,7 @@ impl Client {
         }
     }
 
-    #[cfg(all(test, not(feature = "sync")))]
+    #[cfg(all(test, not(feature = "sync"), not(feature = "tokio-sync")))]
     pub(crate) async fn get_hosts(&self) -> Vec<String> {
         let servers = self.inner.topology.servers().await;
         servers

src/client/options/mod.rs

@@ -1,4 +1,4 @@
-#[cfg(all(test, not(feature = "sync")))]
+#[cfg(all(test, not(feature = "sync"), not(feature = "tokio-sync")))]
 mod test;
 
 mod resolver_config;
@@ -54,6 +54,9 @@ use crate::{
     srv::{OriginalSrvInfo, SrvResolver},
 };
 
+#[cfg(any(feature = "sync", feature = "tokio-sync"))]
+use crate::runtime;
+
 pub use resolver_config::ResolverConfig;
 
 const DEFAULT_PORT: u16 = 27017;
@@ -244,7 +247,7 @@ impl ServerAddress {
         })
     }
 
-    #[cfg(all(test, not(feature = "sync")))]
+    #[cfg(all(test, not(feature = "sync"), not(feature = "tokio-sync")))]
     pub(crate) fn into_document(self) -> Document {
         match self {
             Self::Tcp { host, port } => {
@@ -1038,17 +1041,17 @@ impl ClientOptions {
     ///
     /// Note: if the `sync` feature is enabled, then this method will be replaced with [the sync
     /// version](#method.parse-1).
-    #[cfg(not(feature = "sync"))]
+    #[cfg(all(not(feature = "sync"), not(feature = "tokio-sync")))]
     pub async fn parse(s: impl AsRef<str>) -> Result<Self> {
         Self::parse_uri(s, None).await
     }
 
     /// This method will be present if the `sync` feature is enabled. It's otherwise identical to
     /// [the async version](#method.parse)
-    #[cfg(any(feature = "sync", docsrs))]
-    #[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
+    #[cfg(any(feature = "sync", feature = "tokio-sync", docsrs))]
+    #[cfg_attr(docsrs, doc(cfg(any(feature = "sync", feature = "tokio-sync"))))]
     pub fn parse(s: impl AsRef<str>) -> Result<Self> {
-        crate::RUNTIME.block_on(Self::parse_uri(s.as_ref(), None))
+        runtime::block_on(Self::parse_uri(s.as_ref(), None))
     }
 
     /// Parses a MongoDB connection string into a `ClientOptions` struct.
@@ -1065,7 +1068,7 @@ impl ClientOptions {
     ///
     /// Note: if the `sync` feature is enabled, then this method will be replaced with [the sync
     /// version](#method.parse_with_resolver_config-1).
-    #[cfg(not(feature = "sync"))]
+    #[cfg(all(not(feature = "sync"), not(feature = "tokio-sync")))]
     pub async fn parse_with_resolver_config(
         uri: impl AsRef<str>,
         resolver_config: ResolverConfig,
@@ -1075,10 +1078,10 @@ impl ClientOptions {
 
     /// This method will be present if the `sync` feature is enabled. It's otherwise identical to
     /// [the async version](#method.parse_with_resolver_config)
-    #[cfg(any(feature = "sync", docsrs))]
-    #[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
+    #[cfg(any(feature = "sync", feature = "tokio-sync", docsrs))]
+    #[cfg_attr(docsrs, doc(cfg(any(feature = "sync", feature = "tokio-sync"))))]
     pub fn parse_with_resolver_config(uri: &str, resolver_config: ResolverConfig) -> Result<Self> {
-        crate::RUNTIME.block_on(Self::parse_uri(uri, Some(resolver_config)))
+        runtime::block_on(Self::parse_uri(uri, Some(resolver_config)))
     }
 
     /// Populate this `ClientOptions` from the given URI, optionally using the resolver config for
@@ -2015,7 +2018,7 @@ impl ClientOptionsParser {
     }
 }
 
-#[cfg(all(test, not(feature = "sync")))]
+#[cfg(all(test, not(feature = "sync"), not(feature = "tokio-sync")))]
 mod tests {
     use std::time::Duration;
 

src/client/session/mod.rs

@@ -18,10 +18,10 @@ use crate::{
     error::{ErrorKind, Result},
     operation::{AbortTransaction, CommitTransaction, Operation},
     options::{SessionOptions, TransactionOptions},
+    runtime,
     sdam::{ServerInfo, TransactionSupportStatus},
     selection_criteria::SelectionCriteria,
     Client,
-    RUNTIME,
 };
 pub use cluster_time::ClusterTime;
 pub(super) use pool::ServerSessionPool;
@@ -607,14 +607,14 @@ impl Drop for ClientSession {
                 snapshot_time: self.snapshot_time,
                 operation_time: self.operation_time,
             };
-            RUNTIME.execute(async move {
+            runtime::execute(async move {
                 let mut session: ClientSession = dropped_session.into();
                 let _result = session.abort_transaction().await;
             });
         } else {
             let client = self.client.clone();
             let server_session = self.server_session.clone();
-            RUNTIME.execute(async move {
+            runtime::execute(async move {
                 client.check_in_server_session(server_session).await;
             });
         }