pub async fn execute(mut self) -> Result<u64, TransactionServiceProtocolError> {
        let mut timeout_update_receiver = self
            .timeout_update_receiver
            .take()
            .ok_or_else(|| {
                TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::InvalidStateError)
            })?
            .fuse();

        let mut base_node_update_receiver = self
            .base_node_update_receiver
            .take()
            .ok_or_else(|| {
                TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::InvalidStateError)
            })?
            .fuse();

        let mut shutdown = self.resources.shutdown_signal.clone();
        // Main protocol loop
        loop {
            let base_node_node_id = NodeId::from_key(&self.base_node_public_key);
            let mut connection: Option<PeerConnection> = None;

            let delay = delay_for(self.timeout);

            debug!(
                target: LOG_TARGET,
                "Connecting to Base Node (Public Key: {})", self.base_node_public_key,
            );
            futures::select! {
                dial_result = self.resources.connectivity_manager.dial_peer(base_node_node_id.clone()).fuse() => {
                    match dial_result {
                        Ok(base_node_connection) => {
                            connection = Some(base_node_connection);
                        },
                        Err(e) => {
                            info!(target: LOG_TARGET, "Problem connecting to base node: {} for Transaction Broadcast Protocol (TxId: {})", e, self.tx_id);
                            let _ = self
                            .resources
                            .event_publisher
                            .send(Arc::new(TransactionEvent::TransactionBaseNodeConnectionProblem(
                                self.tx_id,
                            )))
                            .map_err(|e| {
                                trace!(
                                    target: LOG_TARGET,
                                    "Error sending event because there are no subscribers: {:?}",
                                    e
                                );
                                e
                            });
                        },
                    }
                },
                updated_timeout = timeout_update_receiver.select_next_some() => {
                    match updated_timeout {
                        Ok(to) => {
                            self.timeout = to;
                             info!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol (TxId: {}) timeout updated to {:?}", self.tx_id, self.timeout
                            );
                        },
                        Err(e) => {
                            trace!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol (TxId: {}) event 'updated_timeout' triggered with error: {:?}",
                                self.tx_id,
                                e,
                            );
                        }
                    }
                },
                new_base_node = base_node_update_receiver.select_next_some() => {
                    match new_base_node {
                        Ok(bn) => {
                            self.base_node_public_key = bn;
                             info!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol (TxId: {}) Base Node Public key updated to {:?}", self.tx_id, self.base_node_public_key
                            );
                            self.first_rejection = false;
                            continue;
                        },
                        Err(e) => {
                            trace!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol (TxId: {}) event 'base_node_update' triggered with error: {:?}",
                                self.tx_id,
                                e,
                            );
                        }
                    }
                }
                _ = shutdown => {
                    info!(target: LOG_TARGET, "Transaction Broadcast Protocol (TxId: {}) shutting down because it received the shutdown signal", self.tx_id);
                    return Err(TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::Shutdown))
                },
            }

            let mut base_node_connection = match connection {
                None => {
                    futures::select! {
                        _ = delay.fuse() => {
                            continue;
                        },
                        _ = shutdown => {
                            info!(target: LOG_TARGET, "Transaction Broadcast Protocol (TxId: {}) shutting down because it received the shutdown signal", self.tx_id);
                            return Err(TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::Shutdown))
                        },
                    }
                },
                Some(c) => c,
            };

            let completed_tx = match self.resources.db.get_completed_transaction(self.tx_id).await {
                Ok(tx) => tx,
                Err(e) => {
                    error!(
                        target: LOG_TARGET,
                        "Cannot find Completed Transaction (TxId: {}) referred to by this Broadcast protocol: {:?}",
                        self.tx_id,
                        e
                    );
                    return Err(TransactionServiceProtocolError::new(
                        self.tx_id,
                        TransactionServiceError::TransactionDoesNotExistError,
                    ));
                },
            };

            if !(completed_tx.status == TransactionStatus::Completed ||
                completed_tx.status == TransactionStatus::Broadcast ||
                completed_tx.status == TransactionStatus::MinedUnconfirmed)
            {
                debug!(
                    target: LOG_TARGET,
                    "Transaction (TxId: {}) no longer in Completed state and will stop being broadcast", self.tx_id
                );
                return Ok(self.tx_id);
            }

            let mut client = match base_node_connection
                .connect_rpc_using_builder(
                    BaseNodeWalletRpcClient::builder()
                        .with_deadline(self.resources.config.broadcast_monitoring_timeout),
                )
                .await
            {
                Ok(c) => c,
                Err(e) => {
                    warn!(target: LOG_TARGET, "Problem establishing RPC connection: {}", e);
                    delay.await;
                    continue;
                },
            };

            let delay = delay_for(self.timeout);
            loop {
                futures::select! {
                    new_base_node = base_node_update_receiver.select_next_some() => {
                        match new_base_node {
                            Ok(bn) => {
                                self.base_node_public_key = bn;
                                 info!(
                                    target: LOG_TARGET,
                                    "Transaction Broadcast protocol (TxId: {}) Base Node Public key updated to {:?}", self.tx_id, self.base_node_public_key
                                );
                                self.first_rejection = false;
                                continue;
                            },
                            Err(e) => {
                                trace!(
                                    target: LOG_TARGET,
                                    "Transaction Broadcast protocol (TxId: {}) event 'base_node_update' triggered with error: {:?}",
                                    self.tx_id,
                                    e,
                                );
                            }
                        }
                    },
                    result = self.query_or_submit_transaction(completed_tx.clone(), &mut client).fuse() => {
                        match self.mode {
                            TxBroadcastMode::TransactionSubmission => {
                                if result? {
                                    self.mode = TxBroadcastMode::TransactionQuery;
                                }
                            },
                            TxBroadcastMode::TransactionQuery => {
                                if result? {
                                    // We are done!
                                    self.resources
                                        .output_manager_service
                                        .confirm_transaction(
                                            completed_tx.tx_id,
                                            completed_tx.transaction.body.inputs().clone(),
                                            completed_tx.transaction.body.outputs().clone(),
                                        )
                                        .await
                                        .map_err(|e| TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::from(e)))?;

                                    self.resources
                                        .db
                                        .confirm_broadcast_or_coinbase_transaction(completed_tx.tx_id)
                                        .await
                                        .map_err(|e| TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::from(e)))?;

                                   let _ = self
                                        .resources
                                        .event_publisher
                                        .send(Arc::new(TransactionEvent::TransactionMined(self.tx_id)))
                                        .map_err(|e| {
                                            trace!(
                                                target: LOG_TARGET,
                                                "Error sending event because there are no subscribers: {:?}",
                                                e
                                            );
                                            e
                                        });

                                    return Ok(self.tx_id)
                                }
                            },
                        }
                        // Wait out the remainder of the delay before proceeding with next loop
                        drop(client);
                        delay.await;
                        break;
                    },
                    updated_timeout = timeout_update_receiver.select_next_some() => {
                        if let Ok(to) = updated_timeout {
                            self.timeout = to;
                             info!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol (TxId: {}) timeout updated to {:?}", self.tx_id, self.timeout
                            );
                            break;
                        } else {
                            trace!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol event 'updated_timeout' triggered (TxId: {}) ({:?})",
                                self.tx_id,
                                updated_timeout,
                            );
                        }
                    },
                    _ = shutdown => {
                        info!(target: LOG_TARGET, "Transaction Broadcast Protocol (TxId: {}) shutting down because it received the shutdown signal", self.tx_id);
                        return Err(TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::Shutdown))
                    },
                }
            }
        }
    }
/// The task that defines the execution of the protocol.
The task that defines the execution of the protocol.
The task that defines the execution of the protocol .
pub async fn execute(mut self) -> Result<u64, TransactionServiceProtocolError> {
        let mut timeout_update_receiver = self
            .timeout_update_receiver
            .take()
            .ok_or_else(|| {
                TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::InvalidStateError)
            })?
            .fuse();
        let mut base_node_update_receiver = self
            .base_node_update_receiver
            .take()
            .ok_or_else(|| {
                TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::InvalidStateError)
            })?
            .fuse();
        let mut shutdown = self.resources.shutdown_signal.clone();
        // Main protocol loop
        loop {
            let base_node_node_id = NodeId::from_key(&self.base_node_public_key);
            let mut connection: Option<PeerConnection> = None;
            let delay = delay_for(self.timeout);
            debug!(
                target: LOG_TARGET,
                "Connecting to Base Node (Public Key: {})", self.base_node_public_key,
            );
            futures::select! {
                dial_result = self.resources.connectivity_manager.dial_peer(base_node_node_id.clone()).fuse() => {
                    match dial_result {
                        Ok(base_node_connection) => {
                            connection = Some(base_node_connection);
                        },
                        Err(e) => {
                            info!(target: LOG_TARGET, "Problem connecting to base node: {} for Transaction Broadcast Protocol (TxId: {})", e, self.tx_id);
                            let _ = self
                            .resources
                            .event_publisher
                            .send(Arc::new(TransactionEvent::TransactionBaseNodeConnectionProblem(
                                self.tx_id,
                            )))
                            .map_err(|e| {
                                trace!(
                                    target: LOG_TARGET,
                                    "Error sending event because there are no subscribers: {:?}",
                                    e
                                );
                                e
                            });
                        },
                    }
                },
                updated_timeout = timeout_update_receiver.select_next_some() => {
                    match updated_timeout {
                        Ok(to) => {
                            self.timeout = to;
                             info!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol (TxId: {}) timeout updated to {:?}", self.tx_id, self.timeout
                            );
                        },
                        Err(e) => {
                            trace!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol (TxId: {}) event 'updated_timeout' triggered with error: {:?}",
                                self.tx_id,
                                e,
                            );
                        }
                    }
                },
                new_base_node = base_node_update_receiver.select_next_some() => {
                    match new_base_node {
                        Ok(bn) => {
                            self.base_node_public_key = bn;
                             info!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol (TxId: {}) Base Node Public key updated to {:?}", self.tx_id, self.base_node_public_key
                            );
                            self.first_rejection = false;
                            continue;
                        },
                        Err(e) => {
                            trace!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol (TxId: {}) event 'base_node_update' triggered with error: {:?}",
                                self.tx_id,
                                e,
                            );
                        }
                    }
                }
                _ = shutdown => {
                    info!(target: LOG_TARGET, "Transaction Broadcast Protocol (TxId: {}) shutting down because it received the shutdown signal", self.tx_id);
                    return Err(TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::Shutdown))
                },
            }
            let mut base_node_connection = match connection {
                None => {
                    futures::select! {
                        _ = delay.fuse() => {
                            continue;
                        },
                        _ = shutdown => {
                            info!(target: LOG_TARGET, "Transaction Broadcast Protocol (TxId: {}) shutting down because it received the shutdown signal", self.tx_id);
                            return Err(TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::Shutdown))
                        },
                    }
                },
                Some(c) => c,
            };
            let completed_tx = match self.resources.db.get_completed_transaction(self.tx_id).await {
                Ok(tx) => tx,
                Err(e) => {
                    error!(
                        target: LOG_TARGET,
                        "Cannot find Completed Transaction (TxId: {}) referred to by this Broadcast protocol: {:?}",
                        self.tx_id,
                        e
                    );
                    return Err(TransactionServiceProtocolError::new(
                        self.tx_id,
                        TransactionServiceError::TransactionDoesNotExistError,
                    ));
                },
            };
            if !(completed_tx.status == TransactionStatus::Completed ||
                completed_tx.status == TransactionStatus::Broadcast ||
                completed_tx.status == TransactionStatus::MinedUnconfirmed)
            {
                debug!(
                    target: LOG_TARGET,
                    "Transaction (TxId: {}) no longer in Completed state and will stop being broadcast", self.tx_id
                );
                return Ok(self.tx_id);
            }
            let mut client = match base_node_connection
                .connect_rpc_using_builder(
                    BaseNodeWalletRpcClient::builder()
                        .with_deadline(self.resources.config.broadcast_monitoring_timeout),
                )
                .await
            {
                Ok(c) => c,
                Err(e) => {
                    warn!(target: LOG_TARGET, "Problem establishing RPC connection: {}", e);
                    delay.await;
                    continue;
                },
            };
            let delay = delay_for(self.timeout);
            loop {
                futures::select! {
                    new_base_node = base_node_update_receiver.select_next_some() => {
                        match new_base_node {
                            Ok(bn) => {
                                self.base_node_public_key = bn;
                                 info!(
                                    target: LOG_TARGET,
                                    "Transaction Broadcast protocol (TxId: {}) Base Node Public key updated to {:?}", self.tx_id, self.base_node_public_key
                                );
                                self.first_rejection = false;
                                continue;
                            },
                            Err(e) => {
                                trace!(
                                    target: LOG_TARGET,
                                    "Transaction Broadcast protocol (TxId: {}) event 'base_node_update' triggered with error: {:?}",
                                    self.tx_id,
                                    e,
                                );
                            }
                        }
                    },
                    result = self.query_or_submit_transaction(completed_tx.clone(), &mut client).fuse() => {
                        match self.mode {
                            TxBroadcastMode::TransactionSubmission => {
                                if result? {
                                    self.mode = TxBroadcastMode::TransactionQuery;
                                }
                            },
                            TxBroadcastMode::TransactionQuery => {
                                if result? {
                                    // We are done!
                                    self.resources
                                        .output_manager_service
                                        .confirm_transaction(
                                            completed_tx.tx_id,
                                            completed_tx.transaction.body.inputs().clone(),
                                            completed_tx.transaction.body.outputs().clone(),
                                        )
                                        .await
                                        .map_err(|e| TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::from(e)))?;
                                    self.resources
                                        .db
                                        .confirm_broadcast_or_coinbase_transaction(completed_tx.tx_id)
                                        .await
                                        .map_err(|e| TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::from(e)))?;
                                   let _ = self
                                        .resources
                                        .event_publisher
                                        .send(Arc::new(TransactionEvent::TransactionMined(self.tx_id)))
                                        .map_err(|e| {
                                            trace!(
                                                target: LOG_TARGET,
                                                "Error sending event because there are no subscribers: {:?}",
                                                e
                                            );
                                            e
                                        });
                                    return Ok(self.tx_id)
                                }
                            },
                        }
                        // Wait out the remainder of the delay before proceeding with next loop
                        drop(client);
                        delay.await;
                        break;
                    },
                    updated_timeout = timeout_update_receiver.select_next_some() => {
                        if let Ok(to) = updated_timeout {
                            self.timeout = to;
                             info!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol (TxId: {}) timeout updated to {:?}", self.tx_id, self.timeout
                            );
                            break;
                        } else {
                            trace!(
                                target: LOG_TARGET,
                                "Transaction Broadcast protocol event 'updated_timeout' triggered (TxId: {}) ({:?})",
                                self.tx_id,
                                updated_timeout,
                            );
                        }
                    },
                    _ = shutdown => {
                        info!(target: LOG_TARGET, "Transaction Broadcast Protocol (TxId: {}) shutting down because it received the shutdown signal", self.tx_id);
                        return Err(TransactionServiceProtocolError::new(self.tx_id, TransactionServiceError::Shutdown))
                    },
                }
            }
        }
    }
pub async fn execute ( mut self ) -> Result < u64 , TransactionServiceProtocolError > { let mut timeout_update_receiver = self . timeout_update_receiver . take ( ) . ok_or_else ( | | { TransactionServiceProtocolError :: new ( self . tx_id , TransactionServiceError :: InvalidStateError ) } ) ? . fuse ( ) ; let mut base_node_update_receiver = self . base_node_update_receiver . take ( ) . ok_or_else ( | | { TransactionServiceProtocolError :: new ( self . tx_id , TransactionServiceError :: InvalidStateError ) } ) ? . fuse ( ) ; let mut shutdown = self . resources . shutdown_signal . clone ( ) ; // Main protocol loop loop { let base_node_node_id = NodeId :: from_key ( & self . base_node_public_key ) ; let mut connection : Option < PeerConnection > = None ; let delay = delay_for ( self . timeout ) ; debug ! ( target LOG_TARGET " " self base_node_public_key ) ; futures :: select ! { dial_result self resources connectivity_manager dial_peer ( base_node_node_id clone ( ) ) fuse ( ) { match dial_result { Ok ( base_node_connection ) { connection Some ( base_node_connection ) } Err ( e ) { info ( target LOG_TARGET " " e self tx_id ) let self resources event_publisher send ( Arc new ( TransactionEvent TransactionBaseNodeConnectionProblem ( self tx_id ) ) ) map_err ( e { trace ( target LOG_TARGET " " e ) e } ) } } } updated_timeout timeout_update_receiver select_next_some ( ) { match updated_timeout { Ok ( to ) { self timeout to info ( target LOG_TARGET " " self tx_id self timeout ) } Err ( e ) { trace ( target LOG_TARGET " " self tx_id e ) } } } new_base_node base_node_update_receiver select_next_some ( ) { match new_base_node { Ok ( bn ) { self base_node_public_key bn info ( target LOG_TARGET " " self tx_id self base_node_public_key ) self first_rejection false continue } Err ( e ) { trace ( target LOG_TARGET " " self tx_id e ) } } } shutdown { info ( target LOG_TARGET " " self tx_id ) return Err ( TransactionServiceProtocolError new ( self tx_id TransactionServiceError Shutdown ) ) } } let mut base_node_connection = match connection { None => { futures :: select ! { delay fuse ( ) { continue } shutdown { info ( target LOG_TARGET " " self tx_id ) return Err ( TransactionServiceProtocolError new ( self tx_id TransactionServiceError Shutdown ) ) } } } , Some ( c ) => c , } ; let completed_tx = match self . resources . db . get_completed_transaction ( self . tx_id ) . await { Ok ( tx ) => tx , Err ( e ) => { error ! ( target LOG_TARGET " " self tx_id e ) ; return Err ( TransactionServiceProtocolError :: new ( self . tx_id , TransactionServiceError :: TransactionDoesNotExistError , ) ) ; } , } ; if ! ( completed_tx . status == TransactionStatus :: Completed || completed_tx . status == TransactionStatus :: Broadcast || completed_tx . status == TransactionStatus :: MinedUnconfirmed ) { debug ! ( target LOG_TARGET " " self tx_id ) ; return Ok ( self . tx_id ) ; } let mut client = match base_node_connection . connect_rpc_using_builder ( BaseNodeWalletRpcClient :: builder ( ) . with_deadline ( self . resources . config . broadcast_monitoring_timeout ) , ) . await { Ok ( c ) => c , Err ( e ) => { warn ! ( target LOG_TARGET " " e ) ; delay . await ; continue ; } , } ; let delay = delay_for ( self . timeout ) ; loop { futures :: select ! { new_base_node base_node_update_receiver select_next_some ( ) { match new_base_node { Ok ( bn ) { self base_node_public_key bn info ( target LOG_TARGET " " self tx_id self base_node_public_key ) self first_rejection false continue } Err ( e ) { trace ( target LOG_TARGET " " self tx_id e ) } } } result self query_or_submit_transaction ( completed_tx clone ( ) mut client ) fuse ( ) { match self mode { TxBroadcastMode TransactionSubmission { if result { self mode TxBroadcastMode TransactionQuery } } TxBroadcastMode TransactionQuery { if result { // We are done! self resources output_manager_service confirm_transaction ( completed_tx tx_id completed_tx transaction body inputs ( ) clone ( ) completed_tx transaction body outputs ( ) clone ( ) ) await map_err ( e TransactionServiceProtocolError new ( self tx_id TransactionServiceError from ( e ) ) ) self resources db confirm_broadcast_or_coinbase_transaction ( completed_tx tx_id ) await map_err ( e TransactionServiceProtocolError new ( self tx_id TransactionServiceError from ( e ) ) ) let self resources event_publisher send ( Arc new ( TransactionEvent TransactionMined ( self tx_id ) ) ) map_err ( e { trace ( target LOG_TARGET " " e ) e } ) return Ok ( self tx_id ) } } } // Wait out the remainder of the delay before proceeding with next loop drop ( client ) delay await break } updated_timeout timeout_update_receiver select_next_some ( ) { if let Ok ( to ) updated_timeout { self timeout to info ( target LOG_TARGET " " self tx_id self timeout ) break } else { trace ( target LOG_TARGET " " self tx_id updated_timeout ) } } shutdown { info ( target LOG_TARGET " " self tx_id ) return Err ( TransactionServiceProtocolError new ( self tx_id TransactionServiceError Shutdown ) ) } } } } }
The task that defines the execution of the protocol.
The task that defines the execution of the protocol .
mut self
mut self