Struct OutboxManager 

pub struct OutboxManager<S, P, PT>
where
    PT: Debug + Clone + Serialize,
{ /* private fields */ }

Long-running worker that publishes pending outbox events to the broker.

The manager owns one concrete OutboxStorage implementation (S), one Transport implementation (P), and the user’s domain event payload type (PT). After run is invoked, it will keep processing events until the watched shutdown channel flips to true.

Construct one through OutboxManagerBuilder: it reports missing dependencies with a structured OutboxError::ConfigError and shields callers from the manager’s cfg-dependent constructor signature.

Implementations

impl<S, P, PT> OutboxManager<S, P, PT>

where S: OutboxStorage<PT> + Send + Sync + 'static, P: Transport<PT> + Send + Sync + 'static, PT: Debug + Clone + Serialize + Send + Sync + 'static,

pub async fn run(self) -> Result<(), OutboxError>

Starts the main outbox worker loop.

This method will run until a shutdown signal is received via the shutdown_rx channel. It coordinates three concerns:

• Event processing — on each wake-up it drives OutboxProcessor in an inner drain loop until the fetched batch is empty, at which point it returns to waiting.
• Wake-up sources — a tokio::select! races a storage-level LISTEN/notify call, a poll interval (config.poll_interval_secs), and the shutdown receiver. A notification error is logged and the loop sleeps for 5 seconds before retrying.
• Garbage collection — a background task is spawned that ticks on config.gc_interval_secs and calls [GarbageCollector::collect_garbage], exiting when the shutdown signal fires.

Errors

The Result is reserved for forward compatibility. In the current implementation run only returns Ok(()) — it returns when the shutdown signal flips to true. Transient storage and transport failures are logged via tracing::error! and the loop continues after a short back-off. A future revision may surface terminal errors (for example, an unrecoverable storage configuration) through the Err arm; treat Err as a hard stop if and when it appears.

Example

use std::sync::Arc;
use tokio::sync::watch;
use outbox_core::prelude::*;

let (shutdown_tx, shutdown_rx) = watch::channel(false);
let manager = OutboxManagerBuilder::new()
    .storage(storage)
    .publisher(publisher)
    .config(Arc::new(OutboxConfig::default()))
    .shutdown_rx(shutdown_rx)
    .build()?;

let handle = tokio::spawn(async move { manager.run().await });

// ... later, on a signal or process exit:
let _ = shutdown_tx.send(true);
handle.await.expect("worker panicked")?;