use serde::{Deserialize, Serialize};
use super::meta::{NextStreamEntryIdStrategy, StreamId};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[repr(u8)]
pub enum StreamTrimStrategy {
#[default]
None = 0,
MaxLen = 1,
MinId = 2,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct StreamTrimOptions {
pub strategy: StreamTrimStrategy,
pub max_len: u64,
pub min_id: StreamId,
pub limit: Option<usize>,
}
impl StreamTrimOptions {
pub fn none() -> Self {
Self::default()
}
pub fn maxlen(max_len: u64) -> Self {
Self {
strategy: StreamTrimStrategy::MaxLen,
max_len,
min_id: StreamId::min(),
limit: None,
}
}
pub fn minid(min_id: StreamId) -> Self {
Self {
strategy: StreamTrimStrategy::MinId,
max_len: 0,
min_id,
limit: None,
}
}
pub fn with_limit(mut self, limit: usize) -> Self {
self.limit = Some(limit);
self
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct StreamAddOptions {
pub trim_options: StreamTrimOptions,
pub next_id_strategy: NextStreamEntryIdStrategy,
pub nomkstream: bool,
}
impl Default for StreamAddOptions {
fn default() -> Self {
Self {
trim_options: StreamTrimOptions::none(),
next_id_strategy: NextStreamEntryIdStrategy::Auto,
nomkstream: false,
}
}
}
impl StreamAddOptions {
pub fn auto() -> Self {
Self::default()
}
pub fn with_id(id: StreamId) -> Self {
Self {
trim_options: StreamTrimOptions::none(),
next_id_strategy: NextStreamEntryIdStrategy::FullySpecified(id),
nomkstream: false,
}
}
pub fn with_strategy(strategy: NextStreamEntryIdStrategy) -> Self {
Self {
trim_options: StreamTrimOptions::none(),
next_id_strategy: strategy,
nomkstream: false,
}
}
pub fn with_trim(mut self, trim_options: StreamTrimOptions) -> Self {
self.trim_options = trim_options;
self
}
pub fn nomkstream(mut self, nomkstream: bool) -> Self {
self.nomkstream = nomkstream;
self
}
}
impl From<Option<StreamId>> for StreamAddOptions {
fn from(opt: Option<StreamId>) -> Self {
match opt {
Some(id) => Self::with_id(id),
None => Self::auto(),
}
}
}
impl From<StreamId> for StreamAddOptions {
fn from(id: StreamId) -> Self {
Self::with_id(id)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct StreamRangeOptions {
pub start: StreamId,
pub end: StreamId,
pub count: Option<usize>,
pub reverse: bool,
pub exclude_start: bool,
pub exclude_end: bool,
}
impl Default for StreamRangeOptions {
fn default() -> Self {
Self {
start: StreamId::min(),
end: StreamId::max(),
count: None,
reverse: false,
exclude_start: false,
exclude_end: false,
}
}
}
impl StreamRangeOptions {
pub fn new(start: StreamId, end: StreamId) -> Self {
Self {
start,
end,
count: None,
reverse: false,
exclude_start: false,
exclude_end: false,
}
}
pub fn reverse(start: StreamId, end: StreamId) -> Self {
Self {
start,
end,
count: None,
reverse: true,
exclude_start: false,
exclude_end: false,
}
}
pub fn with_count(mut self, count: usize) -> Self {
self.count = Some(count);
self
}
pub fn exclude_start(mut self, exclude: bool) -> Self {
self.exclude_start = exclude;
self
}
pub fn exclude_end(mut self, exclude: bool) -> Self {
self.exclude_end = exclude;
self
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct StreamLenOptions {
pub entry_id: StreamId,
pub with_entry_id: bool,
pub to_first: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct StreamXGroupCreateOptions {
pub mkstream: bool,
pub entries_read: Option<i64>,
pub last_id: String,
}
impl Default for StreamXGroupCreateOptions {
fn default() -> Self {
Self {
mkstream: false,
entries_read: None,
last_id: "$".to_string(),
}
}
}
impl StreamXGroupCreateOptions {
pub fn new(last_id: impl Into<String>) -> Self {
Self {
mkstream: false,
entries_read: None,
last_id: last_id.into(),
}
}
pub fn mkstream(mut self, mkstream: bool) -> Self {
self.mkstream = mkstream;
self
}
pub fn entries_read(mut self, entries_read: i64) -> Self {
self.entries_read = Some(entries_read);
self
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct StreamClaimOptions {
pub idle_time_ms: u64,
pub with_time: bool,
pub last_delivery_time_ms: u64,
pub with_retry_count: bool,
pub last_delivery_count: u64,
pub force: bool,
pub just_id: bool,
pub last_delivered_id: Option<StreamId>,
}
impl StreamClaimOptions {
pub fn new(idle_time_ms: u64) -> Self {
Self {
idle_time_ms,
..Default::default()
}
}
pub fn with_time(mut self, last_delivery_time_ms: u64) -> Self {
self.with_time = true;
self.last_delivery_time_ms = last_delivery_time_ms;
self
}
pub fn with_retry_count(mut self, last_delivery_count: u64) -> Self {
self.with_retry_count = true;
self.last_delivery_count = last_delivery_count;
self
}
pub fn force(mut self, force: bool) -> Self {
self.force = force;
self
}
pub fn just_id(mut self, just_id: bool) -> Self {
self.just_id = just_id;
self
}
pub fn with_last_id(mut self, last_delivered_id: StreamId) -> Self {
self.last_delivered_id = Some(last_delivered_id);
self
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct StreamAutoClaimOptions {
pub min_idle_time_ms: u64,
pub start_id: StreamId,
pub count: usize,
pub attempts_factors: usize,
pub just_id: bool,
pub exclude_start: bool,
}
impl Default for StreamAutoClaimOptions {
fn default() -> Self {
Self {
min_idle_time_ms: 0,
start_id: StreamId::min(),
count: 100,
attempts_factors: 10,
just_id: false,
exclude_start: false,
}
}
}
impl StreamAutoClaimOptions {
pub fn new(min_idle_time_ms: u64, start_id: StreamId) -> Self {
Self {
min_idle_time_ms,
start_id,
count: 100,
attempts_factors: 10,
just_id: false,
exclude_start: false,
}
}
pub fn count(mut self, count: usize) -> Self {
self.count = count;
self
}
pub fn just_id(mut self, just_id: bool) -> Self {
self.just_id = just_id;
self
}
pub fn exclude_start(mut self, exclude: bool) -> Self {
self.exclude_start = exclude;
self
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct StreamPendingOptions {
pub idle_time: u64,
pub with_time: bool,
pub start_id: StreamId,
pub end_id: StreamId,
pub exclude_start: bool,
pub exclude_end: bool,
pub count: Option<usize>,
pub consumer: Option<String>,
}
impl Default for StreamPendingOptions {
fn default() -> Self {
Self {
idle_time: 0,
with_time: false,
start_id: StreamId::min(),
end_id: StreamId::max(),
exclude_start: false,
exclude_end: false,
count: None,
consumer: None,
}
}
}
impl StreamPendingOptions {
pub fn summary() -> Self {
Self::default()
}
pub fn range(start_id: StreamId, end_id: StreamId, count: usize) -> Self {
Self {
idle_time: 0,
with_time: false,
start_id,
end_id,
exclude_start: false,
exclude_end: false,
count: Some(count),
consumer: None,
}
}
pub fn idle(mut self, idle_time: u64) -> Self {
self.with_time = true;
self.idle_time = idle_time;
self
}
pub fn consumer(mut self, consumer: impl Into<String>) -> Self {
self.consumer = Some(consumer.into());
self
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct StreamReadOptions {
pub count: Option<usize>,
pub block: Option<u64>,
pub noack: bool,
}
pub type XAdd = StreamAddOptions;
pub type XTrim = StreamTrimOptions;
pub type XRange = StreamRangeOptions;
pub type XRead = StreamReadOptions;
pub type XGroup = StreamXGroupCreateOptions;