Struct ReactionBase 

pub struct ReactionBase {
    pub id: String,
    pub queries: Vec<String>,
    pub auto_start: bool,
    pub status: Arc<RwLock<ComponentStatus>>,
    pub priority_queue: PriorityQueue<QueryResult>,
    pub subscription_tasks: Arc<RwLock<Vec<JoinHandle<()>>>>,
    pub processing_task: Arc<RwLock<Option<JoinHandle<()>>>>,
    pub shutdown_tx: Arc<RwLock<Option<Sender<()>>>>,
    /* private fields */
}

Base implementation for common reaction functionality

Fields

id: String

Reaction identifier

queries: Vec<String>

List of query IDs to subscribe to

auto_start: bool

Whether this reaction should auto-start

status: Arc<RwLock<ComponentStatus>>

Current component status

priority_queue: PriorityQueue<QueryResult>

Priority queue for timestamp-ordered result processing

subscription_tasks: Arc<RwLock<Vec<JoinHandle<()>>>>

Handles to subscription forwarder tasks

processing_task: Arc<RwLock<Option<JoinHandle<()>>>>

Handle to the main processing task

shutdown_tx: Arc<RwLock<Option<Sender<()>>>>

Sender for shutdown signal to processing task

Implementations

impl ReactionBase

pub fn new(params: ReactionBaseParams) -> Self

Create a new ReactionBase with the given parameters

Dependencies (event channel, query subscriber, state store) are not required during construction - they will be provided via initialize() when the reaction is added to DrasiLib.

pub async fn initialize(&self, context: ReactionRuntimeContext)

Initialize the reaction with runtime context.

This method is called automatically by DrasiLib’s add_reaction() method. Plugin developers do not need to call this directly.

The context provides access to:

• reaction_id: The reaction’s unique identifier
• status_tx: Channel for reporting component status events
• state_store: Optional persistent state storage
• query_provider: Access to query instances for subscription

pub async fn context(&self) -> Option<ReactionRuntimeContext>

Get the runtime context if initialized.

Returns None if initialize() has not been called yet.

pub async fn state_store(&self) -> Option<Arc<dyn StateStoreProvider>>

Get the state store if configured.

Returns None if no state store was provided in the context.

pub fn get_auto_start(&self) -> bool

Get whether this reaction should auto-start

pub fn status_tx(&self) -> Arc<RwLock<Option<ComponentEventSender>>>

Get the status channel Arc for internal use by spawned tasks

This returns the internal status_tx wrapped in Arc<RwLock<Option<…>>> which allows background tasks to send component status events.

Returns a clone of the Arc that can be moved into spawned tasks.

pub fn clone_shared(&self) -> Self

Clone the ReactionBase with shared Arc references

This creates a new ReactionBase that shares the same underlying data through Arc references. Useful for passing to spawned tasks.

pub async fn create_shutdown_channel(&self) -> Receiver<()>

Create a shutdown channel and store the sender

Returns the receiver which should be passed to the processing task. The sender is stored internally and will be triggered by stop_common().

This should be called before spawning the processing task.

pub fn get_id(&self) -> &str

Get the reaction ID

pub fn get_queries(&self) -> &[String]

Get the query IDs

pub async fn get_status(&self) -> ComponentStatus

Get current status

pub async fn send_component_event( &self, status: ComponentStatus, message: Option<String>, ) -> Result<()>

Send a component lifecycle event

If the event channel has not been injected yet, this method silently succeeds without sending anything. This allows reactions to be used in a standalone fashion without DrasiLib if needed.

pub async fn set_status_with_event( &self, status: ComponentStatus, message: Option<String>, ) -> Result<()>

Transition to a new status and send event

pub async fn subscribe_to_queries(&self) -> Result<()>

Subscribe to all configured queries and spawn forwarder tasks

This method handles the common pattern of:

1. Getting query instances via the injected QueryProvider
2. Subscribing to each configured query
3. Spawning forwarder tasks to enqueue results to priority queue

Prerequisites

• inject_query_provider() must have been called (done automatically by DrasiLib)

Returns

• Ok(()) if all subscriptions succeeded
• Err(...) if QueryProvider not injected or any subscription failed

pub async fn stop_common(&self) -> Result<()>

Perform common cleanup operations

This method handles:

1. Sending shutdown signal to processing task (for graceful termination)
2. Aborting all subscription forwarder tasks
3. Waiting for or aborting the processing task
4. Draining the priority queue

pub async fn set_processing_task(&self, task: JoinHandle<()>)

Set the processing task handle