pub async fn accept_recipient_reply(
        &mut self,
        source_pubkey: CommsPublicKey,
        recipient_reply: proto::RecipientSignedMessage,
    ) -> Result<(), TransactionServiceError>
    {
        let recipient_reply: RecipientSignedMessage = recipient_reply
            .try_into()
            .map_err(TransactionServiceError::InvalidMessageError)?;

        let tx_id = recipient_reply.tx_id;

        // First we check if this Reply is for a cancelled Pending Outbound Tx or a Completed Tx
        let cancelled_outbound_tx = self.db.get_cancelled_pending_outbound_transaction(tx_id).await;
        let completed_tx = self.db.get_completed_transaction_cancelled_or_not(tx_id).await;

        // This closure will check if the timestamps are beyond the cooldown period
        let check_cooldown = |timestamp: Option<NaiveDateTime>| {
            if let Some(t) = timestamp {
                // Check if the last reply is beyond the resend cooldown
                if let Ok(elapsed_time) = Utc::now().naive_utc().signed_duration_since(t).to_std() {
                    if elapsed_time < self.resources.config.resend_response_cooldown {
                        trace!(
                            target: LOG_TARGET,
                            "A repeated Transaction Reply (TxId: {}) has been received before the resend cooldown has \
                             expired. Ignoring.",
                            tx_id
                        );
                        return false;
                    }
                }
            }
            true
        };

        if let Ok(ctx) = completed_tx {
            // Check that it is from the same person
            if ctx.destination_public_key != source_pubkey {
                return Err(TransactionServiceError::InvalidSourcePublicKey);
            }
            if !check_cooldown(ctx.last_send_timestamp) {
                return Ok(());
            }

            if ctx.cancelled {
                // Send a cancellation message
                debug!(
                    target: LOG_TARGET,
                    "A repeated Transaction Reply (TxId: {}) has been received for cancelled completed transaction. \
                     Transaction Cancelled response is being sent.",
                    tx_id
                );
                tokio::spawn(send_transaction_cancelled_message(
                    tx_id,
                    source_pubkey.clone(),
                    self.resources.outbound_message_service.clone(),
                ));
            } else {
                // Resend the reply
                debug!(
                    target: LOG_TARGET,
                    "A repeated Transaction Reply (TxId: {}) has been received. Reply is being resent.", tx_id
                );
                tokio::spawn(send_finalized_transaction_message(
                    tx_id,
                    ctx.transaction,
                    source_pubkey.clone(),
                    self.resources.outbound_message_service.clone(),
                    self.resources.config.direct_send_timeout,
                ));
            }

            if let Err(e) = self.resources.db.increment_send_count(tx_id).await {
                warn!(
                    target: LOG_TARGET,
                    "Could not increment send count for completed transaction TxId {}: {:?}", tx_id, e
                );
            }
            return Ok(());
        }

        if let Ok(otx) = cancelled_outbound_tx {
            // Check that it is from the same person
            if otx.destination_public_key != source_pubkey {
                return Err(TransactionServiceError::InvalidSourcePublicKey);
            }
            if !check_cooldown(otx.last_send_timestamp) {
                return Ok(());
            }

            // Send a cancellation message
            debug!(
                target: LOG_TARGET,
                "A repeated Transaction Reply (TxId: {}) has been received for cancelled pending outbound \
                 transaction. Transaction Cancelled response is being sent.",
                tx_id
            );
            tokio::spawn(send_transaction_cancelled_message(
                tx_id,
                source_pubkey.clone(),
                self.resources.outbound_message_service.clone(),
            ));

            if let Err(e) = self.resources.db.increment_send_count(tx_id).await {
                warn!(
                    target: LOG_TARGET,
                    "Could not increment send count for completed transaction TxId {}: {:?}", tx_id, e
                );
            }
            return Ok(());
        }

        // Is this a new Transaction Reply for an existing pending transaction?
        let sender = match self.pending_transaction_reply_senders.get_mut(&tx_id) {
            None => return Err(TransactionServiceError::TransactionDoesNotExistError),
            Some(s) => s,
        };

        sender
            .send((source_pubkey, recipient_reply))
            .await
            .map_err(|_| TransactionServiceError::ProtocolChannelError)?;

        Ok(())
    }
/// Accept the public reply from a recipient and apply the reply to the relevant transaction protocol
/// # Arguments
/// 'recipient_reply' - The public response from a recipient with data required to complete the transaction
Accept the public reply from a recipient and apply the reply to the relevant transaction protocol
Accept the public reply from a recipient and apply the reply to the relevant transaction protocol
pub async fn accept_recipient_reply(
        &mut self,
        source_pubkey: CommsPublicKey,
        recipient_reply: proto::RecipientSignedMessage,
    ) -> Result<(), TransactionServiceError>
    {
        let recipient_reply: RecipientSignedMessage = recipient_reply
            .try_into()
            .map_err(TransactionServiceError::InvalidMessageError)?;
        let tx_id = recipient_reply.tx_id;
        // First we check if this Reply is for a cancelled Pending Outbound Tx or a Completed Tx
        let cancelled_outbound_tx = self.db.get_cancelled_pending_outbound_transaction(tx_id).await;
        let completed_tx = self.db.get_completed_transaction_cancelled_or_not(tx_id).await;
        // This closure will check if the timestamps are beyond the cooldown period
        let check_cooldown = |timestamp: Option<NaiveDateTime>| {
            if let Some(t) = timestamp {
                // Check if the last reply is beyond the resend cooldown
                if let Ok(elapsed_time) = Utc::now().naive_utc().signed_duration_since(t).to_std() {
                    if elapsed_time < self.resources.config.resend_response_cooldown {
                        trace!(
                            target: LOG_TARGET,
                            "A repeated Transaction Reply (TxId: {}) has been received before the resend cooldown has \
                             expired. Ignoring.",
                            tx_id
                        );
                        return false;
                    }
                }
            }
            true
        };
        if let Ok(ctx) = completed_tx {
            // Check that it is from the same person
            if ctx.destination_public_key != source_pubkey {
                return Err(TransactionServiceError::InvalidSourcePublicKey);
            }
            if !check_cooldown(ctx.last_send_timestamp) {
                return Ok(());
            }
            if ctx.cancelled {
                // Send a cancellation message
                debug!(
                    target: LOG_TARGET,
                    "A repeated Transaction Reply (TxId: {}) has been received for cancelled completed transaction. \
                     Transaction Cancelled response is being sent.",
                    tx_id
                );
                tokio::spawn(send_transaction_cancelled_message(
                    tx_id,
                    source_pubkey.clone(),
                    self.resources.outbound_message_service.clone(),
                ));
            } else {
                // Resend the reply
                debug!(
                    target: LOG_TARGET,
                    "A repeated Transaction Reply (TxId: {}) has been received. Reply is being resent.", tx_id
                );
                tokio::spawn(send_finalized_transaction_message(
                    tx_id,
                    ctx.transaction,
                    source_pubkey.clone(),
                    self.resources.outbound_message_service.clone(),
                    self.resources.config.direct_send_timeout,
                ));
            }
            if let Err(e) = self.resources.db.increment_send_count(tx_id).await {
                warn!(
                    target: LOG_TARGET,
                    "Could not increment send count for completed transaction TxId {}: {:?}", tx_id, e
                );
            }
            return Ok(());
        }
        if let Ok(otx) = cancelled_outbound_tx {
            // Check that it is from the same person
            if otx.destination_public_key != source_pubkey {
                return Err(TransactionServiceError::InvalidSourcePublicKey);
            }
            if !check_cooldown(otx.last_send_timestamp) {
                return Ok(());
            }
            // Send a cancellation message
            debug!(
                target: LOG_TARGET,
                "A repeated Transaction Reply (TxId: {}) has been received for cancelled pending outbound \
                 transaction. Transaction Cancelled response is being sent.",
                tx_id
            );
            tokio::spawn(send_transaction_cancelled_message(
                tx_id,
                source_pubkey.clone(),
                self.resources.outbound_message_service.clone(),
            ));
            if let Err(e) = self.resources.db.increment_send_count(tx_id).await {
                warn!(
                    target: LOG_TARGET,
                    "Could not increment send count for completed transaction TxId {}: {:?}", tx_id, e
                );
            }
            return Ok(());
        }
        // Is this a new Transaction Reply for an existing pending transaction?
        let sender = match self.pending_transaction_reply_senders.get_mut(&tx_id) {
            None => return Err(TransactionServiceError::TransactionDoesNotExistError),
            Some(s) => s,
        };
        sender
            .send((source_pubkey, recipient_reply))
            .await
            .map_err(|_| TransactionServiceError::ProtocolChannelError)?;
        Ok(())
    }
pub async fn accept_recipient_reply ( & mut self , source_pubkey : CommsPublicKey , recipient_reply : proto :: RecipientSignedMessage , ) -> Result < ( ) , TransactionServiceError > { let recipient_reply : RecipientSignedMessage = recipient_reply . try_into ( ) . map_err ( TransactionServiceError :: InvalidMessageError ) ? ; let tx_id = recipient_reply . tx_id ; // First we check if this Reply is for a cancelled Pending Outbound Tx or a Completed Tx let cancelled_outbound_tx = self . db . get_cancelled_pending_outbound_transaction ( tx_id ) . await ; let completed_tx = self . db . get_completed_transaction_cancelled_or_not ( tx_id ) . await ; // This closure will check if the timestamps are beyond the cooldown period let check_cooldown = | timestamp : Option < NaiveDateTime > | { if let Some ( t ) = timestamp { // Check if the last reply is beyond the resend cooldown if let Ok ( elapsed_time ) = Utc :: now ( ) . naive_utc ( ) . signed_duration_since ( t ) . to_std ( ) { if elapsed_time < self . resources . config . resend_response_cooldown { trace ! ( target LOG_TARGET " \
 " tx_id ) ; return false ; } } } true } ; if let Ok ( ctx ) = completed_tx { // Check that it is from the same person if ctx . destination_public_key != source_pubkey { return Err ( TransactionServiceError :: InvalidSourcePublicKey ) ; } if ! check_cooldown ( ctx . last_send_timestamp ) { return Ok ( ( ) ) ; } if ctx . cancelled { // Send a cancellation message debug ! ( target LOG_TARGET " \
 " tx_id ) ; tokio :: spawn ( send_transaction_cancelled_message ( tx_id , source_pubkey . clone ( ) , self . resources . outbound_message_service . clone ( ) , ) ) ; } else { // Resend the reply debug ! ( target LOG_TARGET " " tx_id ) ; tokio :: spawn ( send_finalized_transaction_message ( tx_id , ctx . transaction , source_pubkey . clone ( ) , self . resources . outbound_message_service . clone ( ) , self . resources . config . direct_send_timeout , ) ) ; } if let Err ( e ) = self . resources . db . increment_send_count ( tx_id ) . await { warn ! ( target LOG_TARGET " " tx_id e ) ; } return Ok ( ( ) ) ; } if let Ok ( otx ) = cancelled_outbound_tx { // Check that it is from the same person if otx . destination_public_key != source_pubkey { return Err ( TransactionServiceError :: InvalidSourcePublicKey ) ; } if ! check_cooldown ( otx . last_send_timestamp ) { return Ok ( ( ) ) ; } // Send a cancellation message debug ! ( target LOG_TARGET " \
 " tx_id ) ; tokio :: spawn ( send_transaction_cancelled_message ( tx_id , source_pubkey . clone ( ) , self . resources . outbound_message_service . clone ( ) , ) ) ; if let Err ( e ) = self . resources . db . increment_send_count ( tx_id ) . await { warn ! ( target LOG_TARGET " " tx_id e ) ; } return Ok ( ( ) ) ; } // Is this a new Transaction Reply for an existing pending transaction? let sender = match self . pending_transaction_reply_senders . get_mut ( & tx_id ) { None => return Err ( TransactionServiceError :: TransactionDoesNotExistError ) , Some ( s ) => s , } ; sender . send ( ( source_pubkey , recipient_reply ) ) . await . map_err ( | _ | TransactionServiceError :: ProtocolChannelError ) ? ; Ok ( ( ) ) }
Accept the public reply from a recipient and apply the reply to the relevant transaction protocol
Accept the public reply from a recipient and apply the reply to the relevant transaction protocol
&mut self
source_pubkey
CommsPublicKey
recipient_reply
proto::RecipientSignedMessage
&mut self
source_pubkey
CommsPublicKey
recipient_reply
proto::RecipientSignedMessage