reifydb_core/interface/catalog/
queue.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_value::value::duration::Duration;
5use serde::{Deserialize, Serialize};
6
7use crate::{
8	common::TimeSource,
9	interface::catalog::{
10		column::Column,
11		id::{NamespaceId, QueueId},
12	},
13};
14
15#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
16pub struct Queue {
17	pub id: QueueId,
18	pub namespace: NamespaceId,
19	pub name: String,
20	pub columns: Vec<Column>,
21	pub dispatch: QueueDispatch,
22	pub deduplicate: Option<QueueDeduplicate>,
23	pub retention: QueueRetention,
24	pub retry: QueueRetry,
25	pub underlying: bool,
26	pub time: TimeSource,
27}
28
29impl Queue {
30	pub const DEFAULT_PARTITIONS: u16 = 16;
31	pub const MIN_PARTITIONS: u16 = 1;
32	pub const MAX_PARTITIONS: u16 = 1024;
33	pub const DEFAULT_RETRY_ATTEMPTS: u32 = 5;
34	pub const DEFAULT_RETRY_BACKOFF: Duration = Duration::from_seconds_const(10);
35
36	pub fn name(&self) -> &str {
37		&self.name
38	}
39
40	pub fn partitions(&self) -> u16 {
41		self.dispatch.partitions()
42	}
43
44	pub fn ordered_by(&self) -> Option<&str> {
45		self.dispatch.ordered_by()
46	}
47}
48
49#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
50pub enum QueueDispatch {
51	Fifo {
52		partitions: u16,
53		ordered_by: Option<String>,
54	},
55}
56
57impl QueueDispatch {
58	pub const TAG_FIFO: u8 = 0;
59
60	pub fn tag(&self) -> u8 {
61		match self {
62			Self::Fifo {
63				..
64			} => Self::TAG_FIFO,
65		}
66	}
67
68	pub fn partitions(&self) -> u16 {
69		match self {
70			Self::Fifo {
71				partitions,
72				..
73			} => *partitions,
74		}
75	}
76
77	pub fn ordered_by(&self) -> Option<&str> {
78		match self {
79			Self::Fifo {
80				ordered_by,
81				..
82			} => ordered_by.as_deref(),
83		}
84	}
85}
86
87impl Default for QueueDispatch {
88	fn default() -> Self {
89		Self::Fifo {
90			partitions: Queue::DEFAULT_PARTITIONS,
91			ordered_by: None,
92		}
93	}
94}
95
96#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
97pub struct QueueDeduplicate {
98	pub by: Vec<String>,
99	pub ttl: Duration,
100}
101
102impl QueueDeduplicate {
103	pub fn is_forever(&self) -> bool {
104		self.ttl == Duration::MAX
105	}
106}
107
108#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
109pub struct QueueRetention {
110	pub done: Option<Duration>,
111}
112
113#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
114pub struct QueueRetry {
115	pub attempts: u32,
116	pub backoff: Duration,
117}
118
119impl Default for QueueRetry {
120	fn default() -> Self {
121		Self {
122			attempts: Queue::DEFAULT_RETRY_ATTEMPTS,
123			backoff: Queue::DEFAULT_RETRY_BACKOFF,
124		}
125	}
126}
reifydb_core/interface/catalog/queue.rs

reifydb_core/interface/catalog/
queue.rs
