ursula_stream/
response.rs

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

ursula_stream/
response.rs
