Struct CacheService 

pub struct CacheService { /* private fields */ }

Outcome-aware Cache segment (Caching EIP).

Wraps a named CacheRepository and an on-miss sub-pipeline (OutcomeSegment). On each exchange:

1. Evaluate key_expr. None → not cacheable; forward directly to the on-miss sub-pipeline (no lookup, no write-back).
2. repository.get(&key):
   • Err(e) → Failed(e) (contract C1).
   • Ok(Some(entry)) → HIT: reconstruct Body from the entry, set it on the exchange, return Completed (skip on-miss).
   • Ok(None) → MISS: proceed to step 3.
3. Run the on-miss sub-pipeline.
   • Stopped(ex) / Failed(e) → propagate as-is (NO write-back).
   • Completed(ex) → proceed to write-back.
4. Write-back the resulting body (when it fits max_entry_bytes):
   • materialized variants (Bytes/Text/Json/Xml) → serialize, store.
   • Stream → materialize via Body::into_bytes (consumes the body, replaces it with Body::Bytes); StreamLimitExceeded propagates.
   • Empty / oversized body → pass through uncached, return Completed.

With coalesce_misses enabled (CacheService::with_coalesce), concurrent misses on the same resolved key are coalesced (singleflight): the first exchange (leader) runs on_miss and the single write-back set; concurrent exchanges (waiters) await the leader’s terminal state instead of running on_miss. HIT, key-None, and coalesce_misses == false paths bypass the in-flight map entirely.

Implementations

impl CacheService

pub fn new( repository: Arc<dyn CacheRepository>, key_expr: MessageIdExpression, ttl: Option<Duration>, max_entry_bytes: usize, on_miss: OutcomeSegment, rt: Arc<dyn RuntimeObservability>, ) -> Self

Build a new cache segment.

repository_name is derived from repository.name() so OTel tags stay in sync with the resolved backend.

pub fn with_coalesce(self, coalesce_misses: bool) -> Self

Enable (or explicitly disable) singleflight miss coalescing.

With coalescing on, concurrent misses on the same resolved key run the on_miss sub-pipeline exactly once per wave (leader runs + writes back; waiters receive the leader’s terminal state).

pub fn repository_name(&self) -> &str

The configured repository name (for OTel tagging).

Trait Implementations

impl Clone for CacheService

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl OutcomePipeline for CacheService

fn clone_box(&self) -> Box<dyn OutcomePipeline>

Clone the segment into a new boxed instance. Required because Box<dyn OutcomePipeline> cannot directly derive Clone.

fn run<'a>( &'a mut self, exchange: Exchange, ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>>

Execute the segment against exchange, returning a PipelineOutcome.