async fn execute_operation_with_retry<T: Operation>(
        &self,
        mut op: T,
        mut session: Option<&mut ClientSession>,
    ) -> Result<ExecutionOutput<T>> {
        // If the current transaction has been committed/aborted and it is not being
        // re-committed/re-aborted, reset the transaction's state to TransactionState::None.
        if let Some(ref mut session) = session {
            if matches!(
                session.transaction.state,
                TransactionState::Committed { .. }
            ) && op.name() != CommitTransaction::NAME
                || session.transaction.state == TransactionState::Aborted
                    && op.name() != AbortTransaction::NAME
            {
                session.transaction.reset();
            }
        }

        let selection_criteria = session
            .as_ref()
            .and_then(|s| s.transaction.pinned_mongos())
            .or_else(|| op.selection_criteria());

        let server = match self.select_server(selection_criteria).await {
            Ok(server) => server,
            Err(mut err) => {
                err.add_labels_and_update_pin(None, &mut session, None)?;
                return Err(err);
            }
        };

        let mut conn = match get_connection(&session, &op, &server.pool).await {
            Ok(c) => c,
            Err(mut err) => {
                err.add_labels_and_update_pin(None, &mut session, None)?;

                if err.is_pool_cleared() {
                    return self.execute_retry(&mut op, &mut session, None, err).await;
                } else {
                    return Err(err);
                }
            }
        };

        let retryability = self.get_retryability(&conn, &op, &session).await?;

        let txn_number = match session {
            Some(ref mut session) => {
                if session.transaction.state != TransactionState::None {
                    Some(session.txn_number())
                } else {
                    match retryability {
                        Retryability::Write => Some(session.get_and_increment_txn_number()),
                        _ => None,
                    }
                }
            }
            None => None,
        };

        match self
            .execute_operation_on_connection(
                &mut op,
                &mut conn,
                &mut session,
                txn_number,
                &retryability,
            )
            .await
        {
            Ok(operation_output) => Ok(ExecutionOutput {
                operation_output,
                connection: conn,
            }),
            Err(mut err) => {
                // Retryable writes are only supported by storage engines with document-level
                // locking, so users need to disable retryable writes if using mmapv1.
                if let ErrorKind::Command(ref mut command_error) = *err.kind {
                    if command_error.code == 20
                        && command_error.message.starts_with("Transaction numbers")
                    {
                        command_error.message = "This MongoDB deployment does not support \
                                                 retryable writes. Please add retryWrites=false \
                                                 to your connection string."
                            .to_string();
                    }
                }

                self.inner
                    .topology
                    .handle_application_error(
                        err.clone(),
                        HandshakePhase::after_completion(&conn),
                        &server,
                    )
                    .await;
                // release the connection to be processed by the connection pool
                drop(conn);
                // release the selected server to decrement its operation count
                drop(server);

                if retryability == Retryability::Read && err.is_read_retryable()
                    || retryability == Retryability::Write && err.is_write_retryable()
                {
                    self.execute_retry(&mut op, &mut session, txn_number, err)
                        .await
                } else {
                    Err(err)
                }
            }
        }
    }
/// Selects a server and executes the given operation on it, optionally using a provided
/// session. Retries the operation upon failure if retryability is supported.
Selects a server and executes the given operation on it, optionally using a provided
session. Retries the operation upon failure if retryability is supported.
Selects a server and executes the given operation on it optionally using a provided session . Retries the operation upon failure if retryability is supported .
async fn execute_operation_with_retry<T: Operation>(
        &self,
        mut op: T,
        mut session: Option<&mut ClientSession>,
    ) -> Result<ExecutionOutput<T>> {
        // If the current transaction has been committed/aborted and it is not being
        // re-committed/re-aborted, reset the transaction's state to TransactionState::None.
        if let Some(ref mut session) = session {
            if matches!(
                session.transaction.state,
                TransactionState::Committed { .. }
            ) && op.name() != CommitTransaction::NAME
                || session.transaction.state == TransactionState::Aborted
                    && op.name() != AbortTransaction::NAME
            {
                session.transaction.reset();
            }
        }
        let selection_criteria = session
            .as_ref()
            .and_then(|s| s.transaction.pinned_mongos())
            .or_else(|| op.selection_criteria());
        let server = match self.select_server(selection_criteria).await {
            Ok(server) => server,
            Err(mut err) => {
                err.add_labels_and_update_pin(None, &mut session, None)?;
                return Err(err);
            }
        };
        let mut conn = match get_connection(&session, &op, &server.pool).await {
            Ok(c) => c,
            Err(mut err) => {
                err.add_labels_and_update_pin(None, &mut session, None)?;
                if err.is_pool_cleared() {
                    return self.execute_retry(&mut op, &mut session, None, err).await;
                } else {
                    return Err(err);
                }
            }
        };
        let retryability = self.get_retryability(&conn, &op, &session).await?;
        let txn_number = match session {
            Some(ref mut session) => {
                if session.transaction.state != TransactionState::None {
                    Some(session.txn_number())
                } else {
                    match retryability {
                        Retryability::Write => Some(session.get_and_increment_txn_number()),
                        _ => None,
                    }
                }
            }
            None => None,
        };
        match self
            .execute_operation_on_connection(
                &mut op,
                &mut conn,
                &mut session,
                txn_number,
                &retryability,
            )
            .await
        {
            Ok(operation_output) => Ok(ExecutionOutput {
                operation_output,
                connection: conn,
            }),
            Err(mut err) => {
                // Retryable writes are only supported by storage engines with document-level
                // locking, so users need to disable retryable writes if using mmapv1.
                if let ErrorKind::Command(ref mut command_error) = *err.kind {
                    if command_error.code == 20
                        && command_error.message.starts_with("Transaction numbers")
                    {
                        command_error.message = "This MongoDB deployment does not support \
                                                 retryable writes. Please add retryWrites=false \
                                                 to your connection string."
                            .to_string();
                    }
                }
                self.inner
                    .topology
                    .handle_application_error(
                        err.clone(),
                        HandshakePhase::after_completion(&conn),
                        &server,
                    )
                    .await;
                // release the connection to be processed by the connection pool
                drop(conn);
                // release the selected server to decrement its operation count
                drop(server);
                if retryability == Retryability::Read && err.is_read_retryable()
                    || retryability == Retryability::Write && err.is_write_retryable()
                {
                    self.execute_retry(&mut op, &mut session, txn_number, err)
                        .await
                } else {
                    Err(err)
                }
            }
        }
    }
async fn execute_operation_with_retry < T : Operation > ( & self , mut op : T , mut session : Option < & mut ClientSession > , ) -> Result < ExecutionOutput < T > > { // If the current transaction has been committed/aborted and it is not being // re-committed/re-aborted, reset the transaction's state to TransactionState::None. if let Some ( ref mut session ) = session { if matches ! ( session transaction state TransactionState Committed { } ) && op . name ( ) != CommitTransaction :: NAME || session . transaction . state == TransactionState :: Aborted && op . name ( ) != AbortTransaction :: NAME { session . transaction . reset ( ) ; } } let selection_criteria = session . as_ref ( ) . and_then ( | s | s . transaction . pinned_mongos ( ) ) . or_else ( | | op . selection_criteria ( ) ) ; let server = match self . select_server ( selection_criteria ) . await { Ok ( server ) => server , Err ( mut err ) => { err . add_labels_and_update_pin ( None , & mut session , None ) ? ; return Err ( err ) ; } } ; let mut conn = match get_connection ( & session , & op , & server . pool ) . await { Ok ( c ) => c , Err ( mut err ) => { err . add_labels_and_update_pin ( None , & mut session , None ) ? ; if err . is_pool_cleared ( ) { return self . execute_retry ( & mut op , & mut session , None , err ) . await ; } else { return Err ( err ) ; } } } ; let retryability = self . get_retryability ( & conn , & op , & session ) . await ? ; let txn_number = match session { Some ( ref mut session ) => { if session . transaction . state != TransactionState :: None { Some ( session . txn_number ( ) ) } else { match retryability { Retryability :: Write => Some ( session . get_and_increment_txn_number ( ) ) , _ => None , } } } None => None , } ; match self . execute_operation_on_connection ( & mut op , & mut conn , & mut session , txn_number , & retryability , ) . await { Ok ( operation_output ) => Ok ( ExecutionOutput { operation_output , connection : conn , } ) , Err ( mut err ) => { // Retryable writes are only supported by storage engines with document-level // locking, so users need to disable retryable writes if using mmapv1. if let ErrorKind :: Command ( ref mut command_error ) = * err . kind { if command_error . code == 20 && command_error . message . starts_with ( " " ) { command_error . message = " \
 \
 " . to_string ( ) ; } } self . inner . topology . handle_application_error ( err . clone ( ) , HandshakePhase :: after_completion ( & conn ) , & server , ) . await ; // release the connection to be processed by the connection pool drop ( conn ) ; // release the selected server to decrement its operation count drop ( server ) ; if retryability == Retryability :: Read && err . is_read_retryable ( ) || retryability == Retryability :: Write && err . is_write_retryable ( ) { self . execute_retry ( & mut op , & mut session , txn_number , err ) . await } else { Err ( err ) } } } }
Selects a server and executes the given operation on it, optionally using a provided
session.
Selects a server and executes the given operation on it optionally using a provided session .
&self
op
T
session
Option<&mut ClientSession>
&self
op
T
session
Option<&mut ClientSession>