RUST-289 Expose operations' write concerns

Author: patrickfreed

src/client/executor.rs

@@ -93,6 +93,10 @@ impl Client {
         op: &T,
         connection: &mut Connection,
     ) -> Result<T::O> {
+        if let Some(wc) = op.write_concern() {
+            wc.validate()?;
+        }
+
         let mut cmd = op.build(connection.stream_description()?)?;
         self.inner
             .topology

src/concern.rs renamed to src/concern/mod.rs

@@ -1,9 +1,12 @@
 //! Contains the types for read concerns and write concerns.
 
+#[cfg(test)]
+mod test;
+
 use std::time::Duration;
 
 use bson::doc;
-use serde::{Serialize, Serializer};
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use serde_with::skip_serializing_none;
 use typed_builder::TypedBuilder;
 
@@ -72,7 +75,7 @@ impl Serialize for ReadConcern {
 /// See the documentation [here](https://docs.mongodb.com/manual/reference/write-concern/) for more
 /// information about write concerns.
 #[skip_serializing_none]
-#[derive(Clone, Debug, Default, PartialEq, TypedBuilder, Serialize)]
+#[derive(Clone, Debug, Default, PartialEq, TypedBuilder, Serialize, Deserialize)]
 pub struct WriteConcern {
     /// Requests acknowledgement that the operation has propagated to a specific number or variety
     /// of servers.
@@ -88,6 +91,8 @@ pub struct WriteConcern {
     #[builder(default)]
     #[serde(rename = "wtimeout")]
     #[serde(serialize_with = "bson_util::serialize_duration_as_i64_millis")]
+    #[serde(deserialize_with = "bson_util::deserialize_duration_from_u64_millis")]
+    #[serde(default)]
     pub w_timeout: Option<Duration>,
 
     /// Requests acknowledgement that the operation has propagated to the on-disk journal.
@@ -121,6 +126,24 @@ impl Serialize for Acknowledgment {
     }
 }
 
+impl<'de> Deserialize<'de> for Acknowledgment {
+    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        #[derive(Deserialize)]
+        #[serde(untagged)]
+        enum IntOrString {
+            Int(i32),
+            String(String),
+        }
+        match IntOrString::deserialize(deserializer)? {
+            IntOrString::String(s) => Ok(s.into()),
+            IntOrString::Int(i) => Ok(i.into()),
+        }
+    }
+}
+
 impl From<i32> for Acknowledgment {
     fn from(i: i32) -> Self {
         Acknowledgment::Nodes(i)
@@ -152,6 +175,11 @@ impl Acknowledgment {
 }
 
 impl WriteConcern {
+    #[allow(dead_code)]
+    pub(crate) fn is_acknowledged(&self) -> bool {
+        self.w != Some(Acknowledgment::Nodes(0)) || self.journal == Some(true)
+    }
+
     /// Validates that the write concern. A write concern is invalid if the `w` field is 0
     /// and the `j` field is `true`.
     pub fn validate(&self) -> Result<()> {

src/concern/test.rs

@@ -0,0 +1,124 @@
+use std::time::Duration;
+
+use bson::{doc, Bson};
+
+use crate::{
+    error::ErrorKind,
+    options::{Acknowledgment, InsertOneOptions, WriteConcern},
+    test::{TestClient, LOCK},
+};
+
+#[test]
+fn write_concern_is_acknowledged() {
+    let w_1 = WriteConcern::builder()
+        .w(Acknowledgment::Nodes(1))
+        .journal(false)
+        .build();
+    assert!(w_1.is_acknowledged());
+
+    let w_majority = WriteConcern::builder()
+        .w(Acknowledgment::Majority)
+        .journal(false)
+        .build();
+    assert!(w_majority.is_acknowledged());
+
+    let w_0 = WriteConcern::builder()
+        .w(Acknowledgment::Nodes(0))
+        .journal(false)
+        .build();
+    assert!(!w_0.is_acknowledged());
+
+    let w_0 = WriteConcern::builder().w(Acknowledgment::Nodes(0)).build();
+    assert!(!w_0.is_acknowledged());
+
+    let empty = WriteConcern::builder().build();
+    assert!(empty.is_acknowledged());
+
+    let empty = WriteConcern::builder().journal(false).build();
+    assert!(empty.is_acknowledged());
+
+    let empty = WriteConcern::builder().journal(true).build();
+    assert!(empty.is_acknowledged());
+}
+
+#[test]
+fn write_concern_deserialize() {
+    let w_1 = doc! { "w": 1 };
+    let wc: WriteConcern = bson::from_bson(Bson::Document(w_1)).unwrap();
+    assert_eq!(
+        wc,
+        WriteConcern {
+            w: Acknowledgment::Nodes(1).into(),
+            w_timeout: None,
+            journal: None
+        }
+    );
+
+    let w_majority = doc! { "w": "majority" };
+    let wc: WriteConcern = bson::from_bson(Bson::Document(w_majority)).unwrap();
+    assert_eq!(
+        wc,
+        WriteConcern {
+            w: Acknowledgment::Majority.into(),
+            w_timeout: None,
+            journal: None
+        }
+    );
+
+    let w_timeout = doc! { "w": "majority", "wtimeout": 100 };
+    let wc: WriteConcern = bson::from_bson(Bson::Document(w_timeout)).unwrap();
+    assert_eq!(
+        wc,
+        WriteConcern {
+            w: Acknowledgment::Majority.into(),
+            w_timeout: Duration::from_millis(100).into(),
+            journal: None
+        }
+    );
+
+    let journal = doc! { "w": "majority", "j": true };
+    let wc: WriteConcern = bson::from_bson(Bson::Document(journal)).unwrap();
+    assert_eq!(
+        wc,
+        WriteConcern {
+            w: Acknowledgment::Majority.into(),
+            w_timeout: None,
+            journal: true.into()
+        }
+    );
+}
+
+#[cfg_attr(feature = "tokio-runtime", tokio::test)]
+#[cfg_attr(feature = "async-std-runtime", async_std::test)]
+#[function_name::named]
+async fn inconsistent_write_concern_rejected() {
+    let _guard = LOCK.run_concurrently().await;
+
+    let client = TestClient::new().await;
+    let db = client.database(function_name!());
+    let error = db
+        .run_command(
+            doc! {
+                "insert": function_name!(),
+                "documents": [ {} ],
+                "writeConcern": { "w": 0, "j": true }
+            },
+            None,
+        )
+        .await
+        .expect_err("insert should fail");
+    assert!(matches!(error.kind.as_ref(), ErrorKind::ArgumentError { .. }));
+
+    let coll = db.collection(function_name!());
+    let wc = WriteConcern {
+        w: Acknowledgment::Nodes(0).into(),
+        journal: true.into(),
+        w_timeout: None,
+    };
+    let options = InsertOneOptions::builder().write_concern(wc).build();
+    let error = coll
+        .insert_one(doc! {}, options)
+        .await
+        .expect_err("insert should fail");
+    assert!(matches!(error.kind.as_ref(), ErrorKind::ArgumentError { .. }));
+}

src/db/mod.rs

@@ -231,7 +231,7 @@ impl Database {
         command: Document,
         selection_criteria: impl Into<Option<SelectionCriteria>>,
     ) -> Result<Document> {
-        let operation = RunCommand::new(self.name().into(), command, selection_criteria.into());
+        let operation = RunCommand::new(self.name().into(), command, selection_criteria.into())?;
         self.client().execute_operation(&operation, None).await
     }
 

src/operation/aggregate/mod.rs

@@ -9,7 +9,7 @@ use crate::{
     cursor::CursorSpecification,
     error::Result,
     operation::{append_options, CursorBody, Operation, WriteConcernOnlyBody},
-    options::{AggregateOptions, SelectionCriteria},
+    options::{AggregateOptions, SelectionCriteria, WriteConcern},
     Namespace,
 };
 
@@ -87,6 +87,12 @@ impl Operation for Aggregate {
             .as_ref()
             .and_then(|opts| opts.selection_criteria.as_ref())
     }
+
+    fn write_concern(&self) -> Option<&WriteConcern> {
+        self.options
+            .as_ref()
+            .and_then(|opts| opts.write_concern.as_ref())
+    }
 }
 
 impl Aggregate {

src/operation/create/mod.rs

@@ -7,7 +7,7 @@ use crate::{
     cmap::{Command, CommandResponse, StreamDescription},
     error::Result,
     operation::{append_options, Operation, WriteConcernOnlyBody},
-    options::CreateCollectionOptions,
+    options::{CreateCollectionOptions, WriteConcern},
     Namespace,
 };
 
@@ -54,4 +54,10 @@ impl Operation for Create {
     fn handle_response(&self, response: CommandResponse) -> Result<Self::O> {
         response.body::<WriteConcernOnlyBody>()?.validate()
     }
+
+    fn write_concern(&self) -> Option<&WriteConcern> {
+        self.options
+            .as_ref()
+            .and_then(|opts| opts.write_concern.as_ref())
+    }
 }

src/operation/delete/mod.rs

@@ -9,7 +9,7 @@ use crate::{
     collation::Collation,
     error::{convert_bulk_errors, Result},
     operation::{append_options, Operation, WriteResponseBody},
-    options::DeleteOptions,
+    options::{DeleteOptions, WriteConcern},
     results::DeleteResult,
 };
 
@@ -88,4 +88,10 @@ impl Operation for Delete {
             deleted_count: body.n,
         })
     }
+
+    fn write_concern(&self) -> Option<&WriteConcern> {
+        self.options
+            .as_ref()
+            .and_then(|opts| opts.write_concern.as_ref())
+    }
 }

src/operation/drop_collection/mod.rs

@@ -7,7 +7,7 @@ use crate::{
     cmap::{Command, CommandResponse, StreamDescription},
     error::Result,
     operation::{append_options, Operation, WriteConcernOnlyBody},
-    options::DropCollectionOptions,
+    options::{DropCollectionOptions, WriteConcern},
     Namespace,
 };
 
@@ -55,4 +55,10 @@ impl Operation for DropCollection {
     fn handle_response(&self, response: CommandResponse) -> Result<Self::O> {
         response.body::<WriteConcernOnlyBody>()?.validate()
     }
+
+    fn write_concern(&self) -> Option<&WriteConcern> {
+        self.options
+            .as_ref()
+            .and_then(|opts| opts.write_concern.as_ref())
+    }
 }

src/operation/drop_database/mod.rs

@@ -7,7 +7,7 @@ use crate::{
     cmap::{Command, CommandResponse, StreamDescription},
     error::Result,
     operation::{append_options, Operation, WriteConcernOnlyBody},
-    options::DropDatabaseOptions,
+    options::{DropDatabaseOptions, WriteConcern},
 };
 
 #[derive(Debug)]
@@ -48,4 +48,10 @@ impl Operation for DropDatabase {
     fn handle_response(&self, response: CommandResponse) -> Result<Self::O> {
         response.body::<WriteConcernOnlyBody>()?.validate()
     }
+
+    fn write_concern(&self) -> Option<&WriteConcern> {
+        self.options
+            .as_ref()
+            .and_then(|opts| opts.write_concern.as_ref())
+    }
 }

src/operation/find_and_modify/mod.rs

@@ -20,6 +20,7 @@ use crate::{
     },
     error::{ErrorKind, Result},
     operation::{append_options, Operation},
+    options::WriteConcern,
 };
 
 pub(crate) struct FindAndModify {
@@ -88,6 +89,7 @@ impl Operation for FindAndModify {
             body,
         ))
     }
+
     fn handle_response(&self, response: CommandResponse) -> Result<Self::O> {
         let body: ResponseBody = response.body()?;
         match body.value {
@@ -103,6 +105,10 @@ impl Operation for FindAndModify {
             .into()),
         }
     }
+
+    fn write_concern(&self) -> Option<&WriteConcern> {
+        self.options.write_concern.as_ref()
+    }
 }
 
 #[derive(Debug, Deserialize)]