camel_api/
resequencer.rs

1//! Resequencer EIP contract types.
2//!
3//! Batch/stream policy configs (Tasks 2/3).
4
5/// Window-based completion trigger for batch resequencing.
6///
7/// Narrowed to size and/or timeout — NOT the general `CompletionCondition`
8/// (which carries a `Predicate` variant semantically wrong for a
9/// resequencer window). Timeout values are in milliseconds.
10#[derive(Debug, Clone, PartialEq, Eq)]
11#[cfg_attr(feature = "schema", derive(schemars::JsonSchema, ts_rs::TS))]
12#[non_exhaustive]
13pub enum BatchCompletion {
14    /// Emit when `size` exchanges accumulate for a correlation key.
15    Size(usize),
16    /// Emit after `timeout_ms` since the first exchange for a correlation key.
17    Timeout(u64),
18    /// Emit when EITHER condition is met first.
19    SizeOrTimeout(usize, u64),
20}
21
22/// What to do when a stream resequencer gap timer fires.
23#[derive(Debug, Clone, Copy, PartialEq, Eq)]
24#[cfg_attr(feature = "schema", derive(schemars::JsonSchema, ts_rs::TS))]
25#[non_exhaustive]
26pub enum GapPolicy {
27    /// Emit the contiguous run from `next_expected` and advance past the gap.
28    EmitPartial,
29    /// Drop all held exchanges with a warning log (no dead-letter sink wired).
30    DropAndLog,
31}
32
33/// What to do when the stream resequencer priority queue reaches capacity.
34#[derive(Debug, Clone, Copy, PartialEq, Eq)]
35#[cfg_attr(feature = "schema", derive(schemars::JsonSchema, ts_rs::TS))]
36#[non_exhaustive]
37pub enum CapacityPolicy {
38    /// Log a warning and drop the incoming exchange (no dead-letter sink wired).
39    LogAndDrop,
40    /// Drop the oldest exchange from the queue to make room.
41    DropOldest,
42}
43
44/// Configurable resequencing policy mode.
45#[derive(Debug, Clone, PartialEq, Eq)]
46#[cfg_attr(feature = "schema", derive(schemars::JsonSchema, ts_rs::TS))]
47#[non_exhaustive]
48pub enum ResequenceMode {
49    /// Window-based batch resequencing.
50    Batch {
51        /// Simple-language expression for the correlation key
52        /// (e.g. `"${header.region}"`).
53        correlation: String,
54        /// Simple-language expression for the sort key
55        /// (e.g. `"${header.sequence}"`).
56        sort: String,
57        /// Window completion trigger.
58        completion: BatchCompletion,
59    },
60    /// Stream resequencing — bounded priority queue with gap detection.
61    Stream {
62        /// Simple-language expression for the sequence number
63        /// (e.g. `"${header.seqNum}"`). Must evaluate to a u64.
64        sequence: String,
65        /// Maximum queue size (default 1000).
66        capacity: usize,
67        /// Gap timeout in milliseconds (default 5000).
68        gap_timeout: u64,
69        /// What to do when a gap timer fires.
70        on_gap: GapPolicy,
71        /// What to do when the queue reaches capacity.
72        on_capacity_exceeded: CapacityPolicy,
73        /// When true, duplicate/late sequence numbers are ignored.
74        /// Default false (Camel 4.x behavior: duplicates are inserted).
75        dedup: bool,
76    },
77}
78
79/// Configuration for the Resequencer EIP.
80#[derive(Debug, Clone, PartialEq, Eq)]
81#[cfg_attr(feature = "schema", derive(schemars::JsonSchema, ts_rs::TS))]
82pub struct ResequencePolicyConfig {
83    pub mode: ResequenceMode,
84}
85
86impl Default for ResequencePolicyConfig {
87    /// Safe-ish defaults: batch mode by correlation + sort on `header.id`,
88    /// 100-size window with 30s timeout fallback.
89    fn default() -> Self {
90        Self {
91            mode: ResequenceMode::Batch {
92                correlation: "header.id".into(),
93                sort: "header.id".into(),
94                completion: BatchCompletion::SizeOrTimeout(100, 30_000),
95            },
96        }
97    }
98}
camel_api/resequencer.rs

camel_api/
resequencer.rs
