add sync/async enum

Author: isabelatkinson

src/client/executor.rs

@@ -55,6 +55,7 @@ use crate::{
         CommandErrorBody,
         CommitTransaction,
         Operation,
+        OperationResponse,
         Retryability,
     },
     options::{ChangeStreamOptions, SelectionCriteria},
@@ -766,14 +767,15 @@ impl Client {
                     }
                 };
 
-                match op
-                    .handle_response(
-                        response,
-                        connection.stream_description()?,
-                        session.as_deref_mut(),
-                    )
-                    .await
-                {
+                let response_result = match op.handle_response(
+                    response,
+                    connection.stream_description()?,
+                    session.as_deref_mut(),
+                ) {
+                    OperationResponse::Sync(result) => result,
+                    OperationResponse::Async(future) => future.await,
+                };
+                match response_result {
                     Ok(response) => Ok(response),
                     Err(mut err) => {
                         err.add_labels_and_update_pin(

src/coll.rs

@@ -1137,7 +1137,7 @@ where
 
 impl<T> Collection<T>
 where
-    T: Serialize + Send + Sync,
+    T: Serialize,
 {
     #[allow(clippy::needless_option_as_deref)]
     async fn insert_many_common(

src/operation.rs

@@ -83,6 +83,28 @@ const MAX_ENCRYPTED_WRITE_SIZE: usize = 2_097_152;
 // The amount of overhead bytes to account for when building a document sequence.
 const COMMAND_OVERHEAD_SIZE: usize = 16_000;
 
+pub(crate) enum OperationResponse<'a, O> {
+    Sync(Result<O>),
+    Async(BoxFuture<'a, Result<O>>),
+}
+
+impl<'a, O> OperationResponse<'a, O> {
+    fn get_sync_result(self) -> Result<O> {
+        match self {
+            Self::Sync(result) => result,
+            Self::Async(_) => unreachable!(),
+        }
+    }
+}
+
+macro_rules! sync {
+    ($result:block) => {
+        let result = || $result;
+        OperationResponse::Sync(result())
+    };
+}
+use sync;
+
 /// A trait modeling the behavior of a server side operation.
 ///
 /// No methods in this trait should have default behaviors to ensure that wrapper operations
@@ -111,7 +133,7 @@ pub(crate) trait Operation {
         response: RawCommandResponse,
         description: &'a StreamDescription,
         session: Option<&'a mut ClientSession>,
-    ) -> BoxFuture<'a, Result<Self::O>>;
+    ) -> OperationResponse<'a, Self::O>;
 
     /// Interpret an error encountered while sending the built command to the server, potentially
     /// recovering.
@@ -425,7 +447,7 @@ pub(crate) trait OperationWithDefaults {
         response: RawCommandResponse,
         description: &'a StreamDescription,
         session: Option<&'a mut ClientSession>,
-    ) -> BoxFuture<'a, Result<Self::O>>;
+    ) -> OperationResponse<'a, Self::O>;
 
     /// Interpret an error encountered while sending the built command to the server, potentially
     /// recovering.
@@ -492,7 +514,7 @@ impl<T: OperationWithDefaults> Operation for T {
         response: RawCommandResponse,
         description: &'a StreamDescription,
         session: Option<&'a mut ClientSession>,
-    ) -> BoxFuture<'a, Result<Self::O>> {
+    ) -> OperationResponse<'a, Self::O> {
         self.handle_response(response, description, session)
     }
     fn handle_error(&self, error: Error) -> Result<Self::O> {

src/operation/abort_transaction.rs

@@ -1,5 +1,4 @@
 use bson::Document;
-use futures_util::FutureExt;
 
 use crate::{
     bson::doc,
@@ -9,11 +8,10 @@ use crate::{
     operation::Retryability,
     options::WriteConcern,
     selection_criteria::SelectionCriteria,
-    BoxFuture,
     ClientSession,
 };
 
-use super::{OperationWithDefaults, WriteConcernOnlyBody};
+use super::{sync, OperationResponse, OperationWithDefaults, WriteConcernOnlyBody};
 
 pub(crate) struct AbortTransaction {
     write_concern: Option<WriteConcern>,
@@ -52,19 +50,28 @@ impl OperationWithDefaults for AbortTransaction {
         ))
     }
 
-    fn handle_response(
-        &self,
+    fn handle_response<'a>(
+        &'a self,
         response: RawCommandResponse,
-        _description: &StreamDescription,
-        _session: Option<&mut ClientSession>,
-    ) -> BoxFuture<'static, Result<Self::O>> {
-        async move {
+        _description: &'a StreamDescription,
+        _session: Option<&'a mut ClientSession>,
+    ) -> OperationResponse<'a, Self::O> {
+        sync! {{
             let response: WriteConcernOnlyBody = response.body()?;
             response.validate()
-        }
-        .boxed()
+        }}
     }
 
+    // handle_response!((
+    //     &self,
+    //     response: RawCommandResponse,
+    //     description: &StreamDescription,
+    //     session: Option<&mut ClientSession
+    // ) -> Result<Self::O> {
+    //     let response: WriteConcernOnlyBody = response.body()?;
+    //     response.validate()
+    // });
+
     fn selection_criteria(&self) -> Option<&SelectionCriteria> {
         match &self.pinned {
             Some(TransactionPin::Mongos(s)) => Some(s),

src/operation/aggregate.rs

@@ -8,21 +8,21 @@ use crate::{
     error::Result,
     operation::{append_options, remove_empty_write_concern, Retryability},
     options::{AggregateOptions, SelectionCriteria, WriteConcern},
-    BoxFuture,
     ClientSession,
     Namespace,
 };
 
 use super::{
+    sync,
     CursorBody,
+    OperationResponse,
     OperationWithDefaults,
     WriteConcernOnlyBody,
     SERVER_4_2_0_WIRE_VERSION,
     SERVER_4_4_0_WIRE_VERSION,
 };
 
 pub(crate) use change_stream::ChangeStreamAggregate;
-use futures_util::FutureExt;
 
 #[derive(Debug)]
 pub(crate) struct Aggregate {
@@ -97,8 +97,8 @@ impl OperationWithDefaults for Aggregate {
         response: RawCommandResponse,
         description: &'a StreamDescription,
         _session: Option<&mut ClientSession>,
-    ) -> BoxFuture<'a, Result<Self::O>> {
-        async move {
+    ) -> OperationResponse<'static, Self::O> {
+        sync! {{
             let cursor_response: CursorBody = response.body()?;
 
             if self.is_out_or_merge() {
@@ -122,8 +122,7 @@ impl OperationWithDefaults for Aggregate {
                 self.options.as_ref().and_then(|opts| opts.max_await_time),
                 comment,
             ))
-        }
-        .boxed()
+        }}
     }
 
     fn selection_criteria(&self) -> Option<&SelectionCriteria> {

src/operation/aggregate/change_stream.rs

@@ -1,5 +1,3 @@
-use futures_util::FutureExt;
-
 use crate::{
     bson::{doc, Document},
     change_stream::{event::ResumeToken, ChangeStreamData, WatchArgs},
@@ -8,11 +6,10 @@ use crate::{
     error::Result,
     operation::{append_options, OperationWithDefaults, Retryability},
     options::{ChangeStreamOptions, SelectionCriteria, WriteConcern},
-    BoxFuture,
     ClientSession,
 };
 
-use super::Aggregate;
+use super::{sync, Aggregate, OperationResponse};
 
 pub(crate) struct ChangeStreamAggregate {
     inner: Aggregate,
@@ -91,16 +88,15 @@ impl OperationWithDefaults for ChangeStreamAggregate {
         response: RawCommandResponse,
         description: &'a StreamDescription,
         session: Option<&'a mut ClientSession>,
-    ) -> BoxFuture<'a, Result<Self::O>> {
-        async move {
+    ) -> OperationResponse<'static, Self::O> {
+        sync! {{
             let op_time = response
                 .raw_body()
                 .get("operationTime")?
                 .and_then(bson::RawBsonRef::as_timestamp);
             let spec = self
                 .inner
-                .handle_response(response, description, session)
-                .await?;
+                .handle_response(response, description, session).get_sync_result()?;
 
             let mut data = ChangeStreamData {
                 resume_token: ResumeToken::initial(self.args.options.as_ref(), &spec),
@@ -120,8 +116,7 @@ impl OperationWithDefaults for ChangeStreamAggregate {
             }
 
             Ok((spec, data))
-        }
-        .boxed()
+        }}
     }
 
     fn selection_criteria(&self) -> Option<&SelectionCriteria> {

src/operation/bulk_write.rs

@@ -21,7 +21,6 @@ use crate::{
     error::{Error, ErrorKind, Result, WriteError},
     operation::OperationWithDefaults,
     results::{DeleteResult, InsertOneResult, UpdateResult},
-    BoxFuture,
     Client,
     ClientSession,
     Cursor,
@@ -31,6 +30,7 @@ use crate::{
 
 use super::{
     CursorInfo,
+    OperationResponse,
     Retryability,
     WriteResponseBody,
     COMMAND_OVERHEAD_SIZE,
@@ -263,67 +263,69 @@ impl<'a> OperationWithDefaults for BulkWrite<'a> {
         response: RawCommandResponse,
         description: &'b StreamDescription,
         session: Option<&'b mut ClientSession>,
-    ) -> BoxFuture<'b, Result<Self::O>> {
-        async move {
-            let response: WriteResponseBody<BulkWriteResponse> = response.body()?;
-
-            let mut result = BulkWriteResult::new();
-            result.populate_summary_info(&response.summary);
-
-            let mut bulk_write_error = BulkWriteError::default();
-            if let Some(write_concern_error) = response.write_concern_error {
-                bulk_write_error
-                    .write_concern_errors
-                    .push(write_concern_error);
-            }
-
-            let specification = CursorSpecification::new(
-                response.body.cursor,
-                description.server_address.clone(),
-                None,
-                None,
-                self.options.and_then(|options| options.comment.clone()),
-            );
-            let iteration_result = match session {
-                Some(session) => {
-                    let mut session_cursor =
-                        SessionCursor::new(self.client.clone(), specification, None);
-                    self.iterate_results_cursor(
-                        session_cursor.stream(session),
-                        &mut bulk_write_error,
-                        &mut result,
-                    )
-                    .await
+    ) -> OperationResponse<'b, Self::O> {
+        OperationResponse::Async(
+            async move {
+                let response: WriteResponseBody<BulkWriteResponse> = response.body()?;
+
+                let mut result = BulkWriteResult::new();
+                result.populate_summary_info(&response.summary);
+
+                let mut bulk_write_error = BulkWriteError::default();
+                if let Some(write_concern_error) = response.write_concern_error {
+                    bulk_write_error
+                        .write_concern_errors
+                        .push(write_concern_error);
                 }
-                None => {
-                    let cursor = Cursor::new(self.client.clone(), specification, None, None);
-                    self.iterate_results_cursor(cursor, &mut bulk_write_error, &mut result)
+
+                let specification = CursorSpecification::new(
+                    response.body.cursor,
+                    description.server_address.clone(),
+                    None,
+                    None,
+                    self.options.and_then(|options| options.comment.clone()),
+                );
+                let iteration_result = match session {
+                    Some(session) => {
+                        let mut session_cursor =
+                            SessionCursor::new(self.client.clone(), specification, None);
+                        self.iterate_results_cursor(
+                            session_cursor.stream(session),
+                            &mut bulk_write_error,
+                            &mut result,
+                        )
                         .await
-                }
-            };
-
-            let source = match iteration_result {
-                Ok(()) => {
-                    if bulk_write_error.is_empty() {
-                        return Ok(result);
-                    } else {
-                        None
                     }
+                    None => {
+                        let cursor = Cursor::new(self.client.clone(), specification, None, None);
+                        self.iterate_results_cursor(cursor, &mut bulk_write_error, &mut result)
+                            .await
+                    }
+                };
+
+                let source = match iteration_result {
+                    Ok(()) => {
+                        if bulk_write_error.is_empty() {
+                            return Ok(result);
+                        } else {
+                            None
+                        }
+                    }
+                    Err(top_level_error) => Some(top_level_error),
+                };
+
+                if !result.is_empty() {
+                    bulk_write_error.partial_result = Some(result);
                 }
-                Err(top_level_error) => Some(top_level_error),
-            };
 
-            if !result.is_empty() {
-                bulk_write_error.partial_result = Some(result);
+                let error = Error::new(
+                    ErrorKind::ClientBulkWrite(bulk_write_error),
+                    response.labels,
+                );
+                Err(error.with_source(source))
             }
-
-            let error = Error::new(
-                ErrorKind::ClientBulkWrite(bulk_write_error),
-                response.labels,
-            );
-            Err(error.with_source(source))
-        }
-        .boxed()
+            .boxed(),
+        )
     }
 
     fn handle_error(&self, error: Error) -> Result<Self::O> {