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    ColdGcAcked {
65        removed: u64,
66    },
67    Error {
68        code: StreamErrorCode,
69        message: String,
70        next_offset: Option<u64>,
71        context: Vec<StreamErrorContext>,
72    },
73}
74
75#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
76pub enum StreamErrorCode {
77    InvalidBucketId,
78    InvalidStreamId,
79    BucketNotFound,
80    BucketNotEmpty,
81    StreamNotFound,
82    StreamGone,
83    StreamAlreadyExistsConflict,
84    MissingContentType,
85    ContentTypeMismatch,
86    EmptyAppend,
87    StreamClosed,
88    StreamSeqConflict,
89    InvalidProducer,
90    ProducerEpochStale,
91    ProducerSeqConflict,
92    InvalidRetention,
93    InvalidFork,
94    OffsetOutOfRange,
95    InvalidColdFlush,
96    InvalidSnapshot,
97    SnapshotNotFound,
98    SnapshotConflict,
99}
100
101#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
102pub enum StreamErrorContext {
103    StreamClosed,
104    StaleColdFlushCandidate,
105    ProducerEpochStale {
106        current_epoch: u64,
107    },
108    ProducerSeqConflict {
109        expected_seq: u64,
110        received_seq: u64,
111    },
112}
113
114impl StreamResponse {
115    pub(crate) fn error(code: StreamErrorCode, message: impl Into<String>) -> Self {
116        Self::error_with_context(code, message, Vec::new())
117    }
118
119    pub(crate) fn error_with_context(
120        code: StreamErrorCode,
121        message: impl Into<String>,
122        context: Vec<StreamErrorContext>,
123    ) -> Self {
124        Self::Error {
125            code,
126            message: message.into(),
127            next_offset: None,
128            context,
129        }
130    }
131
132    pub(crate) fn error_with_next_offset(
133        code: StreamErrorCode,
134        message: impl Into<String>,
135        next_offset: u64,
136    ) -> Self {
137        Self::error_with_next_offset_and_context(code, message, next_offset, Vec::new())
138    }
139
140    pub(crate) fn error_with_next_offset_and_context(
141        code: StreamErrorCode,
142        message: impl Into<String>,
143        next_offset: u64,
144        context: Vec<StreamErrorContext>,
145    ) -> Self {
146        Self::Error {
147            code,
148            message: message.into(),
149            next_offset: Some(next_offset),
150            context,
151        }
152    }
153}
ursula_stream/response.rs

ursula_stream/
response.rs
