kaspa_index_processor/
service.rs

1use crate::{IDENT, processor::Processor};
2use kaspa_consensus_notify::{
3    connection::ConsensusChannelConnection, notification::Notification as ConsensusNotification, notifier::ConsensusNotifier,
4};
5use kaspa_core::{
6    task::service::{AsyncService, AsyncServiceError, AsyncServiceFuture},
7    trace, warn,
8};
9use kaspa_index_core::notifier::IndexNotifier;
10use kaspa_notify::{
11    connection::ChannelType,
12    events::{EventSwitches, EventType},
13    listener::ListenerLifespan,
14    scope::{PruningPointUtxoSetOverrideScope, UtxosChangedScope},
15    subscription::{MutationPolicies, UtxosChangedMutationPolicy, context::SubscriptionContext},
16};
17use kaspa_utils::{channel::Channel, triggers::SingleTrigger};
18use kaspa_utxoindex::api::UtxoIndexProxy;
19use std::sync::Arc;
20
21const INDEX_SERVICE: &str = IDENT;
22
23pub struct IndexService {
24    utxoindex: Option<UtxoIndexProxy>,
25    notifier: Arc<IndexNotifier>,
26    shutdown: SingleTrigger,
27}
28
29impl IndexService {
30    pub fn new(
31        consensus_notifier: &Arc<ConsensusNotifier>,
32        subscription_context: SubscriptionContext,
33        utxoindex: Option<UtxoIndexProxy>,
34    ) -> Self {
35        // This notifier UTXOs subscription granularity to consensus notifier
36        let policies = MutationPolicies::new(UtxosChangedMutationPolicy::Wildcard);
37
38        // Prepare consensus-notify objects
39        let consensus_notify_channel = Channel::<ConsensusNotification>::default();
40        let consensus_notify_listener_id = consensus_notifier.register_new_listener(
41            ConsensusChannelConnection::new(INDEX_SERVICE, consensus_notify_channel.sender(), ChannelType::Closable),
42            ListenerLifespan::Static(policies),
43        );
44
45        // Prepare the index-processor notifier
46        // No subscriber is defined here because the subscription are manually created during the construction and never changed after that.
47        let events: EventSwitches = [EventType::UtxosChanged, EventType::PruningPointUtxoSetOverride].as_ref().into();
48        let collector = Arc::new(Processor::new(utxoindex.clone(), consensus_notify_channel.receiver()));
49        let notifier = Arc::new(IndexNotifier::new(INDEX_SERVICE, events, vec![collector], vec![], subscription_context, 1, policies));
50
51        // Manually subscribe to index-processor related event types
52        consensus_notifier
53            .try_start_notify(consensus_notify_listener_id, UtxosChangedScope::default().into())
54            .expect("the subscription always succeeds");
55        consensus_notifier
56            .try_start_notify(consensus_notify_listener_id, PruningPointUtxoSetOverrideScope::default().into())
57            .expect("the subscription always succeeds");
58
59        Self { utxoindex, notifier, shutdown: SingleTrigger::default() }
60    }
61
62    pub fn notifier(&self) -> Arc<IndexNotifier> {
63        self.notifier.clone()
64    }
65
66    pub fn utxoindex(&self) -> Option<UtxoIndexProxy> {
67        self.utxoindex.clone()
68    }
69}
70
71impl AsyncService for IndexService {
72    fn ident(self: Arc<Self>) -> &'static str {
73        INDEX_SERVICE
74    }
75
76    fn start(self: Arc<Self>) -> AsyncServiceFuture {
77        trace!("{} starting", INDEX_SERVICE);
78
79        // Prepare a shutdown signal receiver
80        let shutdown_signal = self.shutdown.listener.clone();
81
82        // Launch the service and wait for a shutdown signal
83        Box::pin(async move {
84            self.notifier.clone().start();
85
86            // Keep the notifier running until a service shutdown signal is received
87            shutdown_signal.await;
88            match self.notifier.join().await {
89                Ok(_) => Ok(()),
90                Err(err) => {
91                    warn!("Error while stopping {}: {}", INDEX_SERVICE, err);
92                    Err(AsyncServiceError::Service(err.to_string()))
93                }
94            }
95        })
96    }
97
98    fn signal_exit(self: Arc<Self>) {
99        trace!("sending an exit signal to {}", INDEX_SERVICE);
100        self.shutdown.trigger.trigger();
101    }
102
103    fn stop(self: Arc<Self>) -> AsyncServiceFuture {
104        Box::pin(async move {
105            trace!("{} stopped", INDEX_SERVICE);
106            Ok(())
107        })
108    }
109}
kaspa_index_processor/service.rs

kaspa_index_processor/
service.rs
