/// Accept a new transaction from a sender by handling a public SenderMessage. The reply is generated and sent.
/// # Arguments
/// 'source_pubkey' - The pubkey from which the message was sent and to which the reply will be sent.
/// 'sender_message' - Message from a sender containing the setup of the transaction being sent to you
Accept a new transaction from a sender by handling a public SenderMessage. The reply is generated and sent.
Accept a new transaction from a sender by handling a public SenderMessage . The reply is generated and sent .
pub async fn accept_transaction(
        &mut self,
        source_pubkey: CommsPublicKey,
        sender_message: proto::TransactionSenderMessage,
        traced_message_tag: u64,
        join_handles: &mut FuturesUnordered<JoinHandle<Result<u64, TransactionServiceProtocolError>>>,
    ) -> Result<(), TransactionServiceError> {
        let sender_message: TransactionSenderMessage = sender_message
            .try_into()
            .map_err(TransactionServiceError::InvalidMessageError)?;
        // Currently we will only reply to a Single sender transaction protocol
        if let TransactionSenderMessage::Single(data) = sender_message.clone() {
            trace!(
                target: LOG_TARGET,
                "Transaction (TxId: {}) received from {}, Trace: {}",
                data.tx_id,
                source_pubkey,
                traced_message_tag
            );
            // Check if this transaction has already been received.
            if let Ok(inbound_tx) = self.db.get_pending_inbound_transaction(data.tx_id).await {
                // Check that it is from the same person
                if inbound_tx.source_public_key != source_pubkey {
                    return Err(TransactionServiceError::InvalidSourcePublicKey);
                }
                // Check if the last reply is beyond the resend cooldown
                if let Some(timestamp) = inbound_tx.last_send_timestamp {
                    let elapsed_time = Utc::now()
                        .naive_utc()
                        .signed_duration_since(timestamp)
                        .to_std()
                        .map_err(|_| {
                            TransactionServiceError::ConversionError("duration::OutOfRangeError".to_string())
                        })?;
                    if elapsed_time < self.resources.config.resend_response_cooldown {
                        trace!(
                            target: LOG_TARGET,
                            "A repeated Transaction (TxId: {}) has been received before the resend cooldown has \
                             expired. Ignoring.",
                            inbound_tx.tx_id
                        );
                        return Ok(());
                    }
                }
                debug!(
                    target: LOG_TARGET,
                    "A repeated Transaction (TxId: {}) has been received. Reply is being resent.", inbound_tx.tx_id
                );
                let tx_id = inbound_tx.tx_id;
                // Ok we will resend the reply
                tokio::spawn(send_transaction_reply(
                    inbound_tx,
                    self.resources.outbound_message_service.clone(),
                    self.resources.config.direct_send_timeout,
                    self.resources.config.transaction_routing_mechanism,
                ));
                if let Err(e) = self.resources.db.increment_send_count(tx_id).await {
                    warn!(
                        target: LOG_TARGET,
                        "Could not increment send count for inbound transaction TxId {}: {:?}", tx_id, e
                    );
                }
                return Ok(());
            }
            if self.finalized_transaction_senders.contains_key(&data.tx_id) ||
                self.receiver_transaction_cancellation_senders.contains_key(&data.tx_id)
            {
                trace!(
                    target: LOG_TARGET,
                    "Transaction (TxId: {}) has already been received, this is probably a repeated message, Trace:
            {}.",
                    data.tx_id,
                    traced_message_tag
                );
                return Err(TransactionServiceError::RepeatedMessageError);
            }
            let (tx_finalized_sender, tx_finalized_receiver) = mpsc::channel(100);
            let (cancellation_sender, cancellation_receiver) = oneshot::channel();
            self.finalized_transaction_senders
                .insert(data.tx_id, tx_finalized_sender);
            self.receiver_transaction_cancellation_senders
                .insert(data.tx_id, cancellation_sender);
            let protocol = TransactionReceiveProtocol::new(
                data.tx_id,
                source_pubkey,
                sender_message,
                TransactionReceiveProtocolStage::Initial,
                self.resources.clone(),
                tx_finalized_receiver,
                cancellation_receiver,
            );
            let join_handle = tokio::spawn(protocol.execute());
            join_handles.push(join_handle);
            Ok(())
        } else {
            Err(TransactionServiceError::InvalidStateError)
        }
    }
pub async fn accept_transaction ( & mut self , source_pubkey : CommsPublicKey , sender_message : proto :: TransactionSenderMessage , traced_message_tag : u64 , join_handles : & mut FuturesUnordered < JoinHandle < Result < u64 , TransactionServiceProtocolError > > > , ) -> Result < ( ) , TransactionServiceError > { let sender_message : TransactionSenderMessage = sender_message . try_into ( ) . map_err ( TransactionServiceError :: InvalidMessageError ) ? ; // Currently we will only reply to a Single sender transaction protocol if let TransactionSenderMessage :: Single ( data ) = sender_message . clone ( ) { trace ! ( target LOG_TARGET " " data tx_id source_pubkey traced_message_tag ) ; // Check if this transaction has already been received. if let Ok ( inbound_tx ) = self . db . get_pending_inbound_transaction ( data . tx_id ) . await { // Check that it is from the same person if inbound_tx . source_public_key != source_pubkey { return Err ( TransactionServiceError :: InvalidSourcePublicKey ) ; } // Check if the last reply is beyond the resend cooldown if let Some ( timestamp ) = inbound_tx . last_send_timestamp { let elapsed_time = Utc :: now ( ) . naive_utc ( ) . signed_duration_since ( timestamp ) . to_std ( ) . map_err ( | _ | { TransactionServiceError :: ConversionError ( " " . to_string ( ) ) } ) ? ; if elapsed_time < self . resources . config . resend_response_cooldown { trace ! ( target LOG_TARGET " \
 " inbound_tx tx_id ) ; return Ok ( ( ) ) ; } } debug ! ( target LOG_TARGET " " inbound_tx tx_id ) ; let tx_id = inbound_tx . tx_id ; // Ok we will resend the reply tokio :: spawn ( send_transaction_reply ( inbound_tx , self . resources . outbound_message_service . clone ( ) , self . resources . config . direct_send_timeout , self . resources . config . transaction_routing_mechanism , ) ) ; if let Err ( e ) = self . resources . db . increment_send_count ( tx_id ) . await { warn ! ( target LOG_TARGET " " tx_id e ) ; } return Ok ( ( ) ) ; } if self . finalized_transaction_senders . contains_key ( & data . tx_id ) || self . receiver_transaction_cancellation_senders . contains_key ( & data . tx_id ) { trace ! ( target LOG_TARGET " " data tx_id traced_message_tag ) ; return Err ( TransactionServiceError :: RepeatedMessageError ) ; } let ( tx_finalized_sender , tx_finalized_receiver ) = mpsc :: channel ( 100 ) ; let ( cancellation_sender , cancellation_receiver ) = oneshot :: channel ( ) ; self . finalized_transaction_senders . insert ( data . tx_id , tx_finalized_sender ) ; self . receiver_transaction_cancellation_senders . insert ( data . tx_id , cancellation_sender ) ; let protocol = TransactionReceiveProtocol :: new ( data . tx_id , source_pubkey , sender_message , TransactionReceiveProtocolStage :: Initial , self . resources . clone ( ) , tx_finalized_receiver , cancellation_receiver , ) ; let join_handle = tokio :: spawn ( protocol . execute ( ) ) ; join_handles . push ( join_handle ) ; Ok ( ( ) ) } else { Err ( TransactionServiceError :: InvalidStateError ) } }
Accept a new transaction from a sender by handling a public SenderMessage.
Accept a new transaction from a sender by handling a public SenderMessage .
&mut self
source_pubkey
CommsPublicKey
sender_message
proto::TransactionSenderMessage
traced_message_tag
u64
join_handles
&mut FuturesUnordered<JoinHandle<Result<u64, TransactionServiceProtocolError>>>