ursula_stream/
response.rs

1use serde::Deserialize;
2use serde::Serialize;
3use ursula_shard::BucketStreamId;
4
5use crate::model::ProducerRequest;
6use crate::record_index::StreamRecordRange;
7
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub enum StreamResponse {
10    BucketCreated {
11        bucket_id: String,
12    },
13    BucketAlreadyExists {
14        bucket_id: String,
15    },
16    BucketDeleted {
17        bucket_id: String,
18    },
19    Created {
20        stream_id: BucketStreamId,
21        next_offset: u64,
22        closed: bool,
23    },
24    AlreadyExists {
25        next_offset: u64,
26        closed: bool,
27        content_type: String,
28        stream_ttl_seconds: Option<u64>,
29        stream_expires_at_ms: Option<u64>,
30    },
31    Appended {
32        offset: u64,
33        next_offset: u64,
34        closed: bool,
35        deduplicated: bool,
36        producer: Option<ProducerRequest>,
37    },
38    Closed {
39        next_offset: u64,
40        deduplicated: bool,
41        producer: Option<ProducerRequest>,
42    },
43    Deleted,
44    ColdFlushed {
45        hot_start_offset: u64,
46    },
47    SnapshotPublished {
48        snapshot_offset: u64,
49        record_range: Option<StreamRecordRange>,
50    },
51    Accessed {
52        changed: bool,
53        expired: bool,
54    },
55    AttrsUpdated {
56        changed: bool,
57    },
58    ColdGcAcked {
59        removed: u64,
60    },
61    Error {
62        code: StreamErrorCode,
63        message: String,
64        next_offset: Option<u64>,
65        context: Vec<StreamErrorContext>,
66    },
67}
68
69#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
70pub enum StreamErrorCode {
71    InvalidBucketId,
72    InvalidStreamId,
73    BucketNotFound,
74    BucketNotEmpty,
75    StreamNotFound,
76    StreamGone,
77    StreamAlreadyExistsConflict,
78    MissingContentType,
79    ContentTypeMismatch,
80    EmptyAppend,
81    StreamClosed,
82    StreamSeqConflict,
83    InvalidProducer,
84    ProducerEpochStale,
85    ProducerSeqConflict,
86    InvalidRetention,
87    OffsetOutOfRange,
88    InvalidColdFlush,
89    InvalidSnapshot,
90    SnapshotNotFound,
91    SnapshotConflict,
92    InvalidStreamAttrs,
93    InvalidRecordBoundaries,
94    RecordPreconditionFailed,
95}
96
97#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
98pub enum StreamErrorContext {
99    StreamClosed,
100    StaleColdFlushCandidate,
101    ProducerEpochStale {
102        current_epoch: u64,
103    },
104    ProducerSeqConflict {
105        expected_seq: u64,
106        received_seq: u64,
107    },
108    RecordTailMismatch {
109        current_record: u64,
110    },
111}
112
113impl StreamResponse {
114    pub(crate) fn error(code: StreamErrorCode, message: impl Into<String>) -> Self {
115        Self::error_with_context(code, message, Vec::new())
116    }
117
118    pub(crate) fn error_with_context(
119        code: StreamErrorCode,
120        message: impl Into<String>,
121        context: Vec<StreamErrorContext>,
122    ) -> Self {
123        Self::Error {
124            code,
125            message: message.into(),
126            next_offset: None,
127            context,
128        }
129    }
130
131    pub(crate) fn error_with_next_offset(
132        code: StreamErrorCode,
133        message: impl Into<String>,
134        next_offset: u64,
135    ) -> Self {
136        Self::error_with_next_offset_and_context(code, message, next_offset, Vec::new())
137    }
138
139    pub(crate) fn error_with_next_offset_and_context(
140        code: StreamErrorCode,
141        message: impl Into<String>,
142        next_offset: u64,
143        context: Vec<StreamErrorContext>,
144    ) -> Self {
145        Self::Error {
146            code,
147            message: message.into(),
148            next_offset: Some(next_offset),
149            context,
150        }
151    }
152}
ursula_stream/response.rs

ursula_stream/
response.rs
