RUST-1876 Require T: Send + Sync for Collection<T>

src/action.rs

@@ -202,7 +202,7 @@ pub(crate) struct CollRef<'a> {
 }
 
 impl<'a> CollRef<'a> {
-    fn new<T>(coll: &'a Collection<T>) -> Self {
+    fn new<T: Send + Sync>(coll: &'a Collection<T>) -> Self {
         Self {
             inner: coll.clone_with_type(),
             _ref: PhantomData,

src/action/aggregate.rs

@@ -36,7 +36,10 @@ impl Database {
     }
 }
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     /// Runs an aggregation operation.
     ///
     /// See the documentation [here](https://www.mongodb.com/docs/manual/aggregation/) for more
@@ -69,7 +72,10 @@ impl crate::sync::Database {
 }
 
 #[cfg(feature = "sync")]
-impl<T> crate::sync::Collection<T> {
+impl<T> crate::sync::Collection<T>
+where
+    T: Send + Sync,
+{
     /// Runs an aggregation operation.
     ///
     /// See the documentation [here](https://www.mongodb.com/docs/manual/aggregation/) for more

src/action/count.rs

@@ -9,7 +9,10 @@ use crate::{
 
 use super::{action_impl, option_setters, CollRef};
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     /// Estimates the number of documents in the collection using collection metadata.
     ///
     /// Due to an oversight in versions 5.0.0 - 5.0.7 of MongoDB, the `count` server command,
@@ -46,7 +49,10 @@ impl<T> Collection<T> {
 }
 
 #[cfg(feature = "sync")]
-impl<T> crate::sync::Collection<T> {
+impl<T> crate::sync::Collection<T>
+where
+    T: Send + Sync,
+{
     /// Estimates the number of documents in the collection using collection metadata.
     ///
     /// Due to an oversight in versions 5.0.0 - 5.0.7 of MongoDB, the `count` server command,

src/action/create_index.rs

@@ -15,7 +15,10 @@ use crate::{
 
 use super::{action_impl, option_setters, CollRef, Multiple, Single};
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     /// Creates the given index on this collection.
     ///
     /// `await` will return `Result<CreateIndexResult>`.
@@ -46,8 +49,11 @@ impl<T> Collection<T> {
     }
 }
 
-#[cfg(any(feature = "sync", feature = "tokio-sync"))]
-impl<T> crate::sync::Collection<T> {
+#[cfg(feature = "sync")]
+impl<T> crate::sync::Collection<T>
+where
+    T: Send + Sync,
+{
     /// Creates the given index on this collection.
     ///
     /// [`run`](CreateIndex::run) will return `Result<CreateIndexResult>`.

src/action/delete.rs

@@ -13,7 +13,10 @@ use crate::{
 
 use super::{action_impl, option_setters, CollRef};
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     /// Deletes up to one document found matching `query`.
     ///
     /// This operation will retry once upon failure if the connection and encountered error support
@@ -46,8 +49,11 @@ impl<T> Collection<T> {
     }
 }
 
-#[cfg(any(feature = "sync", feature = "tokio-sync"))]
-impl<T> crate::sync::Collection<T> {
+#[cfg(feature = "sync")]
+impl<T> crate::sync::Collection<T>
+where
+    T: Send + Sync,
+{
     /// Deletes up to one document found matching `query`.
     ///
     /// This operation will retry once upon failure if the connection and encountered error support

src/action/distinct.rs

@@ -15,7 +15,10 @@ use crate::{
 
 use super::{action_impl, option_setters, CollRef};
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     /// Finds the distinct values of the field specified by `field_name` across the collection.
     ///
     /// `await` will return `Result<Vec<Bson>>`.
@@ -30,8 +33,11 @@ impl<T> Collection<T> {
     }
 }
 
-#[cfg(any(feature = "sync", feature = "tokio-sync"))]
-impl<T> crate::sync::Collection<T> {
+#[cfg(feature = "sync")]
+impl<T> crate::sync::Collection<T>
+where
+    T: Send + Sync,
+{
     /// Finds the distinct values of the field specified by `field_name` across the collection.
     ///
     /// [`run`](Distinct::run) will return `Result<Vec<Bson>>`.

src/action/drop.rs

@@ -69,7 +69,10 @@ action_impl! {
     }
 }
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     /// Drops the collection, deleting all data and indexes stored in it.
     ///
     /// `await` will return `Result<()>`.
@@ -83,7 +86,10 @@ impl<T> Collection<T> {
 }
 
 #[cfg(feature = "sync")]
-impl<T> crate::sync::Collection<T> {
+impl<T> crate::sync::Collection<T>
+where
+    T: Send + Sync,
+{
     /// Drops the collection, deleting all data and indexes stored in it.
     ///
     /// [`run`](DropCollection::run) will return `Result<()>`.

src/action/drop_index.rs

@@ -12,7 +12,10 @@ use crate::{
     Collection,
 };
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     /// Drops the index specified by `name` from this collection.
     ///
     /// `await` will return `Result<()>`.
@@ -38,8 +41,11 @@ impl<T> Collection<T> {
     }
 }
 
-#[cfg(any(feature = "sync", feature = "tokio-sync"))]
-impl<T> crate::sync::Collection<T> {
+#[cfg(feature = "sync")]
+impl<T> crate::sync::Collection<T>
+where
+    T: Send + Sync,
+{
     /// Drops the index specified by `name` from this collection.
     ///
     /// [`run`](DropIndex::run) will return `Result<()>`.

src/action/list_indexes.rs

@@ -24,7 +24,10 @@ use super::{
     ListSpecifications,
 };
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     /// Lists all indexes on this collection.
     ///
     /// `await` will return `Result<Cursor<IndexModel>>` (or `Result<SessionCursor<IndexModel>>` if
@@ -51,8 +54,11 @@ impl<T> Collection<T> {
     }
 }
 
-#[cfg(any(feature = "sync", feature = "tokio-sync"))]
-impl<T> crate::sync::Collection<T> {
+#[cfg(feature = "sync")]
+impl<T> crate::sync::Collection<T>
+where
+    T: Send + Sync,
+{
     /// Lists all indexes on this collection.
     ///
     /// [`run`](ListIndexes::run) will return `Result<Cursor<IndexModel>>` (or

src/action/update.rs

@@ -13,7 +13,10 @@ use crate::{
 
 use super::{action_impl, option_setters, CollRef};
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     /// Updates all documents matching `query` in the collection.
     ///
     /// Both `Document` and `Vec<Document>` implement `Into<UpdateModifications>`, so either can be
@@ -58,8 +61,11 @@ impl<T> Collection<T> {
     }
 }
 
-#[cfg(any(feature = "sync", feature = "tokio-sync"))]
-impl<T> crate::sync::Collection<T> {
+#[cfg(feature = "sync")]
+impl<T> crate::sync::Collection<T>
+where
+    T: Send + Sync,
+{
     /// Updates all documents matching `query` in the collection.
     ///
     /// Both `Document` and `Vec<Document>` implement `Into<UpdateModifications>`, so either can be

src/action/watch.rs

@@ -79,7 +79,10 @@ impl Database {
     }
 }
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     /// Starts a new [`ChangeStream`](change_stream/struct.ChangeStream.html) that receives events
     /// for all changes in this collection. A
     /// [`ChangeStream`](change_stream/struct.ChangeStream.html) cannot be started on system
@@ -133,7 +136,10 @@ impl crate::sync::Database {
 }
 
 #[cfg(feature = "sync")]
-impl<T> crate::sync::Collection<T> {
+impl<T> crate::sync::Collection<T>
+where
+    T: Send + Sync,
+{
     /// Starts a new [`ChangeStream`](change_stream/struct.ChangeStream.html) that receives events
     /// for all changes in this collection. A
     /// [`ChangeStream`](change_stream/struct.ChangeStream.html) cannot be started on system

src/coll.rs

@@ -82,14 +82,20 @@ use crate::{
 /// # }
 /// ```
 #[derive(Debug)]
-pub struct Collection<T> {
+pub struct Collection<T>
+where
+    T: Send + Sync,
+{
     inner: Arc<CollectionInner>,
     _phantom: std::marker::PhantomData<fn() -> T>,
 }
 
 // Because derive is too conservative, derive only implements Clone if T is Clone.
 // Collection<T> does not actually store any value of type T (so T does not need to be clone).
-impl<T> Clone for Collection<T> {
+impl<T> Clone for Collection<T>
+where
+    T: Send + Sync,
+{
     fn clone(&self) -> Self {
         Self {
             inner: self.inner.clone(),
@@ -109,7 +115,10 @@ struct CollectionInner {
     human_readable_serialization: bool,
 }
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     pub(crate) fn new(db: Database, name: &str, options: Option<CollectionOptions>) -> Self {
         let options = options.unwrap_or_default();
         let selection_criteria = options
@@ -139,7 +148,7 @@ impl<T> Collection<T> {
     }
 
     /// Gets a clone of the `Collection` with a different type `U`.
-    pub fn clone_with_type<U>(&self) -> Collection<U> {
+    pub fn clone_with_type<U: Send + Sync>(&self) -> Collection<U> {
         Collection {
             inner: self.inner.clone(),
             _phantom: Default::default(),
@@ -294,7 +303,7 @@ where
 
 impl<T> Collection<T>
 where
-    T: DeserializeOwned,
+    T: DeserializeOwned + Send + Sync,
 {
     async fn find_one_and_delete_common(
         &self,
@@ -402,7 +411,7 @@ where
 
 impl<T> Collection<T>
 where
-    T: Serialize + DeserializeOwned,
+    T: Serialize + DeserializeOwned + Send + Sync,
 {
     async fn find_one_and_replace_common(
         &self,
@@ -463,7 +472,7 @@ where
 
 impl<T> Collection<T>
 where
-    T: Serialize,
+    T: Serialize + Send + Sync,
 {
     #[allow(clippy::needless_option_as_deref)]
     async fn insert_many_common(

src/coll/action/drop.rs

@@ -24,7 +24,10 @@ action_impl! {
 }
 
 #[cfg(feature = "in-use-encryption-unstable")]
-impl<T> crate::Collection<T> {
+impl<T> crate::Collection<T>
+where
+    T: Send + Sync,
+{
     #[allow(clippy::needless_option_as_deref)]
     async fn drop_aux_collections(
         &self,

src/db.rs

@@ -117,7 +117,7 @@ impl Database {
     ///
     /// This method does not send or receive anything across the wire to the database, so it can be
     /// used repeatedly without incurring any costs from I/O.
-    pub fn collection<T>(&self, name: &str) -> Collection<T> {
+    pub fn collection<T: Send + Sync>(&self, name: &str) -> Collection<T> {
         Collection::new(self.clone(), name, None)
     }
 
@@ -130,7 +130,7 @@ impl Database {
     ///
     /// This method does not send or receive anything across the wire to the database, so it can be
     /// used repeatedly without incurring any costs from I/O.
-    pub fn collection_with_options<T>(
+    pub fn collection_with_options<T: Send + Sync>(
         &self,
         name: &str,
         options: CollectionOptions,

src/gridfs/upload.rs

@@ -174,7 +174,11 @@ impl GridFsBucket {
         Ok(())
     }
 
-    async fn create_index<T>(&self, coll: &Collection<T>, keys: Document) -> Result<()> {
+    async fn create_index<T: Send + Sync>(
+        &self,
+        coll: &Collection<T>,
+        keys: Document,
+    ) -> Result<()> {
         // listIndexes returns an error if the collection has not yet been created.
         // Ignore NamespaceExists errors if the collection has already been created.
         if let Err(error) = self

src/search_index.rs

@@ -12,7 +12,10 @@ use bson::doc;
 use serde::{Deserialize, Serialize};
 use typed_builder::TypedBuilder;
 
-impl<T> Collection<T> {
+impl<T> Collection<T>
+where
+    T: Send + Sync,
+{
     /// Convenience method for creating a single search index.
     pub async fn create_search_index(
         &self,