async fn handle_request(
        &mut self,
        request: TransactionServiceRequest,
        send_transaction_join_handles: &mut FuturesUnordered<JoinHandle<Result<u64, TransactionServiceProtocolError>>>,
        receive_transaction_join_handles: &mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >,
        transaction_broadcast_join_handles: &mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >,
        chain_monitoring_join_handles: &mut FuturesUnordered<JoinHandle<Result<u64, TransactionServiceProtocolError>>>,
        coinbase_monitoring_join_handles: &mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >,
    ) -> Result<TransactionServiceResponse, TransactionServiceError>
    {
        trace!(target: LOG_TARGET, "Handling Service Request: {}", request);
        match request {
            TransactionServiceRequest::SendTransaction((dest_pubkey, amount, fee_per_gram, message)) => self
                .send_transaction(
                    dest_pubkey,
                    amount,
                    fee_per_gram,
                    message,
                    send_transaction_join_handles,
                )
                .await
                .map(TransactionServiceResponse::TransactionSent),
            TransactionServiceRequest::CancelTransaction(tx_id) => self
                .cancel_transaction(tx_id)
                .await
                .map(|_| TransactionServiceResponse::TransactionCancelled),
            TransactionServiceRequest::GetPendingInboundTransactions => {
                Ok(TransactionServiceResponse::PendingInboundTransactions(
                    self.db.get_pending_inbound_transactions().await?,
                ))
            },
            TransactionServiceRequest::GetPendingOutboundTransactions => {
                Ok(TransactionServiceResponse::PendingOutboundTransactions(
                    self.db.get_pending_outbound_transactions().await?,
                ))
            },

            TransactionServiceRequest::GetCompletedTransactions => Ok(
                TransactionServiceResponse::CompletedTransactions(self.db.get_completed_transactions().await?),
            ),
            TransactionServiceRequest::GetCancelledPendingInboundTransactions => {
                Ok(TransactionServiceResponse::PendingInboundTransactions(
                    self.db.get_cancelled_pending_inbound_transactions().await?,
                ))
            },
            TransactionServiceRequest::GetCancelledPendingOutboundTransactions => {
                Ok(TransactionServiceResponse::PendingOutboundTransactions(
                    self.db.get_cancelled_pending_outbound_transactions().await?,
                ))
            },
            TransactionServiceRequest::GetCancelledCompletedTransactions => {
                Ok(TransactionServiceResponse::CompletedTransactions(
                    self.db.get_cancelled_completed_transactions().await?,
                ))
            },
            TransactionServiceRequest::GetCompletedTransaction(tx_id) => {
                Ok(TransactionServiceResponse::CompletedTransaction(Box::new(
                    self.db.get_completed_transaction(tx_id).await?,
                )))
            },
            TransactionServiceRequest::GetAnyTransaction(tx_id) => Ok(TransactionServiceResponse::AnyTransaction(
                Box::new(self.db.get_any_transaction(tx_id).await?),
            )),
            TransactionServiceRequest::SetBaseNodePublicKey(public_key) => {
                self.set_base_node_public_key(public_key);
                Ok(TransactionServiceResponse::BaseNodePublicKeySet)
            },
            TransactionServiceRequest::ImportUtxo(value, source_public_key, message) => self
                .add_utxo_import_transaction(value, source_public_key, message)
                .await
                .map(TransactionServiceResponse::UtxoImported),
            TransactionServiceRequest::SubmitTransaction((tx_id, tx, fee, amount, message)) => self
                .submit_transaction(transaction_broadcast_join_handles, tx_id, tx, fee, amount, message)
                .await
                .map(|_| TransactionServiceResponse::TransactionSubmitted),
            TransactionServiceRequest::GenerateCoinbaseTransaction(reward, fees, block_height) => self
                .generate_coinbase_transaction(reward, fees, block_height, coinbase_monitoring_join_handles)
                .await
                .map(|tx| TransactionServiceResponse::CoinbaseTransactionGenerated(Box::new(tx))),
            #[cfg(feature = "test_harness")]
            TransactionServiceRequest::CompletePendingOutboundTransaction(completed_transaction) => {
                self.complete_pending_outbound_transaction(completed_transaction)
                    .await?;
                Ok(TransactionServiceResponse::CompletedPendingTransaction)
            },
            #[cfg(feature = "test_harness")]
            TransactionServiceRequest::FinalizePendingInboundTransaction(tx_id) => {
                self.finalize_received_test_transaction(tx_id).await?;
                Ok(TransactionServiceResponse::FinalizedPendingInboundTransaction)
            },
            #[cfg(feature = "test_harness")]
            TransactionServiceRequest::AcceptTestTransaction((tx_id, amount, source_pubkey)) => {
                self.receive_test_transaction(tx_id, amount, source_pubkey).await?;
                Ok(TransactionServiceResponse::AcceptedTestTransaction)
            },
            #[cfg(feature = "test_harness")]
            TransactionServiceRequest::BroadcastTransaction(tx_id) => {
                self.broadcast_transaction(tx_id).await?;
                Ok(TransactionServiceResponse::TransactionBroadcast)
            },
            #[cfg(feature = "test_harness")]
            TransactionServiceRequest::MineTransaction(tx_id) => {
                self.mine_transaction(tx_id).await?;
                Ok(TransactionServiceResponse::TransactionMined)
            },
            TransactionServiceRequest::SetLowPowerMode => {
                self.set_power_mode(PowerMode::Low).await?;
                Ok(TransactionServiceResponse::LowPowerModeSet)
            },
            TransactionServiceRequest::SetNormalPowerMode => {
                self.set_power_mode(PowerMode::Normal).await?;
                Ok(TransactionServiceResponse::NormalPowerModeSet)
            },
            TransactionServiceRequest::ApplyEncryption(cipher) => self
                .db
                .apply_encryption(*cipher)
                .await
                .map(|_| TransactionServiceResponse::EncryptionApplied)
                .map_err(TransactionServiceError::TransactionStorageError),
            TransactionServiceRequest::RemoveEncryption => self
                .db
                .remove_encryption()
                .await
                .map(|_| TransactionServiceResponse::EncryptionRemoved)
                .map_err(TransactionServiceError::TransactionStorageError),
            TransactionServiceRequest::RestartTransactionProtocols => self
                .restart_transaction_negotiation_protocols(
                    send_transaction_join_handles,
                    receive_transaction_join_handles,
                )
                .await
                .map(|_| TransactionServiceResponse::ProtocolsRestarted),

            TransactionServiceRequest::RestartBroadcastProtocols => self
                .restart_broadcast_protocols(
                    transaction_broadcast_join_handles,
                    chain_monitoring_join_handles,
                    coinbase_monitoring_join_handles,
                )
                .await
                .map(|_| TransactionServiceResponse::ProtocolsRestarted),
        }
    }
/// This handler is called when requests arrive from the various streams
This handler is called when requests arrive from the various streams
This handler is called when requests arrive from the various streams
async fn handle_request(
        &mut self,
        request: TransactionServiceRequest,
        send_transaction_join_handles: &mut FuturesUnordered<JoinHandle<Result<u64, TransactionServiceProtocolError>>>,
        receive_transaction_join_handles: &mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >,
        transaction_broadcast_join_handles: &mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >,
        chain_monitoring_join_handles: &mut FuturesUnordered<JoinHandle<Result<u64, TransactionServiceProtocolError>>>,
        coinbase_monitoring_join_handles: &mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >,
    ) -> Result<TransactionServiceResponse, TransactionServiceError>
    {
        trace!(target: LOG_TARGET, "Handling Service Request: {}", request);
        match request {
            TransactionServiceRequest::SendTransaction((dest_pubkey, amount, fee_per_gram, message)) => self
                .send_transaction(
                    dest_pubkey,
                    amount,
                    fee_per_gram,
                    message,
                    send_transaction_join_handles,
                )
                .await
                .map(TransactionServiceResponse::TransactionSent),
            TransactionServiceRequest::CancelTransaction(tx_id) => self
                .cancel_transaction(tx_id)
                .await
                .map(|_| TransactionServiceResponse::TransactionCancelled),
            TransactionServiceRequest::GetPendingInboundTransactions => {
                Ok(TransactionServiceResponse::PendingInboundTransactions(
                    self.db.get_pending_inbound_transactions().await?,
                ))
            },
            TransactionServiceRequest::GetPendingOutboundTransactions => {
                Ok(TransactionServiceResponse::PendingOutboundTransactions(
                    self.db.get_pending_outbound_transactions().await?,
                ))
            },
            TransactionServiceRequest::GetCompletedTransactions => Ok(
                TransactionServiceResponse::CompletedTransactions(self.db.get_completed_transactions().await?),
            ),
            TransactionServiceRequest::GetCancelledPendingInboundTransactions => {
                Ok(TransactionServiceResponse::PendingInboundTransactions(
                    self.db.get_cancelled_pending_inbound_transactions().await?,
                ))
            },
            TransactionServiceRequest::GetCancelledPendingOutboundTransactions => {
                Ok(TransactionServiceResponse::PendingOutboundTransactions(
                    self.db.get_cancelled_pending_outbound_transactions().await?,
                ))
            },
            TransactionServiceRequest::GetCancelledCompletedTransactions => {
                Ok(TransactionServiceResponse::CompletedTransactions(
                    self.db.get_cancelled_completed_transactions().await?,
                ))
            },
            TransactionServiceRequest::GetCompletedTransaction(tx_id) => {
                Ok(TransactionServiceResponse::CompletedTransaction(Box::new(
                    self.db.get_completed_transaction(tx_id).await?,
                )))
            },
            TransactionServiceRequest::GetAnyTransaction(tx_id) => Ok(TransactionServiceResponse::AnyTransaction(
                Box::new(self.db.get_any_transaction(tx_id).await?),
            )),
            TransactionServiceRequest::SetBaseNodePublicKey(public_key) => {
                self.set_base_node_public_key(public_key);
                Ok(TransactionServiceResponse::BaseNodePublicKeySet)
            },
            TransactionServiceRequest::ImportUtxo(value, source_public_key, message) => self
                .add_utxo_import_transaction(value, source_public_key, message)
                .await
                .map(TransactionServiceResponse::UtxoImported),
            TransactionServiceRequest::SubmitTransaction((tx_id, tx, fee, amount, message)) => self
                .submit_transaction(transaction_broadcast_join_handles, tx_id, tx, fee, amount, message)
                .await
                .map(|_| TransactionServiceResponse::TransactionSubmitted),
            TransactionServiceRequest::GenerateCoinbaseTransaction(reward, fees, block_height) => self
                .generate_coinbase_transaction(reward, fees, block_height, coinbase_monitoring_join_handles)
                .await
                .map(|tx| TransactionServiceResponse::CoinbaseTransactionGenerated(Box::new(tx))),
            #[cfg(feature = "test_harness")]
            TransactionServiceRequest::CompletePendingOutboundTransaction(completed_transaction) => {
                self.complete_pending_outbound_transaction(completed_transaction)
                    .await?;
                Ok(TransactionServiceResponse::CompletedPendingTransaction)
            },
            #[cfg(feature = "test_harness")]
            TransactionServiceRequest::FinalizePendingInboundTransaction(tx_id) => {
                self.finalize_received_test_transaction(tx_id).await?;
                Ok(TransactionServiceResponse::FinalizedPendingInboundTransaction)
            },
            #[cfg(feature = "test_harness")]
            TransactionServiceRequest::AcceptTestTransaction((tx_id, amount, source_pubkey)) => {
                self.receive_test_transaction(tx_id, amount, source_pubkey).await?;
                Ok(TransactionServiceResponse::AcceptedTestTransaction)
            },
            #[cfg(feature = "test_harness")]
            TransactionServiceRequest::BroadcastTransaction(tx_id) => {
                self.broadcast_transaction(tx_id).await?;
                Ok(TransactionServiceResponse::TransactionBroadcast)
            },
            #[cfg(feature = "test_harness")]
            TransactionServiceRequest::MineTransaction(tx_id) => {
                self.mine_transaction(tx_id).await?;
                Ok(TransactionServiceResponse::TransactionMined)
            },
            TransactionServiceRequest::SetLowPowerMode => {
                self.set_power_mode(PowerMode::Low).await?;
                Ok(TransactionServiceResponse::LowPowerModeSet)
            },
            TransactionServiceRequest::SetNormalPowerMode => {
                self.set_power_mode(PowerMode::Normal).await?;
                Ok(TransactionServiceResponse::NormalPowerModeSet)
            },
            TransactionServiceRequest::ApplyEncryption(cipher) => self
                .db
                .apply_encryption(*cipher)
                .await
                .map(|_| TransactionServiceResponse::EncryptionApplied)
                .map_err(TransactionServiceError::TransactionStorageError),
            TransactionServiceRequest::RemoveEncryption => self
                .db
                .remove_encryption()
                .await
                .map(|_| TransactionServiceResponse::EncryptionRemoved)
                .map_err(TransactionServiceError::TransactionStorageError),
            TransactionServiceRequest::RestartTransactionProtocols => self
                .restart_transaction_negotiation_protocols(
                    send_transaction_join_handles,
                    receive_transaction_join_handles,
                )
                .await
                .map(|_| TransactionServiceResponse::ProtocolsRestarted),
            TransactionServiceRequest::RestartBroadcastProtocols => self
                .restart_broadcast_protocols(
                    transaction_broadcast_join_handles,
                    chain_monitoring_join_handles,
                    coinbase_monitoring_join_handles,
                )
                .await
                .map(|_| TransactionServiceResponse::ProtocolsRestarted),
        }
    }
async fn handle_request ( & mut self , request : TransactionServiceRequest , send_transaction_join_handles : & mut FuturesUnordered < JoinHandle < Result < u64 , TransactionServiceProtocolError > > > , receive_transaction_join_handles : & mut FuturesUnordered < JoinHandle < Result < u64 , TransactionServiceProtocolError > > , > , transaction_broadcast_join_handles : & mut FuturesUnordered < JoinHandle < Result < u64 , TransactionServiceProtocolError > > , > , chain_monitoring_join_handles : & mut FuturesUnordered < JoinHandle < Result < u64 , TransactionServiceProtocolError > > > , coinbase_monitoring_join_handles : & mut FuturesUnordered < JoinHandle < Result < u64 , TransactionServiceProtocolError > > , > , ) -> Result < TransactionServiceResponse , TransactionServiceError > { trace ! ( target LOG_TARGET " " request ) ; match request { TransactionServiceRequest :: SendTransaction ( ( dest_pubkey , amount , fee_per_gram , message ) ) => self . send_transaction ( dest_pubkey , amount , fee_per_gram , message , send_transaction_join_handles , ) . await . map ( TransactionServiceResponse :: TransactionSent ) , TransactionServiceRequest :: CancelTransaction ( tx_id ) => self . cancel_transaction ( tx_id ) . await . map ( | _ | TransactionServiceResponse :: TransactionCancelled ) , TransactionServiceRequest :: GetPendingInboundTransactions => { Ok ( TransactionServiceResponse :: PendingInboundTransactions ( self . db . get_pending_inbound_transactions ( ) . await ? , ) ) } , TransactionServiceRequest :: GetPendingOutboundTransactions => { Ok ( TransactionServiceResponse :: PendingOutboundTransactions ( self . db . get_pending_outbound_transactions ( ) . await ? , ) ) } , TransactionServiceRequest :: GetCompletedTransactions => Ok ( TransactionServiceResponse :: CompletedTransactions ( self . db . get_completed_transactions ( ) . await ? ) , ) , TransactionServiceRequest :: GetCancelledPendingInboundTransactions => { Ok ( TransactionServiceResponse :: PendingInboundTransactions ( self . db . get_cancelled_pending_inbound_transactions ( ) . await ? , ) ) } , TransactionServiceRequest :: GetCancelledPendingOutboundTransactions => { Ok ( TransactionServiceResponse :: PendingOutboundTransactions ( self . db . get_cancelled_pending_outbound_transactions ( ) . await ? , ) ) } , TransactionServiceRequest :: GetCancelledCompletedTransactions => { Ok ( TransactionServiceResponse :: CompletedTransactions ( self . db . get_cancelled_completed_transactions ( ) . await ? , ) ) } , TransactionServiceRequest :: GetCompletedTransaction ( tx_id ) => { Ok ( TransactionServiceResponse :: CompletedTransaction ( Box :: new ( self . db . get_completed_transaction ( tx_id ) . await ? , ) ) ) } , TransactionServiceRequest :: GetAnyTransaction ( tx_id ) => Ok ( TransactionServiceResponse :: AnyTransaction ( Box :: new ( self . db . get_any_transaction ( tx_id ) . await ? ) , ) ) , TransactionServiceRequest :: SetBaseNodePublicKey ( public_key ) => { self . set_base_node_public_key ( public_key ) ; Ok ( TransactionServiceResponse :: BaseNodePublicKeySet ) } , TransactionServiceRequest :: ImportUtxo ( value , source_public_key , message ) => self . add_utxo_import_transaction ( value , source_public_key , message ) . await . map ( TransactionServiceResponse :: UtxoImported ) , TransactionServiceRequest :: SubmitTransaction ( ( tx_id , tx , fee , amount , message ) ) => self . submit_transaction ( transaction_broadcast_join_handles , tx_id , tx , fee , amount , message ) . await . map ( | _ | TransactionServiceResponse :: TransactionSubmitted ) , TransactionServiceRequest :: GenerateCoinbaseTransaction ( reward , fees , block_height ) => self . generate_coinbase_transaction ( reward , fees , block_height , coinbase_monitoring_join_handles ) . await . map ( | tx | TransactionServiceResponse :: CoinbaseTransactionGenerated ( Box :: new ( tx ) ) ) , # [ cfg ( feature " " ) ] TransactionServiceRequest :: CompletePendingOutboundTransaction ( completed_transaction ) => { self . complete_pending_outbound_transaction ( completed_transaction ) . await ? ; Ok ( TransactionServiceResponse :: CompletedPendingTransaction ) } , # [ cfg ( feature " " ) ] TransactionServiceRequest :: FinalizePendingInboundTransaction ( tx_id ) => { self . finalize_received_test_transaction ( tx_id ) . await ? ; Ok ( TransactionServiceResponse :: FinalizedPendingInboundTransaction ) } , # [ cfg ( feature " " ) ] TransactionServiceRequest :: AcceptTestTransaction ( ( tx_id , amount , source_pubkey ) ) => { self . receive_test_transaction ( tx_id , amount , source_pubkey ) . await ? ; Ok ( TransactionServiceResponse :: AcceptedTestTransaction ) } , # [ cfg ( feature " " ) ] TransactionServiceRequest :: BroadcastTransaction ( tx_id ) => { self . broadcast_transaction ( tx_id ) . await ? ; Ok ( TransactionServiceResponse :: TransactionBroadcast ) } , # [ cfg ( feature " " ) ] TransactionServiceRequest :: MineTransaction ( tx_id ) => { self . mine_transaction ( tx_id ) . await ? ; Ok ( TransactionServiceResponse :: TransactionMined ) } , TransactionServiceRequest :: SetLowPowerMode => { self . set_power_mode ( PowerMode :: Low ) . await ? ; Ok ( TransactionServiceResponse :: LowPowerModeSet ) } , TransactionServiceRequest :: SetNormalPowerMode => { self . set_power_mode ( PowerMode :: Normal ) . await ? ; Ok ( TransactionServiceResponse :: NormalPowerModeSet ) } , TransactionServiceRequest :: ApplyEncryption ( cipher ) => self . db . apply_encryption ( * cipher ) . await . map ( | _ | TransactionServiceResponse :: EncryptionApplied ) . map_err ( TransactionServiceError :: TransactionStorageError ) , TransactionServiceRequest :: RemoveEncryption => self . db . remove_encryption ( ) . await . map ( | _ | TransactionServiceResponse :: EncryptionRemoved ) . map_err ( TransactionServiceError :: TransactionStorageError ) , TransactionServiceRequest :: RestartTransactionProtocols => self . restart_transaction_negotiation_protocols ( send_transaction_join_handles , receive_transaction_join_handles , ) . await . map ( | _ | TransactionServiceResponse :: ProtocolsRestarted ) , TransactionServiceRequest :: RestartBroadcastProtocols => self . restart_broadcast_protocols ( transaction_broadcast_join_handles , chain_monitoring_join_handles , coinbase_monitoring_join_handles , ) . await . map ( | _ | TransactionServiceResponse :: ProtocolsRestarted ) , } }
This handler is called when requests arrive from the various streams
This handler is called when requests arrive from the various streams
&mut self
request
TransactionServiceRequest
send_transaction_join_handles
&mut FuturesUnordered<JoinHandle<Result<u64, TransactionServiceProtocolError>>>
receive_transaction_join_handles
&mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >
transaction_broadcast_join_handles
&mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >
chain_monitoring_join_handles
&mut FuturesUnordered<JoinHandle<Result<u64, TransactionServiceProtocolError>>>
coinbase_monitoring_join_handles
&mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >
&mut self
request
TransactionServiceRequest
send_transaction_join_handles
&mut FuturesUnordered<JoinHandle<Result<u64, TransactionServiceProtocolError>>>
receive_transaction_join_handles
&mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >
transaction_broadcast_join_handles
&mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >
chain_monitoring_join_handles
&mut FuturesUnordered<JoinHandle<Result<u64, TransactionServiceProtocolError>>>
coinbase_monitoring_join_handles
&mut FuturesUnordered<
            JoinHandle<Result<u64, TransactionServiceProtocolError>>,
        >