aeron_archiver_codecs/
replicate_request_codec.rs

1pub use decoder::ReplicateRequestDecoder;
2pub use encoder::ReplicateRequestEncoder;
3
4use crate::*;
5
6pub const SBE_BLOCK_LENGTH: u16 = 36;
7pub const SBE_TEMPLATE_ID: u16 = 50;
8pub const SBE_SCHEMA_ID: u16 = 101;
9pub const SBE_SCHEMA_VERSION: u16 = 7;
10pub const SBE_SEMANTIC_VERSION: &str = "5.2";
11
12pub mod encoder {
13    use message_header_codec::*;
14
15    use super::*;
16
17    #[derive(Debug, Default)]
18    pub struct ReplicateRequestEncoder<'a> {
19        buf: WriteBuf<'a>,
20        initial_offset: usize,
21        offset: usize,
22        limit: usize,
23    }
24
25    impl<'a> Writer<'a> for ReplicateRequestEncoder<'a> {
26        #[inline]
27        fn get_buf_mut(&mut self) -> &mut WriteBuf<'a> {
28            &mut self.buf
29        }
30    }
31
32    impl<'a> Encoder<'a> for ReplicateRequestEncoder<'a> {
33        #[inline]
34        fn get_limit(&self) -> usize {
35            self.limit
36        }
37
38        #[inline]
39        fn set_limit(&mut self, limit: usize) {
40            self.limit = limit;
41        }
42    }
43
44    impl<'a> ReplicateRequestEncoder<'a> {
45        pub fn wrap(mut self, buf: WriteBuf<'a>, offset: usize) -> Self {
46            let limit = offset + SBE_BLOCK_LENGTH as usize;
47            self.buf = buf;
48            self.initial_offset = offset;
49            self.offset = offset;
50            self.limit = limit;
51            self
52        }
53
54        #[inline]
55        pub fn encoded_length(&self) -> usize {
56            self.limit - self.offset
57        }
58
59        pub fn header(self, offset: usize) -> MessageHeaderEncoder<Self> {
60            let mut header = MessageHeaderEncoder::default().wrap(self, offset);
61            header.block_length(SBE_BLOCK_LENGTH);
62            header.template_id(SBE_TEMPLATE_ID);
63            header.schema_id(SBE_SCHEMA_ID);
64            header.version(SBE_SCHEMA_VERSION);
65            header
66        }
67
68        /// primitive field 'controlSessionId'
69        /// - min value: -9223372036854775807
70        /// - max value: 9223372036854775807
71        /// - null value: -9223372036854775808
72        /// - characterEncoding: null
73        /// - semanticType: null
74        /// - encodedOffset: 0
75        /// - encodedLength: 8
76        /// - version: 0
77        #[inline]
78        pub fn control_session_id(&mut self, value: i64) {
79            let offset = self.offset;
80            self.get_buf_mut().put_i64_at(offset, value);
81        }
82
83        /// primitive field 'correlationId'
84        /// - min value: -9223372036854775807
85        /// - max value: 9223372036854775807
86        /// - null value: -9223372036854775808
87        /// - characterEncoding: null
88        /// - semanticType: null
89        /// - encodedOffset: 8
90        /// - encodedLength: 8
91        /// - version: 0
92        #[inline]
93        pub fn correlation_id(&mut self, value: i64) {
94            let offset = self.offset + 8;
95            self.get_buf_mut().put_i64_at(offset, value);
96        }
97
98        /// primitive field 'srcRecordingId'
99        /// - min value: -9223372036854775807
100        /// - max value: 9223372036854775807
101        /// - null value: -9223372036854775808
102        /// - characterEncoding: null
103        /// - semanticType: null
104        /// - encodedOffset: 16
105        /// - encodedLength: 8
106        /// - version: 0
107        #[inline]
108        pub fn src_recording_id(&mut self, value: i64) {
109            let offset = self.offset + 16;
110            self.get_buf_mut().put_i64_at(offset, value);
111        }
112
113        /// primitive field 'dstRecordingId'
114        /// - min value: -9223372036854775807
115        /// - max value: 9223372036854775807
116        /// - null value: -9223372036854775808
117        /// - characterEncoding: null
118        /// - semanticType: null
119        /// - encodedOffset: 24
120        /// - encodedLength: 8
121        /// - version: 0
122        #[inline]
123        pub fn dst_recording_id(&mut self, value: i64) {
124            let offset = self.offset + 24;
125            self.get_buf_mut().put_i64_at(offset, value);
126        }
127
128        /// primitive field 'srcControlStreamId'
129        /// - min value: -2147483647
130        /// - max value: 2147483647
131        /// - null value: -2147483648
132        /// - characterEncoding: null
133        /// - semanticType: null
134        /// - encodedOffset: 32
135        /// - encodedLength: 4
136        /// - version: 0
137        #[inline]
138        pub fn src_control_stream_id(&mut self, value: i32) {
139            let offset = self.offset + 32;
140            self.get_buf_mut().put_i32_at(offset, value);
141        }
142
143        /// VAR_DATA ENCODER - character encoding: 'US-ASCII'
144        #[inline]
145        pub fn src_control_channel(&mut self, value: &[u8]) {
146            let limit = self.get_limit();
147            let data_length = value.len();
148            self.set_limit(limit + 4 + data_length);
149            self.get_buf_mut().put_u32_at(limit, data_length as u32);
150            self.get_buf_mut().put_slice_at(limit + 4, value);
151        }
152
153        /// VAR_DATA ENCODER - character encoding: 'US-ASCII'
154        #[inline]
155        pub fn live_destination(&mut self, value: &[u8]) {
156            let limit = self.get_limit();
157            let data_length = value.len();
158            self.set_limit(limit + 4 + data_length);
159            self.get_buf_mut().put_u32_at(limit, data_length as u32);
160            self.get_buf_mut().put_slice_at(limit + 4, value);
161        }
162    }
163} // end encoder
164
165pub mod decoder {
166    use message_header_codec::*;
167
168    use super::*;
169
170    #[derive(Clone, Copy, Debug, Default)]
171    pub struct ReplicateRequestDecoder<'a> {
172        buf: ReadBuf<'a>,
173        initial_offset: usize,
174        offset: usize,
175        limit: usize,
176        pub acting_block_length: u16,
177        pub acting_version: u16,
178    }
179
180    impl<'a> ActingVersion for ReplicateRequestDecoder<'a> {
181        #[inline]
182        fn acting_version(&self) -> u16 {
183            self.acting_version
184        }
185    }
186
187    impl<'a> Reader<'a> for ReplicateRequestDecoder<'a> {
188        #[inline]
189        fn get_buf(&self) -> &ReadBuf<'a> {
190            &self.buf
191        }
192    }
193
194    impl<'a> Decoder<'a> for ReplicateRequestDecoder<'a> {
195        #[inline]
196        fn get_limit(&self) -> usize {
197            self.limit
198        }
199
200        #[inline]
201        fn set_limit(&mut self, limit: usize) {
202            self.limit = limit;
203        }
204    }
205
206    impl<'a> ReplicateRequestDecoder<'a> {
207        pub fn wrap(mut self, buf: ReadBuf<'a>, offset: usize, acting_block_length: u16, acting_version: u16) -> Self {
208            let limit = offset + acting_block_length as usize;
209            self.buf = buf;
210            self.initial_offset = offset;
211            self.offset = offset;
212            self.limit = limit;
213            self.acting_block_length = acting_block_length;
214            self.acting_version = acting_version;
215            self
216        }
217
218        #[inline]
219        pub fn encoded_length(&self) -> usize {
220            self.limit - self.offset
221        }
222
223        pub fn header(self, mut header: MessageHeaderDecoder<ReadBuf<'a>>) -> Self {
224            debug_assert_eq!(SBE_TEMPLATE_ID, header.template_id());
225            let acting_block_length = header.block_length();
226            let acting_version = header.version();
227
228            self.wrap(
229                header.parent().unwrap(),
230                message_header_codec::ENCODED_LENGTH,
231                acting_block_length,
232                acting_version,
233            )
234        }
235
236        /// primitive field - 'REQUIRED'
237        #[inline]
238        pub fn control_session_id(&self) -> i64 {
239            self.get_buf().get_i64_at(self.offset)
240        }
241
242        /// primitive field - 'REQUIRED'
243        #[inline]
244        pub fn correlation_id(&self) -> i64 {
245            self.get_buf().get_i64_at(self.offset + 8)
246        }
247
248        /// primitive field - 'REQUIRED'
249        #[inline]
250        pub fn src_recording_id(&self) -> i64 {
251            self.get_buf().get_i64_at(self.offset + 16)
252        }
253
254        /// primitive field - 'REQUIRED'
255        #[inline]
256        pub fn dst_recording_id(&self) -> i64 {
257            self.get_buf().get_i64_at(self.offset + 24)
258        }
259
260        /// primitive field - 'REQUIRED'
261        #[inline]
262        pub fn src_control_stream_id(&self) -> i32 {
263            self.get_buf().get_i32_at(self.offset + 32)
264        }
265
266        /// VAR_DATA DECODER - character encoding: 'US-ASCII'
267        #[inline]
268        pub fn src_control_channel_decoder(&mut self) -> (usize, usize) {
269            let offset = self.get_limit();
270            let data_length = self.get_buf().get_u32_at(offset) as usize;
271            self.set_limit(offset + 4 + data_length);
272            (offset + 4, data_length)
273        }
274
275        #[inline]
276        pub fn src_control_channel_slice(&'a self, coordinates: (usize, usize)) -> &'a [u8] {
277            debug_assert!(self.get_limit() >= coordinates.0 + coordinates.1);
278            self.get_buf().get_slice_at(coordinates.0, coordinates.1)
279        }
280
281        /// VAR_DATA DECODER - character encoding: 'US-ASCII'
282        #[inline]
283        pub fn live_destination_decoder(&mut self) -> (usize, usize) {
284            let offset = self.get_limit();
285            let data_length = self.get_buf().get_u32_at(offset) as usize;
286            self.set_limit(offset + 4 + data_length);
287            (offset + 4, data_length)
288        }
289
290        #[inline]
291        pub fn live_destination_slice(&'a self, coordinates: (usize, usize)) -> &'a [u8] {
292            debug_assert!(self.get_limit() >= coordinates.0 + coordinates.1);
293            self.get_buf().get_slice_at(coordinates.0, coordinates.1)
294        }
295    }
296} // end decoder
aeron_archiver_codecs/replicate_request_codec.rs

aeron_archiver_codecs/
replicate_request_codec.rs
