1use ::core::future::Future;
2
3use leb128_tokio::{
4 Leb128DecoderI16, Leb128DecoderI32, Leb128DecoderI64, Leb128DecoderU16, Leb128DecoderU32,
5 Leb128DecoderU64, Leb128Encoder,
6};
7use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _};
8use tokio_util::bytes::{Buf as _, BufMut as _, Bytes, BytesMut};
9use tokio_util::codec::{Decoder, Encoder};
10use utf8_tokio::Utf8Codec;
11
12use crate::CoreNameEncoder;
13
14macro_rules! ensure_capacity {
15 ($src:ident, $n:expr) => {
16 if let Some(n) = $n.checked_sub($src.len()) {
17 if n > 0 {
18 $src.reserve(n);
19 return Ok(None);
20 }
21 }
22 };
23}
24
25fn stateless_decode_eof<D>(dec: &mut D, buf: &mut BytesMut) -> Result<Option<D::Item>, D::Error>
26where
27 D: Decoder,
28{
29 match dec.decode(buf)? {
30 Some(frame) => Ok(Some(frame)),
31 None => {
32 if buf.is_empty() {
33 Ok(None)
34 } else {
35 Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof).into())
36 }
37 }
38 }
39}
40
41macro_rules! impl_encode_copy_ref {
42 ($enc:ident, $t:ty) => {
43 impl Encoder<&$t> for $enc {
44 type Error = std::io::Error;
45
46 #[cfg_attr(
47 feature = "tracing",
48 tracing::instrument(level = "trace", ret, fields(ty = stringify!($t)))
49 )]
50 fn encode(&mut self, item: &$t, dst: &mut BytesMut) -> Result<(), Self::Error> {
51 self.encode(*item, dst)
52 }
53 }
54
55 impl Encoder<&&$t> for $enc {
56 type Error = std::io::Error;
57
58 #[cfg_attr(
59 feature = "tracing",
60 tracing::instrument(level = "trace", ret, fields(ty = stringify!($t)))
61 )]
62 fn encode(&mut self, item: &&$t, dst: &mut BytesMut) -> Result<(), Self::Error> {
63 self.encode(**item, dst)
64 }
65 }
66 };
67}
68
69macro_rules! impl_encode_str {
70 ($enc:ident, $t:ty) => {
71 impl Encoder<$t> for $enc {
72 type Error = std::io::Error;
73
74 #[cfg_attr(
75 feature = "tracing",
76 tracing::instrument(level = "trace", ret, fields(ty = "string"))
77 )]
78 fn encode(&mut self, item: $t, dst: &mut BytesMut) -> Result<(), Self::Error> {
79 CoreNameEncoder.encode(item, dst)
80 }
81 }
82
83 impl Encoder<&$t> for $enc {
84 type Error = std::io::Error;
85
86 #[cfg_attr(
87 feature = "tracing",
88 tracing::instrument(level = "trace", ret, fields(ty = "string"))
89 )]
90 fn encode(&mut self, item: &$t, dst: &mut BytesMut) -> Result<(), Self::Error> {
91 CoreNameEncoder.encode(item, dst)
92 }
93 }
94 };
95}
96
97pub trait AsyncReadValue: AsyncRead {
98 #[cfg_attr(
99 feature = "tracing",
100 tracing::instrument(level = "trace", ret, skip_all, fields(ty = "bool"))
101 )]
102 fn read_bool(&mut self) -> impl Future<Output = std::io::Result<bool>>
103 where
104 Self: Unpin,
105 {
106 async move {
107 match self.read_u8().await? {
108 0 => Ok(false),
109 1 => Ok(true),
110 n => Err(std::io::Error::new(
111 std::io::ErrorKind::InvalidData,
112 format!("invalid bool value byte `{n}`"),
113 )),
114 }
115 }
116 }
117
118 #[cfg_attr(
119 feature = "tracing",
120 tracing::instrument(level = "trace", ret, skip_all, fields(ty = "option"))
121 )]
122 fn read_option_status(&mut self) -> impl Future<Output = std::io::Result<bool>>
123 where
124 Self: Unpin,
125 {
126 async move {
127 match self.read_u8().await? {
128 0 => Ok(false),
129 1 => Ok(true),
130 n => Err(std::io::Error::new(
131 std::io::ErrorKind::InvalidData,
132 format!("invalid option status byte value `{n}`"),
133 )),
134 }
135 }
136 }
137
138 #[cfg_attr(
139 feature = "tracing",
140 tracing::instrument(level = "trace", ret, skip_all, fields(ty = "result"))
141 )]
142 fn read_result_status(&mut self) -> impl Future<Output = std::io::Result<bool>>
143 where
144 Self: Unpin,
145 {
146 async move {
147 match self.read_u8().await? {
148 0 => Ok(true),
149 1 => Ok(false),
150 n => Err(std::io::Error::new(
151 std::io::ErrorKind::InvalidData,
152 format!("invalid result status byte value `{n}`"),
153 )),
154 }
155 }
156 }
157}
158
159impl<T: AsyncRead> AsyncReadValue for T {}
160
161pub trait AsyncWriteValue: AsyncWrite {
162 #[cfg_attr(
163 feature = "tracing",
164 tracing::instrument(level = "trace", ret, skip_all, fields(ty = "bool"))
165 )]
166 fn write_bool(&mut self, v: bool) -> impl Future<Output = std::io::Result<()>>
167 where
168 Self: Unpin,
169 {
170 async move { self.write_u8(v.into()).await }
171 }
172
173 #[cfg_attr(
174 feature = "tracing",
175 tracing::instrument(level = "trace", ret, skip_all, fields(ty = "option"))
176 )]
177 fn write_option_status<T>(&mut self, v: Option<T>) -> impl Future<Output = std::io::Result<()>>
178 where
179 Self: Unpin,
180 {
181 async move { self.write_u8(v.is_some().into()).await }
182 }
183
184 #[cfg_attr(
185 feature = "tracing",
186 tracing::instrument(level = "trace", ret, skip_all, fields(ty = "result"))
187 )]
188 fn write_result_status<T, E>(
189 &mut self,
190 v: Result<T, E>,
191 ) -> impl Future<Output = std::io::Result<()>>
192 where
193 Self: Unpin,
194 {
195 async move { self.write_u8(v.is_err().into()).await }
196 }
197}
198
199impl<T: AsyncWrite> AsyncWriteValue for T {}
200
201#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
202pub struct BoolCodec;
203
204impl Encoder<bool> for BoolCodec {
205 type Error = std::io::Error;
206
207 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
208 fn encode(&mut self, item: bool, dst: &mut BytesMut) -> Result<(), Self::Error> {
209 dst.reserve(1);
210 dst.put_u8(item.into());
211 Ok(())
212 }
213}
214
215impl_encode_copy_ref!(BoolCodec, bool);
216
217impl Decoder for BoolCodec {
218 type Item = bool;
219 type Error = std::io::Error;
220
221 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
222 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
223 ensure_capacity!(src, 1_usize);
224 match src.get_u8() {
225 0 => Ok(Some(false)),
226 1 => Ok(Some(true)),
227 n => Err(std::io::Error::new(
228 std::io::ErrorKind::InvalidData,
229 format!("invalid bool value byte `{n}`"),
230 )),
231 }
232 }
233}
234
235#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
236pub struct S8Codec;
237
238impl Encoder<i8> for S8Codec {
239 type Error = std::io::Error;
240
241 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
242 fn encode(&mut self, item: i8, dst: &mut BytesMut) -> Result<(), Self::Error> {
243 dst.reserve(1);
244 dst.put_i8(item);
245 Ok(())
246 }
247}
248
249impl_encode_copy_ref!(S8Codec, i8);
250
251impl Decoder for S8Codec {
252 type Item = i8;
253 type Error = std::io::Error;
254
255 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
256 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
257 ensure_capacity!(src, 1_usize);
258 Ok(Some(src.get_i8()))
259 }
260}
261
262#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
263pub struct U8Codec;
264
265impl Encoder<u8> for U8Codec {
266 type Error = std::io::Error;
267
268 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
269 fn encode(&mut self, item: u8, dst: &mut BytesMut) -> Result<(), Self::Error> {
270 dst.reserve(1);
271 dst.put_u8(item);
272 Ok(())
273 }
274}
275
276impl_encode_copy_ref!(U8Codec, u8);
277
278impl Decoder for U8Codec {
279 type Item = u8;
280 type Error = std::io::Error;
281
282 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
283 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
284 let Some(v) = src.first().copied() else {
285 return Ok(None);
286 };
287 src.advance(1);
288 Ok(Some(v))
289 }
290}
291
292#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
293pub struct S16Codec;
294
295impl Encoder<i16> for S16Codec {
296 type Error = std::io::Error;
297
298 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
299 fn encode(&mut self, item: i16, dst: &mut BytesMut) -> Result<(), Self::Error> {
300 Leb128Encoder.encode(item, dst)
301 }
302}
303
304impl_encode_copy_ref!(S16Codec, i16);
305
306impl Decoder for S16Codec {
307 type Item = i16;
308 type Error = std::io::Error;
309
310 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
311 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
312 Leb128DecoderI16.decode(src)
313 }
314
315 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
316 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
317 Leb128DecoderI16.decode_eof(src)
318 }
319}
320
321#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
322pub struct U16Codec;
323
324impl Encoder<u16> for U16Codec {
325 type Error = std::io::Error;
326
327 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
328 fn encode(&mut self, item: u16, dst: &mut BytesMut) -> Result<(), Self::Error> {
329 Leb128Encoder.encode(item, dst)
330 }
331}
332
333impl_encode_copy_ref!(U16Codec, u16);
334
335impl Decoder for U16Codec {
336 type Item = u16;
337 type Error = std::io::Error;
338
339 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
340 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
341 Leb128DecoderU16.decode(src)
342 }
343
344 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
345 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
346 Leb128DecoderU16.decode_eof(src)
347 }
348}
349
350#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
351pub struct S32Codec;
352
353impl Encoder<i32> for S32Codec {
354 type Error = std::io::Error;
355
356 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
357 fn encode(&mut self, item: i32, dst: &mut BytesMut) -> Result<(), Self::Error> {
358 Leb128Encoder.encode(item, dst)
359 }
360}
361
362impl_encode_copy_ref!(S32Codec, i32);
363
364impl Decoder for S32Codec {
365 type Item = i32;
366 type Error = std::io::Error;
367
368 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
369 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
370 Leb128DecoderI32.decode(src)
371 }
372
373 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
374 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
375 Leb128DecoderI32.decode_eof(src)
376 }
377}
378
379#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
380pub struct U32Codec;
381
382impl Encoder<u32> for U32Codec {
383 type Error = std::io::Error;
384
385 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
386 fn encode(&mut self, item: u32, dst: &mut BytesMut) -> Result<(), Self::Error> {
387 Leb128Encoder.encode(item, dst)
388 }
389}
390
391impl_encode_copy_ref!(U32Codec, u32);
392
393impl Decoder for U32Codec {
394 type Item = u32;
395 type Error = std::io::Error;
396
397 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
398 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
399 Leb128DecoderU32.decode(src)
400 }
401
402 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
403 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
404 Leb128DecoderU32.decode_eof(src)
405 }
406}
407
408#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
409pub struct S64Codec;
410
411impl Encoder<i64> for S64Codec {
412 type Error = std::io::Error;
413
414 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
415 fn encode(&mut self, item: i64, dst: &mut BytesMut) -> Result<(), Self::Error> {
416 Leb128Encoder.encode(item, dst)
417 }
418}
419
420impl_encode_copy_ref!(S64Codec, i64);
421
422impl Decoder for S64Codec {
423 type Item = i64;
424 type Error = std::io::Error;
425
426 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
427 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
428 Leb128DecoderI64.decode(src)
429 }
430
431 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
432 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
433 Leb128DecoderI64.decode_eof(src)
434 }
435}
436
437#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
438pub struct U64Codec;
439
440impl Encoder<u64> for U64Codec {
441 type Error = std::io::Error;
442
443 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
444 fn encode(&mut self, item: u64, dst: &mut BytesMut) -> Result<(), Self::Error> {
445 Leb128Encoder.encode(item, dst)
446 }
447}
448
449impl_encode_copy_ref!(U64Codec, u64);
450
451impl Decoder for U64Codec {
452 type Item = u64;
453 type Error = std::io::Error;
454
455 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
456 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
457 Leb128DecoderU64.decode(src)
458 }
459
460 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
461 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
462 Leb128DecoderU64.decode_eof(src)
463 }
464}
465
466#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
467pub struct F32Codec;
468
469impl Encoder<f32> for F32Codec {
470 type Error = std::io::Error;
471
472 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
473 fn encode(&mut self, item: f32, dst: &mut BytesMut) -> Result<(), Self::Error> {
474 dst.reserve(4);
475 dst.put_f32_le(item);
476 Ok(())
477 }
478}
479
480impl_encode_copy_ref!(F32Codec, f32);
481
482impl Decoder for F32Codec {
483 type Item = f32;
484 type Error = std::io::Error;
485
486 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
487 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
488 ensure_capacity!(src, 4_usize);
489 Ok(Some(src.get_f32_le()))
490 }
491
492 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
493 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
494 stateless_decode_eof(self, src)
495 }
496}
497
498#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
499pub struct F64Codec;
500
501impl Encoder<f64> for F64Codec {
502 type Error = std::io::Error;
503
504 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
505 fn encode(&mut self, item: f64, dst: &mut BytesMut) -> Result<(), Self::Error> {
506 dst.reserve(8);
507 dst.put_f64_le(item);
508 Ok(())
509 }
510}
511
512impl_encode_copy_ref!(F64Codec, f64);
513
514impl Decoder for F64Codec {
515 type Item = f64;
516 type Error = std::io::Error;
517
518 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
519 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
520 ensure_capacity!(src, 8_usize);
521 Ok(Some(src.get_f64_le()))
522 }
523
524 #[cfg_attr(feature = "tracing", tracing::instrument(level = "trace", ret))]
525 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
526 stateless_decode_eof(self, src)
527 }
528}
529
530#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
531pub struct PrimValEncoder;
532
533impl Encoder<bool> for PrimValEncoder {
534 type Error = std::io::Error;
535
536 fn encode(&mut self, item: bool, dst: &mut BytesMut) -> Result<(), Self::Error> {
537 BoolCodec.encode(item, dst)
538 }
539}
540
541impl Encoder<i8> for PrimValEncoder {
542 type Error = std::io::Error;
543
544 #[cfg_attr(
545 feature = "tracing",
546 tracing::instrument(level = "trace", ret, fields(ty = "s8"))
547 )]
548 fn encode(&mut self, item: i8, dst: &mut BytesMut) -> Result<(), Self::Error> {
549 S8Codec.encode(item, dst)
550 }
551}
552
553impl Encoder<u8> for PrimValEncoder {
554 type Error = std::io::Error;
555
556 #[cfg_attr(
557 feature = "tracing",
558 tracing::instrument(level = "trace", ret, fields(ty = "u8"))
559 )]
560 fn encode(&mut self, item: u8, dst: &mut BytesMut) -> Result<(), Self::Error> {
561 U8Codec.encode(item, dst)
562 }
563}
564
565impl Encoder<i16> for PrimValEncoder {
566 type Error = std::io::Error;
567
568 #[cfg_attr(
569 feature = "tracing",
570 tracing::instrument(level = "trace", ret, fields(ty = "s16"))
571 )]
572 fn encode(&mut self, item: i16, dst: &mut BytesMut) -> Result<(), Self::Error> {
573 S16Codec.encode(item, dst)
574 }
575}
576
577impl Encoder<u16> for PrimValEncoder {
578 type Error = std::io::Error;
579
580 #[cfg_attr(
581 feature = "tracing",
582 tracing::instrument(level = "trace", ret, fields(ty = "u16"))
583 )]
584 fn encode(&mut self, item: u16, dst: &mut BytesMut) -> Result<(), Self::Error> {
585 U16Codec.encode(item, dst)
586 }
587}
588
589impl Encoder<i32> for PrimValEncoder {
590 type Error = std::io::Error;
591
592 #[cfg_attr(
593 feature = "tracing",
594 tracing::instrument(level = "trace", ret, fields(ty = "s32"))
595 )]
596 fn encode(&mut self, item: i32, dst: &mut BytesMut) -> Result<(), Self::Error> {
597 S32Codec.encode(item, dst)
598 }
599}
600
601impl Encoder<u32> for PrimValEncoder {
602 type Error = std::io::Error;
603
604 #[cfg_attr(
605 feature = "tracing",
606 tracing::instrument(level = "trace", ret, fields(ty = "u32"))
607 )]
608 fn encode(&mut self, item: u32, dst: &mut BytesMut) -> Result<(), Self::Error> {
609 U32Codec.encode(item, dst)
610 }
611}
612
613impl Encoder<i64> for PrimValEncoder {
614 type Error = std::io::Error;
615
616 #[cfg_attr(
617 feature = "tracing",
618 tracing::instrument(level = "trace", ret, fields(ty = "s64"))
619 )]
620 fn encode(&mut self, item: i64, dst: &mut BytesMut) -> Result<(), Self::Error> {
621 S64Codec.encode(item, dst)
622 }
623}
624
625impl Encoder<u64> for PrimValEncoder {
626 type Error = std::io::Error;
627
628 #[cfg_attr(
629 feature = "tracing",
630 tracing::instrument(level = "trace", ret, fields(ty = "u64"))
631 )]
632 fn encode(&mut self, item: u64, dst: &mut BytesMut) -> Result<(), Self::Error> {
633 U64Codec.encode(item, dst)
634 }
635}
636
637impl Encoder<f32> for PrimValEncoder {
638 type Error = std::io::Error;
639
640 #[cfg_attr(
641 feature = "tracing",
642 tracing::instrument(level = "trace", ret, fields(ty = "f32"))
643 )]
644 fn encode(&mut self, item: f32, dst: &mut BytesMut) -> Result<(), Self::Error> {
645 F32Codec.encode(item, dst)
646 }
647}
648
649impl Encoder<f64> for PrimValEncoder {
650 type Error = std::io::Error;
651
652 #[cfg_attr(
653 feature = "tracing",
654 tracing::instrument(level = "trace", ret, fields(ty = "f64"))
655 )]
656 fn encode(&mut self, item: f64, dst: &mut BytesMut) -> Result<(), Self::Error> {
657 F64Codec.encode(item, dst)
658 }
659}
660
661impl Encoder<char> for PrimValEncoder {
662 type Error = std::io::Error;
663
664 #[cfg_attr(
665 feature = "tracing",
666 tracing::instrument(level = "trace", ret, fields(ty = "char"))
667 )]
668 fn encode(&mut self, item: char, dst: &mut BytesMut) -> Result<(), Self::Error> {
669 Utf8Codec.encode(item, dst)
670 }
671}
672
673#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
674pub struct FlagEncoder;
675
676impl Encoder<u8> for FlagEncoder {
677 type Error = std::io::Error;
678
679 #[cfg_attr(
680 feature = "tracing",
681 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "flags"))
682 )]
683 fn encode(&mut self, item: u8, dst: &mut BytesMut) -> Result<(), Self::Error> {
684 dst.reserve(1);
685 dst.put_u8(item);
686 Ok(())
687 }
688}
689
690impl Encoder<u16> for FlagEncoder {
691 type Error = std::io::Error;
692
693 #[cfg_attr(
694 feature = "tracing",
695 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "flags"))
696 )]
697 fn encode(&mut self, item: u16, dst: &mut BytesMut) -> Result<(), Self::Error> {
698 dst.reserve(2);
699 dst.put_u16_le(item);
700 Ok(())
701 }
702}
703
704impl Encoder<u32> for FlagEncoder {
705 type Error = std::io::Error;
706
707 #[cfg_attr(
708 feature = "tracing",
709 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "flags"))
710 )]
711 fn encode(&mut self, item: u32, dst: &mut BytesMut) -> Result<(), Self::Error> {
712 dst.reserve(4);
713 dst.put_u32_le(item);
714 Ok(())
715 }
716}
717
718impl Encoder<u64> for FlagEncoder {
719 type Error = std::io::Error;
720
721 #[cfg_attr(
722 feature = "tracing",
723 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "flags"))
724 )]
725 fn encode(&mut self, item: u64, dst: &mut BytesMut) -> Result<(), Self::Error> {
726 dst.reserve(8);
727 dst.put_u64_le(item);
728 Ok(())
729 }
730}
731
732impl Encoder<u128> for FlagEncoder {
733 type Error = std::io::Error;
734
735 #[cfg_attr(
736 feature = "tracing",
737 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "flags"))
738 )]
739 fn encode(&mut self, item: u128, dst: &mut BytesMut) -> Result<(), Self::Error> {
740 dst.reserve(16);
741 dst.put_u128_le(item);
742 Ok(())
743 }
744}
745
746impl Encoder<Vec<u8>> for FlagEncoder {
747 type Error = std::io::Error;
748
749 #[cfg_attr(
750 feature = "tracing",
751 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "flags"))
752 )]
753 fn encode(&mut self, item: Vec<u8>, dst: &mut BytesMut) -> Result<(), Self::Error> {
754 dst.extend_from_slice(&item);
755 Ok(())
756 }
757}
758
759impl Encoder<&[u8]> for FlagEncoder {
760 type Error = std::io::Error;
761
762 #[cfg_attr(
763 feature = "tracing",
764 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "flags"))
765 )]
766 fn encode(&mut self, item: &[u8], dst: &mut BytesMut) -> Result<(), Self::Error> {
767 dst.extend_from_slice(item);
768 Ok(())
769 }
770}
771
772impl Encoder<Bytes> for FlagEncoder {
773 type Error = std::io::Error;
774
775 #[cfg_attr(
776 feature = "tracing",
777 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "flags"))
778 )]
779 fn encode(&mut self, item: Bytes, dst: &mut BytesMut) -> Result<(), Self::Error> {
780 dst.extend_from_slice(&item);
781 Ok(())
782 }
783}
784
785impl Encoder<&Bytes> for FlagEncoder {
786 type Error = std::io::Error;
787
788 #[cfg_attr(
789 feature = "tracing",
790 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "flags"))
791 )]
792 fn encode(&mut self, item: &Bytes, dst: &mut BytesMut) -> Result<(), Self::Error> {
793 dst.extend_from_slice(item);
794 Ok(())
795 }
796}
797
798#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
799pub struct FlagDecoder<const N: usize>;
800
801impl<const N: usize> Decoder for FlagDecoder<N> {
802 type Item = Bytes;
803 type Error = std::io::Error;
804
805 #[cfg_attr(
806 feature = "tracing",
807 tracing::instrument(level = "trace", skip_all, fields(ty = "flags"))
808 )]
809 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
810 let n = N.div_ceil(8);
811 if src.len() < n {
812 ensure_capacity!(src, n);
813 }
814 Ok(Some(src.split_to(n).freeze()))
815 }
816
817 #[cfg_attr(
818 feature = "tracing",
819 tracing::instrument(level = "trace", skip_all, fields(ty = "flags"))
820 )]
821 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
822 stateless_decode_eof(self, src)
823 }
824}
825
826impl_encode_copy_ref!(PrimValEncoder, bool);
827impl_encode_copy_ref!(PrimValEncoder, i8);
828impl_encode_copy_ref!(PrimValEncoder, u8);
829impl_encode_copy_ref!(PrimValEncoder, i16);
830impl_encode_copy_ref!(PrimValEncoder, u16);
831impl_encode_copy_ref!(PrimValEncoder, i32);
832impl_encode_copy_ref!(PrimValEncoder, u32);
833impl_encode_copy_ref!(PrimValEncoder, i64);
834impl_encode_copy_ref!(PrimValEncoder, u64);
835impl_encode_copy_ref!(PrimValEncoder, f32);
836impl_encode_copy_ref!(PrimValEncoder, f64);
837impl_encode_copy_ref!(PrimValEncoder, char);
838
839impl_encode_copy_ref!(FlagEncoder, u8);
840impl_encode_copy_ref!(FlagEncoder, u16);
841impl_encode_copy_ref!(FlagEncoder, u32);
842impl_encode_copy_ref!(FlagEncoder, u64);
843impl_encode_copy_ref!(FlagEncoder, u128);
844
845impl_encode_str!(PrimValEncoder, &str);
846impl_encode_str!(PrimValEncoder, String);
847
848#[derive(Copy, Clone, Debug, Eq, PartialEq)]
849pub struct TupleEncoder<T>(pub T);
850
851#[derive(Debug)]
852pub struct TupleDecoder<C, V> {
853 dec: C,
854 v: V,
855 started: bool,
856}
857
858impl<C, V> TupleDecoder<C, V> {
859 pub fn into_inner(self) -> C {
860 self.dec
861 }
862}
863
864impl<C, V> TupleDecoder<C, V>
865where
866 V: Default,
867{
868 pub fn new(decoder: C) -> Self {
869 Self {
870 dec: decoder,
871 v: V::default(),
872 started: false,
873 }
874 }
875}
876
877macro_rules! impl_tuple_codec {
878 ($($vn:ident),+; $($vt:ident),+; $($cn:ident),+; $($ct:ident),+) => {
879 impl<$($ct),+> Default for TupleEncoder::<($($ct),+,)>
880 where
881 $($ct: Default),+
882 {
883 fn default() -> Self {
884 Self(($($ct::default()),+,))
885 }
886 }
887
888 impl<$($ct),+> From<($($ct),+,)> for TupleEncoder<($($ct),+,)> {
889 fn from(e: ($($ct),+,)) -> Self {
890 Self(e)
891 }
892 }
893
894 impl<E, $($vt, $ct),+> Encoder<($($vt),+,)> for TupleEncoder<($($ct),+,)>
895 where
896 E: From<std::io::Error>,
897 $($ct: Encoder<$vt, Error = E>),+
898 {
899 type Error = E;
900
901 #[cfg_attr(
902 feature = "tracing",
903 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "tuple"))
904 )]
905 fn encode(
906 &mut self,
907 ($($vn),+,): ($($vt),+,),
908 dst: &mut BytesMut,
909 ) -> Result<(), Self::Error> {
910 let ($(ref mut $cn),+,) = self.0;
911 $($cn.encode($vn, dst)?;)+
912 Ok(())
913 }
914 }
915
916 impl<'a, E, $($vt, $ct),+> Encoder<&'a ($($vt),+,)> for TupleEncoder<($($ct),+,)>
917 where
918 E: From<std::io::Error>,
919 $($ct: Encoder<&'a $vt, Error = E>),+
920 {
921 type Error = E;
922
923 #[cfg_attr(
924 feature = "tracing",
925 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "tuple"))
926 )]
927 fn encode(
928 &mut self,
929 ($($vn),+,): &'a ($($vt),+,),
930 dst: &mut BytesMut,
931 ) -> Result<(), Self::Error> {
932 let ($(ref mut $cn),+,) = self.0;
933 $($cn.encode($vn, dst)?;)+
934 Ok(())
935 }
936 }
937
938 impl<$($ct),+> Default for TupleDecoder<($($ct),+,), ($(Option<$ct::Item>),+,)>
939 where
940 $($ct: Decoder + Default),+,
941 {
942 fn default() -> Self {
943 Self{
944 dec: ($($ct::default()),+,),
945 v: ($(Option::<$ct::Item>::None),+,),
946 started: false,
947 }
948 }
949 }
950
951 impl<E, $($ct),+> Decoder for TupleDecoder<($($ct),+,), ($(Option<$ct::Item>),+,)>
952 where
953 E: From<std::io::Error>,
954 $($ct: Decoder<Error = E> + Default),+,
955 {
956 type Error = E;
957 type Item = ($($ct::Item),+,);
958
959 #[cfg_attr(
960 feature = "tracing",
961 tracing::instrument(level = "trace", skip(self), fields(ty = "tuple"))
962 )]
963 fn decode(
964 &mut self,
965 src: &mut BytesMut,
966 ) -> Result<Option<Self::Item>, Self::Error> {
967 self.decode_fields(src, false)
968 }
969
970 #[cfg_attr(
971 feature = "tracing",
972 tracing::instrument(level = "trace", skip(self), fields(ty = "tuple"))
973 )]
974 fn decode_eof(
975 &mut self,
976 src: &mut BytesMut,
977 ) -> Result<Option<Self::Item>, Self::Error> {
978 self.decode_fields(src, true)
979 }
980 }
981
982 impl<E, $($ct),+> TupleDecoder<($($ct),+,), ($(Option<$ct::Item>),+,)>
983 where
984 E: From<std::io::Error>,
985 $($ct: Decoder<Error = E> + Default),+,
986 {
987 fn reset(&mut self) {
988 self.dec = ($($ct::default()),+,);
989 self.v = ($(Option::<$ct::Item>::None),+,);
990 self.started = false;
991 }
992
993 fn decode_fields(
994 &mut self,
995 src: &mut BytesMut,
996 eof: bool,
997 ) -> Result<Option<($($ct::Item),+,)>, E> {
998 let ($(ref mut $vn),+,) = self.v;
999 let ($(ref mut $cn),+,) = self.dec;
1000 let n = src.len();
1001 let res: Result<Option<()>, E> = 'decode: {
1002 $(
1003 if $vn.is_none() {
1004 let v = if eof { $cn.decode_eof(src) } else { $cn.decode(src) };
1005 match v {
1006 Ok(Some(v)) => *$vn = Some(v),
1007 Ok(None) => break 'decode Ok(None),
1008 Err(err) => break 'decode Err(err),
1009 }
1010 }
1011 )+
1012 Ok(Some(()))
1013 };
1014 if src.len() < n {
1015 self.started = true;
1016 }
1017 match res {
1018 Ok(Some(())) if self.started => {
1019 self.started = false;
1020 Ok(Some(($($vn.take().unwrap()),+,)))
1021 }
1022 Ok(Some(())) => {
1023 self.reset();
1024 Err(std::io::Error::new(
1025 std::io::ErrorKind::InvalidData,
1026 "tuple decoded from zero bytes",
1027 )
1028 .into())
1029 }
1030 Ok(None) if !eof => Ok(None),
1031 Ok(None) => {
1032 let started = self.started;
1033 self.reset();
1034 if started || !src.is_empty() {
1035 Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof).into())
1036 } else {
1037 Ok(None)
1038 }
1039 }
1040 Err(err) => {
1041 self.reset();
1042 Err(err)
1043 }
1044 }
1045 }
1046 }
1047 };
1048}
1049
1050impl_tuple_codec!(
1051 v0;
1052 V0;
1053 c0;
1054 C0
1055);
1056
1057impl_tuple_codec!(
1058 v0, v1;
1059 V0, V1;
1060 c0, c1;
1061 C0, C1
1062);
1063
1064impl_tuple_codec!(
1065 v0, v1, v2;
1066 V0, V1, V2;
1067 c0, c1, c2;
1068 C0, C1, C2
1069);
1070
1071impl_tuple_codec!(
1072 v0, v1, v2, v3;
1073 V0, V1, V2, V3;
1074 c0, c1, c2, c3;
1075 C0, C1, C2, C3
1076);
1077
1078impl_tuple_codec!(
1079 v0, v1, v2, v3, v4;
1080 V0, V1, V2, V3, V4;
1081 c0, c1, c2, c3, c4;
1082 C0, C1, C2, C3, C4
1083);
1084
1085impl_tuple_codec!(
1086 v0, v1, v2, v3, v4, v5;
1087 V0, V1, V2, V3, V4, V5;
1088 c0, c1, c2, c3, c4, c5;
1089 C0, C1, C2, C3, C4, C5
1090);
1091
1092impl_tuple_codec!(
1093 v0, v1, v2, v3, v4, v5, v6;
1094 V0, V1, V2, V3, V4, V5, V6;
1095 c0, c1, c2, c3, c4, c5, c6;
1096 C0, C1, C2, C3, C4, C5, C6
1097);
1098
1099impl_tuple_codec!(
1100 v0, v1, v2, v3, v4, v5, v6, v7;
1101 V0, V1, V2, V3, V4, V5, V6, V7;
1102 c0, c1, c2, c3, c4, c5, c6, c7;
1103 C0, C1, C2, C3, C4, C5, C6, C7
1104);
1105
1106impl_tuple_codec!(
1107 v0, v1, v2, v3, v4, v5, v6, v7, v8;
1108 V0, V1, V2, V3, V4, V5, V6, V7, V8;
1109 c0, c1, c2, c3, c4, c5, c6, c7, c8;
1110 C0, C1, C2, C3, C4, C5, C6, C7, C8
1111);
1112
1113impl_tuple_codec!(
1114 v0, v1, v2, v3, v4, v5, v6, v7, v8, v9;
1115 V0, V1, V2, V3, V4, V5, V6, V7, V8, V9;
1116 c0, c1, c2, c3, c4, c5, c6, c7, c8, c9;
1117 C0, C1, C2, C3, C4, C5, C6, C7, C8, C9
1118);
1119
1120impl_tuple_codec!(
1121 v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10;
1122 V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10;
1123 c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10;
1124 C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10
1125);
1126
1127impl_tuple_codec!(
1128 v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11;
1129 V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10, V11;
1130 c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11;
1131 C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10, C11
1132);
1133
1134impl_tuple_codec!(
1135 v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12;
1136 V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10, V11, V12;
1137 c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12;
1138 C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10, C11, C12
1139);
1140
1141impl_tuple_codec!(
1142 v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13;
1143 V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10, V11, V12, V13;
1144 c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13;
1145 C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10, C11, C12, C13
1146);
1147
1148impl_tuple_codec!(
1149 v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14;
1150 V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10, V11, V12, V13, V14;
1151 c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13, c14;
1152 C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10, C11, C12, C13, C14
1153);
1154
1155impl_tuple_codec!(
1156 v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15;
1157 V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10, V11, V12, V13, V14, V15;
1158 c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13, c14, c15;
1159 C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10, C11, C12, C13, C14, C15
1160);
1161
1162#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
1163pub struct OptionEncoder<T>(pub T);
1164
1165impl<C, T> Encoder<Option<T>> for OptionEncoder<C>
1166where
1167 C: Encoder<T>,
1168{
1169 type Error = C::Error;
1170
1171 #[cfg_attr(
1172 feature = "tracing",
1173 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "option"))
1174 )]
1175 fn encode(&mut self, v: Option<T>, dst: &mut BytesMut) -> Result<(), Self::Error> {
1176 dst.reserve(1);
1177 if let Some(v) = v {
1178 dst.put_u8(1);
1179 self.0.encode(v, dst)
1180 } else {
1181 dst.put_u8(0);
1182 Ok(())
1183 }
1184 }
1185}
1186
1187impl<'a, C, T> Encoder<&'a Option<T>> for OptionEncoder<C>
1188where
1189 C: Encoder<&'a T>,
1190{
1191 type Error = C::Error;
1192
1193 #[cfg_attr(
1194 feature = "tracing",
1195 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "option"))
1196 )]
1197 fn encode(&mut self, v: &'a Option<T>, dst: &mut BytesMut) -> Result<(), Self::Error> {
1198 dst.reserve(1);
1199 if let Some(v) = v {
1200 dst.put_u8(1);
1201 self.0.encode(v, dst)
1202 } else {
1203 dst.put_u8(0);
1204 Ok(())
1205 }
1206 }
1207}
1208
1209#[derive(Debug, Default)]
1210pub struct OptionDecoder<T> {
1211 dec: T,
1212 is_some: bool,
1213}
1214
1215impl<T> OptionDecoder<T> {
1216 pub fn into_inner(self) -> T {
1217 self.dec
1218 }
1219}
1220
1221impl<T> OptionDecoder<T> {
1222 pub fn new(decoder: T) -> Self {
1223 Self {
1224 dec: decoder,
1225 is_some: false,
1226 }
1227 }
1228}
1229
1230impl<T> OptionDecoder<T>
1231where
1232 T: Decoder + Default,
1233{
1234 fn reset(&mut self) {
1235 self.dec = T::default();
1236 self.is_some = false;
1237 }
1238
1239 fn decode_value(
1240 &mut self,
1241 src: &mut BytesMut,
1242 eof: bool,
1243 ) -> Result<Option<Option<T::Item>>, T::Error> {
1244 if !self.is_some {
1245 ensure_capacity!(src, 1_usize);
1246 match src.get_u8() {
1247 0 => return Ok(Some(None)),
1248 1 => {
1249 self.is_some = true;
1250 }
1251 n => {
1252 return Err(std::io::Error::new(
1253 std::io::ErrorKind::InvalidData,
1254 format!("invalid option status byte value `{n}`"),
1255 )
1256 .into())
1257 }
1258 }
1259 }
1260 let res = if eof {
1261 self.dec.decode_eof(src)
1262 } else {
1263 self.dec.decode(src)
1264 };
1265 match res {
1266 Ok(Some(v)) => {
1267 self.is_some = false;
1268 Ok(Some(Some(v)))
1269 }
1270 Ok(None) => Ok(None),
1271 Err(err) => {
1272 self.reset();
1273 Err(err)
1274 }
1275 }
1276 }
1277}
1278
1279impl<T> Decoder for OptionDecoder<T>
1280where
1281 T: Decoder + Default,
1282{
1283 type Item = Option<T::Item>;
1284 type Error = T::Error;
1285
1286 #[cfg_attr(
1287 feature = "tracing",
1288 tracing::instrument(level = "trace", skip(self), fields(ty = "option"))
1289 )]
1290 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1291 self.decode_value(src, false)
1292 }
1293
1294 #[cfg_attr(
1295 feature = "tracing",
1296 tracing::instrument(level = "trace", skip(self), fields(ty = "option"))
1297 )]
1298 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1299 match self.decode_value(src, true)? {
1300 Some(frame) => Ok(Some(frame)),
1301 None if !self.is_some => Ok(None),
1302 None => {
1303 self.reset();
1304 Err(std::io::Error::from(std::io::ErrorKind::UnexpectedEof).into())
1305 }
1306 }
1307 }
1308}
1309
1310#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
1311pub struct ResultEncoder<O, E> {
1312 pub ok: O,
1313 pub err: E,
1314}
1315
1316impl<CO, O, CE, E> Encoder<Result<O, E>> for ResultEncoder<CO, CE>
1317where
1318 CO: Encoder<O>,
1319 CE: Encoder<E>,
1320 std::io::Error: From<CO::Error>,
1321 std::io::Error: From<CE::Error>,
1322{
1323 type Error = std::io::Error;
1324
1325 #[cfg_attr(
1326 feature = "tracing",
1327 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "result"))
1328 )]
1329 fn encode(&mut self, v: Result<O, E>, dst: &mut BytesMut) -> Result<(), Self::Error> {
1330 dst.reserve(1);
1331 match v {
1332 Ok(v) => {
1333 dst.put_u8(0);
1334 self.ok.encode(v, dst)?;
1335 }
1336 Err(v) => {
1337 dst.put_u8(1);
1338 self.err.encode(v, dst)?;
1339 }
1340 }
1341 Ok(())
1342 }
1343}
1344
1345impl<'a, CO, O, CE, E> Encoder<&'a Result<O, E>> for ResultEncoder<CO, CE>
1346where
1347 CO: Encoder<&'a O>,
1348 CE: Encoder<&'a E>,
1349 std::io::Error: From<CO::Error>,
1350 std::io::Error: From<CE::Error>,
1351{
1352 type Error = std::io::Error;
1353
1354 #[cfg_attr(
1355 feature = "tracing",
1356 tracing::instrument(level = "trace", skip_all, fields(dst, ty = "result"))
1357 )]
1358 fn encode(&mut self, v: &'a Result<O, E>, dst: &mut BytesMut) -> Result<(), Self::Error> {
1359 dst.reserve(1);
1360 match v {
1361 Ok(v) => {
1362 dst.put_u8(0);
1363 self.ok.encode(v, dst)?;
1364 }
1365 Err(v) => {
1366 dst.put_u8(1);
1367 self.err.encode(v, dst)?;
1368 }
1369 }
1370 Ok(())
1371 }
1372}
1373
1374#[derive(Debug, Default)]
1375pub struct ResultDecoder<O, E> {
1376 ok: O,
1377 err: E,
1378 is_ok: Option<bool>,
1379}
1380
1381impl<O, E> ResultDecoder<O, E> {
1382 pub fn into_inner(self) -> (O, E) {
1383 (self.ok, self.err)
1384 }
1385
1386 pub fn into_ok(self) -> O {
1387 self.ok
1388 }
1389
1390 pub fn into_err(self) -> E {
1391 self.err
1392 }
1393}
1394
1395impl<O, E> ResultDecoder<O, E> {
1396 pub fn new(ok: O, err: E) -> Self {
1397 Self {
1398 ok,
1399 err,
1400 is_ok: None,
1401 }
1402 }
1403}
1404
1405impl<O, E> ResultDecoder<O, E>
1406where
1407 O: Decoder + Default,
1408 E: Decoder + Default,
1409 std::io::Error: From<O::Error>,
1410 std::io::Error: From<E::Error>,
1411{
1412 fn reset(&mut self) {
1413 self.ok = O::default();
1414 self.err = E::default();
1415 self.is_ok = None;
1416 }
1417
1418 fn decode_value(
1419 &mut self,
1420 src: &mut BytesMut,
1421 eof: bool,
1422 ) -> std::io::Result<Option<Result<O::Item, E::Item>>> {
1423 let is_ok = if let Some(is_ok) = self.is_ok {
1424 is_ok
1425 } else {
1426 ensure_capacity!(src, 1_usize);
1427 match src.get_u8() {
1428 0 => {
1429 self.is_ok = Some(true);
1430 true
1431 }
1432 1 => {
1433 self.is_ok = Some(false);
1434 false
1435 }
1436 n => {
1437 return Err(std::io::Error::new(
1438 std::io::ErrorKind::InvalidData,
1439 format!("invalid result status byte value `{n}`"),
1440 ))
1441 }
1442 }
1443 };
1444 let res = if is_ok {
1445 let v = if eof {
1446 self.ok.decode_eof(src)
1447 } else {
1448 self.ok.decode(src)
1449 };
1450 v.map(|v| v.map(Ok)).map_err(Into::into)
1451 } else {
1452 let v = if eof {
1453 self.err.decode_eof(src)
1454 } else {
1455 self.err.decode(src)
1456 };
1457 v.map(|v| v.map(Err)).map_err(Into::into)
1458 };
1459 match res {
1460 Ok(Some(_)) => self.is_ok = None,
1461 Ok(None) => {}
1462 Err(_) => self.reset(),
1463 }
1464 res
1465 }
1466}
1467
1468impl<O, E> Decoder for ResultDecoder<O, E>
1469where
1470 O: Decoder + Default,
1471 E: Decoder + Default,
1472 std::io::Error: From<O::Error>,
1473 std::io::Error: From<E::Error>,
1474{
1475 type Item = Result<O::Item, E::Item>;
1476 type Error = std::io::Error;
1477
1478 #[cfg_attr(
1479 feature = "tracing",
1480 tracing::instrument(level = "trace", skip(self), fields(ty = "result"))
1481 )]
1482 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1483 self.decode_value(src, false)
1484 }
1485
1486 #[cfg_attr(
1487 feature = "tracing",
1488 tracing::instrument(level = "trace", skip(self), fields(ty = "result"))
1489 )]
1490 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1491 match self.decode_value(src, true)? {
1492 Some(frame) => Ok(Some(frame)),
1493 None if self.is_ok.is_none() => Ok(None),
1494 None => {
1495 self.reset();
1496 Err(std::io::ErrorKind::UnexpectedEof.into())
1497 }
1498 }
1499 }
1500}
1501
1502#[cfg(test)]
1503mod tests {
1504 use crate::{CoreNameDecoder, CoreVecDecoder};
1505
1506 use super::*;
1507
1508 #[test_log::test]
1509 fn tuple() {
1510 let mut buf = BytesMut::default();
1511 TupleEncoder((
1512 BoolCodec,
1513 PrimValEncoder,
1514 CoreNameEncoder,
1515 Leb128Encoder,
1516 TupleEncoder((ResultEncoder {
1517 ok: BoolCodec,
1518 err: CoreNameEncoder,
1519 },)),
1520 ))
1521 .encode(
1522 (
1523 true,
1524 0xfeu8,
1525 "test",
1526 0x42u32,
1527 (Result::<_, String>::Ok(true),),
1528 ),
1529 &mut buf,
1530 )
1531 .expect("failed to encode tuple");
1532 assert_eq!(buf.as_ref(), b"\x01\xfe\x04test\x42\0\x01");
1533 let (a, b, c, d, (e,)) = TupleDecoder::new((
1534 BoolCodec,
1535 U8Codec,
1536 <CoreNameDecoder>::default(),
1537 Leb128DecoderU32,
1538 TupleDecoder::<
1539 (ResultDecoder<BoolCodec, CoreNameDecoder>,),
1540 (Option<Result<bool, String>>,),
1541 >::default(),
1542 ))
1543 .decode(&mut buf)
1544 .expect("failed to decode tuple")
1545 .expect("short tuple read");
1546 assert!(a);
1547 assert_eq!(b, 0xfe);
1548 assert_eq!(c, "test");
1549 assert_eq!(d, 0x42);
1550 assert_eq!(e, Ok(true));
1551 }
1552
1553 #[test_log::test]
1554 fn option_reset_on_error() {
1555 let mut dec = OptionDecoder::<CoreNameDecoder>::default();
1556 let mut src = BytesMut::from(b"\x01\x01\xff".as_slice());
1557 dec.decode(&mut src).expect_err("invalid UTF-8 must fail");
1558 let mut src = BytesMut::from(b"\x01\x01b".as_slice());
1559 assert_eq!(
1560 dec.decode(&mut src).unwrap().unwrap(),
1561 Some("b".to_string())
1562 );
1563 }
1564
1565 #[test_log::test]
1566 fn result_reset_on_error() {
1567 let mut dec = ResultDecoder::<CoreNameDecoder, BoolCodec>::default();
1568 let mut src = BytesMut::from(b"\x00\x01\xff".as_slice());
1569 dec.decode(&mut src).expect_err("invalid UTF-8 must fail");
1570 let mut src = BytesMut::from(b"\x01\x01".as_slice());
1571 assert_eq!(dec.decode(&mut src).unwrap().unwrap(), Err(true));
1572 }
1573
1574 #[test_log::test]
1575 fn tuple_reset_on_error() {
1576 let mut dec =
1577 TupleDecoder::<(BoolCodec, CoreNameDecoder), (Option<bool>, Option<String>)>::default();
1578 let mut src = BytesMut::from(b"\x01\x01\xff".as_slice());
1579 dec.decode(&mut src).expect_err("invalid UTF-8 must fail");
1580 let mut src = BytesMut::from(b"\x00\x01b".as_slice());
1581 assert_eq!(
1582 dec.decode(&mut src).unwrap().unwrap(),
1583 (false, "b".to_string())
1584 );
1585 }
1586
1587 #[test_log::test]
1588 fn option_reset_on_eof() {
1589 let mut dec = OptionDecoder::<CoreNameDecoder>::default();
1590 let mut src = BytesMut::from(b"\x01\x02a".as_slice());
1591 let err = dec
1592 .decode_eof(&mut src)
1593 .expect_err("truncated option must fail");
1594 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1595 let mut src = BytesMut::from(b"\x01\x01b".as_slice());
1596 assert_eq!(
1597 dec.decode(&mut src).unwrap().unwrap(),
1598 Some("b".to_string())
1599 );
1600 assert!(dec.decode_eof(&mut BytesMut::new()).unwrap().is_none());
1601 }
1602
1603 #[test_log::test]
1604 fn result_reset_on_eof() {
1605 let mut dec = ResultDecoder::<CoreNameDecoder, BoolCodec>::default();
1606 let mut src = BytesMut::from(b"\x00".as_slice());
1607 let err = dec
1608 .decode_eof(&mut src)
1609 .expect_err("truncated result must fail");
1610 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1611 let mut src = BytesMut::from(b"\x01\x01".as_slice());
1612 assert_eq!(dec.decode(&mut src).unwrap().unwrap(), Err(true));
1613 }
1614
1615 #[test_log::test]
1616 fn tuple_reset_on_eof() {
1617 let mut dec = TupleDecoder::<
1618 (BoolCodec, OptionDecoder<CoreNameDecoder>),
1619 (Option<bool>, Option<Option<String>>),
1620 >::default();
1621 let mut src = BytesMut::from(b"\x01\x01".as_slice());
1622 let err = dec
1623 .decode_eof(&mut src)
1624 .expect_err("truncated tuple must fail");
1625 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1626 let mut src = BytesMut::from(b"\x00\x01\x01b".as_slice());
1627 assert_eq!(
1628 dec.decode(&mut src).unwrap().unwrap(),
1629 (false, Some("b".to_string()))
1630 );
1631 }
1632
1633 #[derive(Default)]
1634 struct EofDecoder;
1635
1636 impl Decoder for EofDecoder {
1637 type Item = Bytes;
1638 type Error = std::io::Error;
1639
1640 fn decode(&mut self, _: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1641 Ok(None)
1642 }
1643
1644 fn decode_eof(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1645 if src.is_empty() {
1646 Ok(None)
1647 } else {
1648 Ok(Some(src.split().freeze()))
1649 }
1650 }
1651 }
1652
1653 #[test_log::test]
1654 fn tuple_inner_eof_value() {
1655 let mut dec =
1656 TupleDecoder::<(BoolCodec, EofDecoder), (Option<bool>, Option<Bytes>)>::default();
1657 let mut src = BytesMut::from(b"\x01abc".as_slice());
1658 assert_eq!(
1659 dec.decode_eof(&mut src).unwrap().unwrap(),
1660 (true, Bytes::from_static(b"abc"))
1661 );
1662
1663 let mut dec =
1664 TupleDecoder::<(EofDecoder, BoolCodec), (Option<Bytes>, Option<bool>)>::default();
1665 let mut src = BytesMut::from(b"abc".as_slice());
1666 let err = dec
1667 .decode_eof(&mut src)
1668 .expect_err("truncated tuple must fail");
1669 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1670 }
1671
1672 #[derive(Default)]
1673 struct NoneDecoder;
1674
1675 impl Decoder for NoneDecoder {
1676 type Item = ();
1677 type Error = std::io::Error;
1678
1679 fn decode(&mut self, _: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1680 Ok(None)
1681 }
1682
1683 fn decode_eof(&mut self, _: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1684 Ok(None)
1685 }
1686 }
1687
1688 #[test_log::test]
1689 fn tuple_trailing_eof() {
1690 let mut dec =
1691 TupleDecoder::<(NoneDecoder, BoolCodec), (Option<()>, Option<bool>)>::default();
1692 let mut src = BytesMut::from(b"\x01".as_slice());
1693 let err = dec
1694 .decode_eof(&mut src)
1695 .expect_err("trailing bytes must fail");
1696 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1697 }
1698
1699 #[test_log::test]
1700 fn tuple_zero_size_eof() {
1701 let mut dec =
1702 TupleDecoder::<(FlagDecoder<0>, BoolCodec), (Option<Bytes>, Option<bool>)>::default();
1703 let mut src = BytesMut::default();
1704 assert_eq!(dec.decode_eof(&mut src).unwrap(), None);
1705 let mut src = BytesMut::from(b"\x01".as_slice());
1706 assert_eq!(
1707 dec.decode(&mut src).unwrap().unwrap(),
1708 (Bytes::default(), true)
1709 );
1710 }
1711
1712 #[test_log::test]
1713 fn tuple_zero_size() {
1714 let mut dec = TupleDecoder::<(FlagDecoder<0>,), (Option<Bytes>,)>::default();
1715 let err = dec
1716 .decode(&mut BytesMut::new())
1717 .expect_err("zero-size tuple must fail");
1718 assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
1719 let err = dec
1720 .decode_eof(&mut BytesMut::new())
1721 .expect_err("zero-size tuple must fail");
1722 assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
1723 }
1724
1725 #[derive(Default)]
1726 struct PairDecoder(Option<u8>);
1727
1728 impl Decoder for PairDecoder {
1729 type Item = [u8; 2];
1730 type Error = std::io::Error;
1731
1732 fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1733 while !src.is_empty() {
1734 let b = src.get_u8();
1735 if let Some(a) = self.0.take() {
1736 return Ok(Some([a, b]));
1737 }
1738 self.0 = Some(b);
1739 }
1740 Ok(None)
1741 }
1742 }
1743
1744 #[test_log::test]
1745 fn inner_reset_on_eof() {
1746 let mut dec = OptionDecoder::<PairDecoder>::default();
1747 let err = dec
1748 .decode_eof(&mut BytesMut::from(b"\x01a".as_slice()))
1749 .expect_err("truncated option must fail");
1750 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1751 let mut src = BytesMut::from(b"\x01bc".as_slice());
1752 assert_eq!(dec.decode(&mut src).unwrap().unwrap(), Some(*b"bc"));
1753
1754 let mut dec = ResultDecoder::<PairDecoder, BoolCodec>::default();
1755 let err = dec
1756 .decode_eof(&mut BytesMut::from(b"\x00a".as_slice()))
1757 .expect_err("truncated result must fail");
1758 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1759 let mut src = BytesMut::from(b"\x00bc".as_slice());
1760 assert_eq!(dec.decode(&mut src).unwrap().unwrap(), Ok(*b"bc"));
1761
1762 let mut dec = TupleDecoder::<(PairDecoder,), (Option<[u8; 2]>,)>::default();
1763 let err = dec
1764 .decode_eof(&mut BytesMut::from(b"a".as_slice()))
1765 .expect_err("truncated tuple must fail");
1766 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1767 let mut src = BytesMut::from(b"bc".as_slice());
1768 assert_eq!(dec.decode(&mut src).unwrap().unwrap(), (*b"bc",));
1769
1770 let mut dec = CoreVecDecoder::<PairDecoder>::default();
1771 let err = dec
1772 .decode_eof(&mut BytesMut::from(b"\x01a".as_slice()))
1773 .expect_err("truncated vec must fail");
1774 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1775 let mut src = BytesMut::from(b"\x01bc".as_slice());
1776 assert_eq!(dec.decode(&mut src).unwrap().unwrap(), [*b"bc"]);
1777 }
1778
1779 #[test_log::test]
1780 fn option_inner_eof_value() {
1781 let mut dec = OptionDecoder::<EofDecoder>::default();
1782 let mut src = BytesMut::from(b"\x01abc".as_slice());
1783 assert_eq!(
1784 dec.decode_eof(&mut src).unwrap().unwrap(),
1785 Some(Bytes::from_static(b"abc"))
1786 );
1787 }
1788
1789 #[test_log::test]
1790 fn result_inner_eof_value() {
1791 let mut dec = ResultDecoder::<EofDecoder, BoolCodec>::default();
1792 let mut src = BytesMut::from(b"\x00abc".as_slice());
1793 assert_eq!(
1794 dec.decode_eof(&mut src).unwrap().unwrap(),
1795 Ok(Bytes::from_static(b"abc"))
1796 );
1797 }
1798
1799 #[test_log::test]
1800 fn primitive_truncated_eof() {
1801 let mut src = BytesMut::from(b"\x80".as_slice());
1802 let err = U32Codec
1803 .decode_eof(&mut src)
1804 .expect_err("truncated u32 must fail");
1805 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1806
1807 let mut src = BytesMut::from(b"\x00\x00".as_slice());
1808 let err = F32Codec
1809 .decode_eof(&mut src)
1810 .expect_err("truncated f32 must fail");
1811 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1812
1813 let mut dec = TupleDecoder::<(BoolCodec, U32Codec), (Option<bool>, Option<u32>)>::default();
1814 let mut src = BytesMut::from(b"\x01\x80".as_slice());
1815 let err = dec
1816 .decode_eof(&mut src)
1817 .expect_err("truncated tuple must fail");
1818 assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
1819
1820 assert!(U32Codec.decode_eof(&mut BytesMut::new()).unwrap().is_none());
1821 }
1822}