wrpc_transport/
value.rs

1use core::any::TypeId;
2use core::fmt::{self, Debug};
3use core::future::{Future, pending};
4use core::hash::{Hash, Hasher};
5use core::iter::zip;
6use core::marker::PhantomData;
7use core::mem;
8use core::ops::{Deref, DerefMut};
9use core::pin::Pin;
10
11use bytes::{Buf as _, BufMut as _, Bytes, BytesMut};
12use futures::stream::{self, FuturesUnordered};
13use futures::{Stream, StreamExt as _, TryStreamExt as _};
14use tokio::io::{AsyncRead, AsyncReadExt as _, AsyncWrite, AsyncWriteExt as _};
15use tokio::select;
16use tokio::sync::{mpsc, oneshot};
17use tokio::task::JoinSet;
18use tokio_stream::wrappers::ReceiverStream;
19use tokio_util::codec::{Encoder as _, FramedRead};
20use tokio_util::io::StreamReader;
21use tracing::{Instrument as _, Span, debug, error, instrument, trace};
22use wasm_tokio::cm::{
23    BoolCodec, F32Codec, F64Codec, OptionDecoder, OptionEncoder, PrimValEncoder, ResultDecoder,
24    ResultEncoder, S8Codec, S16Codec, S32Codec, S64Codec, TupleDecoder, TupleEncoder, U8Codec,
25    U16Codec, U32Codec, U64Codec,
26};
27use wasm_tokio::{
28    CoreNameDecoder, CoreNameEncoder, CoreVecDecoder, CoreVecDecoderBytes, CoreVecEncoderBytes,
29    Leb128DecoderI8, Leb128DecoderI16, Leb128DecoderI32, Leb128DecoderI64, Leb128DecoderI128,
30    Leb128DecoderU8, Leb128DecoderU16, Leb128DecoderU32, Leb128DecoderU64, Leb128DecoderU128,
31    Leb128Encoder, Utf8Codec,
32};
33
34use crate::BufferedIncoming;
35use crate::frame::Outgoing;
36
37/// Borrowed resource handle, represented as an opaque byte blob
38#[repr(transparent)]
39pub struct ResourceBorrow<T: ?Sized> {
40    repr: Bytes,
41    _ty: PhantomData<T>,
42}
43
44impl<T: ?Sized> From<Bytes> for ResourceBorrow<T> {
45    fn from(repr: Bytes) -> Self {
46        Self {
47            repr,
48            _ty: PhantomData,
49        }
50    }
51}
52
53impl<T: ?Sized> From<Vec<u8>> for ResourceBorrow<T> {
54    fn from(repr: Vec<u8>) -> Self {
55        Self {
56            repr: repr.into(),
57            _ty: PhantomData,
58        }
59    }
60}
61
62impl<T: ?Sized> From<ResourceBorrow<T>> for Bytes {
63    fn from(ResourceBorrow { repr, .. }: ResourceBorrow<T>) -> Self {
64        repr
65    }
66}
67
68impl<T: ?Sized> PartialEq for ResourceBorrow<T> {
69    fn eq(&self, other: &Self) -> bool {
70        self.repr == other.repr
71    }
72}
73
74impl<T: ?Sized> Eq for ResourceBorrow<T> {}
75
76impl<T: ?Sized> Hash for ResourceBorrow<T> {
77    fn hash<H: Hasher>(&self, state: &mut H) {
78        self.repr.hash(state);
79    }
80}
81
82impl<T: ?Sized> AsRef<[u8]> for ResourceBorrow<T> {
83    fn as_ref(&self) -> &[u8] {
84        &self.repr
85    }
86}
87
88impl<T: ?Sized> AsRef<Bytes> for ResourceBorrow<T> {
89    fn as_ref(&self) -> &Bytes {
90        &self.repr
91    }
92}
93
94impl<T: ?Sized + 'static> Debug for ResourceBorrow<T> {
95    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
96        write!(f, "borrow<{:?}>", TypeId::of::<T>())
97    }
98}
99
100impl<T: ?Sized> Clone for ResourceBorrow<T> {
101    fn clone(&self) -> Self {
102        Self {
103            repr: self.repr.clone(),
104            _ty: PhantomData,
105        }
106    }
107}
108
109impl<T: ?Sized> ResourceBorrow<T> {
110    /// Constructs a new borrowed resource handle
111    pub fn new(repr: impl Into<Bytes>) -> Self {
112        Self::from(repr.into())
113    }
114}
115
116/// Owned resource handle, represented as an opaque byte blob
117#[repr(transparent)]
118pub struct ResourceOwn<T: ?Sized> {
119    repr: Bytes,
120    _ty: PhantomData<T>,
121}
122
123impl<T: ?Sized> From<ResourceOwn<T>> for ResourceBorrow<T> {
124    fn from(ResourceOwn { repr, _ty }: ResourceOwn<T>) -> Self {
125        Self {
126            repr,
127            _ty: PhantomData,
128        }
129    }
130}
131
132impl<T: ?Sized> From<Bytes> for ResourceOwn<T> {
133    fn from(repr: Bytes) -> Self {
134        Self {
135            repr,
136            _ty: PhantomData,
137        }
138    }
139}
140
141impl<T: ?Sized> From<Vec<u8>> for ResourceOwn<T> {
142    fn from(repr: Vec<u8>) -> Self {
143        Self {
144            repr: repr.into(),
145            _ty: PhantomData,
146        }
147    }
148}
149
150impl<T: ?Sized> From<ResourceOwn<T>> for Bytes {
151    fn from(ResourceOwn { repr, .. }: ResourceOwn<T>) -> Self {
152        repr
153    }
154}
155
156impl<T: ?Sized> PartialEq for ResourceOwn<T> {
157    fn eq(&self, other: &Self) -> bool {
158        self.repr == other.repr
159    }
160}
161
162impl<T: ?Sized> Eq for ResourceOwn<T> {}
163
164impl<T: ?Sized> Hash for ResourceOwn<T> {
165    fn hash<H: Hasher>(&self, state: &mut H) {
166        self.repr.hash(state);
167    }
168}
169
170impl<T: ?Sized> AsRef<[u8]> for ResourceOwn<T> {
171    fn as_ref(&self) -> &[u8] {
172        &self.repr
173    }
174}
175
176impl<T: ?Sized> AsRef<Bytes> for ResourceOwn<T> {
177    fn as_ref(&self) -> &Bytes {
178        &self.repr
179    }
180}
181
182impl<T: ?Sized + 'static> Debug for ResourceOwn<T> {
183    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
184        write!(f, "own<{:?}>", TypeId::of::<T>())
185    }
186}
187
188impl<T: ?Sized> Clone for ResourceOwn<T> {
189    fn clone(&self) -> Self {
190        Self {
191            repr: self.repr.clone(),
192            _ty: PhantomData,
193        }
194    }
195}
196
197impl<T: ?Sized> ResourceOwn<T> {
198    /// Constructs a new owned resource handle
199    pub fn new(repr: impl Into<Bytes>) -> Self {
200        Self::from(repr.into())
201    }
202
203    /// Returns the owned handle as [`ResourceBorrow`]
204    pub fn as_borrow(&self) -> ResourceBorrow<T> {
205        ResourceBorrow {
206            repr: self.repr.clone(),
207            _ty: PhantomData,
208        }
209    }
210}
211
212/// Deferred operation used for async value processing
213pub type DeferredFn<T> = Box<
214    dyn FnOnce(T, Vec<usize>) -> Pin<Box<dyn Future<Output = std::io::Result<()>> + Send>> + Send,
215>;
216
217/// Handles async processing state for codecs
218pub trait Deferred<T> {
219    /// Takes a deferred async processing operation, if any
220    fn take_deferred(&mut self) -> Option<DeferredFn<T>>;
221}
222
223macro_rules! impl_deferred_sync {
224    ($t:ty) => {
225        impl<T> Deferred<T> for $t {
226            fn take_deferred(&mut self) -> Option<DeferredFn<T>> {
227                None
228            }
229        }
230    };
231}
232
233impl_deferred_sync!(BoolCodec);
234impl_deferred_sync!(S8Codec);
235impl_deferred_sync!(U8Codec);
236impl_deferred_sync!(S16Codec);
237impl_deferred_sync!(U16Codec);
238impl_deferred_sync!(S32Codec);
239impl_deferred_sync!(U32Codec);
240impl_deferred_sync!(S64Codec);
241impl_deferred_sync!(U64Codec);
242impl_deferred_sync!(F32Codec);
243impl_deferred_sync!(F64Codec);
244impl_deferred_sync!(CoreNameDecoder);
245impl_deferred_sync!(CoreNameEncoder);
246impl_deferred_sync!(CoreVecDecoderBytes);
247impl_deferred_sync!(CoreVecEncoderBytes);
248impl_deferred_sync!(Utf8Codec);
249impl_deferred_sync!(PrimValEncoder);
250impl_deferred_sync!(Leb128Encoder);
251impl_deferred_sync!(Leb128DecoderI8);
252impl_deferred_sync!(Leb128DecoderU8);
253impl_deferred_sync!(Leb128DecoderI16);
254impl_deferred_sync!(Leb128DecoderU16);
255impl_deferred_sync!(Leb128DecoderI32);
256impl_deferred_sync!(Leb128DecoderU32);
257impl_deferred_sync!(Leb128DecoderI64);
258impl_deferred_sync!(Leb128DecoderU64);
259impl_deferred_sync!(Leb128DecoderI128);
260impl_deferred_sync!(Leb128DecoderU128);
261impl_deferred_sync!(ResourceEncoder);
262impl_deferred_sync!(UnitCodec);
263impl_deferred_sync!(ListDecoderU8);
264
265impl_deferred_sync!(CoreVecDecoder<BoolCodec>);
266impl_deferred_sync!(CoreVecDecoder<S8Codec>);
267impl_deferred_sync!(CoreVecDecoder<U8Codec>);
268impl_deferred_sync!(CoreVecDecoder<S16Codec>);
269impl_deferred_sync!(CoreVecDecoder<U16Codec>);
270impl_deferred_sync!(CoreVecDecoder<S32Codec>);
271impl_deferred_sync!(CoreVecDecoder<U32Codec>);
272impl_deferred_sync!(CoreVecDecoder<S64Codec>);
273impl_deferred_sync!(CoreVecDecoder<U64Codec>);
274impl_deferred_sync!(CoreVecDecoder<F32Codec>);
275impl_deferred_sync!(CoreVecDecoder<F64Codec>);
276impl_deferred_sync!(CoreVecDecoder<CoreNameDecoder>);
277impl_deferred_sync!(CoreVecDecoder<CoreVecDecoderBytes>);
278impl_deferred_sync!(CoreVecDecoder<Utf8Codec>);
279impl_deferred_sync!(CoreVecDecoder<Leb128DecoderI8>);
280impl_deferred_sync!(CoreVecDecoder<Leb128DecoderU8>);
281impl_deferred_sync!(CoreVecDecoder<Leb128DecoderI16>);
282impl_deferred_sync!(CoreVecDecoder<Leb128DecoderU16>);
283impl_deferred_sync!(CoreVecDecoder<Leb128DecoderI32>);
284impl_deferred_sync!(CoreVecDecoder<Leb128DecoderU32>);
285impl_deferred_sync!(CoreVecDecoder<Leb128DecoderI64>);
286impl_deferred_sync!(CoreVecDecoder<Leb128DecoderU64>);
287impl_deferred_sync!(CoreVecDecoder<Leb128DecoderI128>);
288impl_deferred_sync!(CoreVecDecoder<Leb128DecoderU128>);
289impl_deferred_sync!(CoreVecDecoder<UnitCodec>);
290
291/// Codec for synchronous values
292///
293/// This is a wrapper struct, which provides a no-op [Deferred] implementation
294/// for any codec.
295pub struct SyncCodec<T>(pub T);
296
297impl<T> Deref for SyncCodec<T> {
298    type Target = T;
299
300    fn deref(&self) -> &Self::Target {
301        &self.0
302    }
303}
304
305impl<T> DerefMut for SyncCodec<T> {
306    fn deref_mut(&mut self) -> &mut Self::Target {
307        &mut self.0
308    }
309}
310
311impl<T, C> Deferred<T> for SyncCodec<C> {
312    fn take_deferred(&mut self) -> Option<DeferredFn<T>> {
313        None
314    }
315}
316
317impl<T: Default> Default for SyncCodec<T> {
318    fn default() -> Self {
319        Self(T::default())
320    }
321}
322
323impl<T, I> tokio_util::codec::Encoder<I> for SyncCodec<T>
324where
325    T: tokio_util::codec::Encoder<I>,
326{
327    type Error = T::Error;
328
329    fn encode(&mut self, item: I, dst: &mut BytesMut) -> Result<(), Self::Error> {
330        self.0.encode(item, dst)
331    }
332}
333
334impl<T> tokio_util::codec::Decoder for SyncCodec<T>
335where
336    T: tokio_util::codec::Decoder,
337{
338    type Item = T::Item;
339    type Error = T::Error;
340
341    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
342        self.0.decode(src)
343    }
344
345    fn decode_eof(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
346        self.0.decode_eof(buf)
347    }
348
349    fn framed<IO: AsyncRead + AsyncWrite + Sized>(
350        self,
351        io: IO,
352    ) -> tokio_util::codec::Framed<IO, Self>
353    where
354        Self: Sized,
355    {
356        self.0.framed(io).map_codec(Self)
357    }
358}
359
360macro_rules! impl_handle_deferred {
361    ($name:ident, $t:ty) => {
362        #[instrument(level = "trace", skip(w, deferred))]
363        async fn $name<I>(w: $t, deferred: I, mut path: Vec<usize>) -> std::io::Result<()>
364        where
365            I: IntoIterator<Item = Option<DeferredFn<$t>>>,
366            I::IntoIter: ExactSizeIterator,
367        {
368            let mut futs = FuturesUnordered::default();
369            for (i, f) in zip(0.., deferred) {
370                if let Some(f) = f {
371                    path.push(i);
372                    let w = w.index(&path).map_err(std::io::Error::other)?;
373                    path.pop();
374                    futs.push(f(w, Vec::default()));
375                }
376            }
377            while let Some(()) = futs.try_next().await? {}
378            Ok(())
379        }
380    };
381}
382
383impl_handle_deferred!(handle_deferred_tx, Outgoing);
384impl_handle_deferred!(handle_deferred_rx, BufferedIncoming);
385
386/// Defines value encoding
387pub trait Encode: Sized {
388    /// Encoder used to encode the value
389    type Encoder: tokio_util::codec::Encoder<Self> + Deferred<Outgoing> + Default + Send;
390
391    /// Convenience function for encoding a value
392    #[instrument(level = "trace", skip(self, enc))]
393    fn encode(
394        self,
395        enc: &mut Self::Encoder,
396        dst: &mut BytesMut,
397    ) -> Result<
398        Option<DeferredFn<Outgoing>>,
399        <Self::Encoder as tokio_util::codec::Encoder<Self>>::Error,
400    > {
401        enc.encode(self, dst)?;
402        Ok(enc.take_deferred())
403    }
404
405    /// Encode an iterator of owned values
406    #[instrument(level = "trace", skip(items, enc))]
407    fn encode_iter_own<I>(
408        items: I,
409        enc: &mut Self::Encoder,
410        dst: &mut BytesMut,
411    ) -> Result<
412        Option<DeferredFn<Outgoing>>,
413        <Self::Encoder as tokio_util::codec::Encoder<Self>>::Error,
414    >
415    where
416        I: IntoIterator<Item = Self>,
417        I::IntoIter: ExactSizeIterator,
418    {
419        let items = items.into_iter();
420        dst.reserve(items.len());
421        let mut deferred = Vec::with_capacity(items.len());
422        for item in items {
423            enc.encode(item, dst)?;
424            deferred.push(enc.take_deferred());
425        }
426        if deferred.iter().any(Option::is_some) {
427            Ok(Some(Box::new(move |w, path| {
428                Box::pin(handle_deferred_tx(w, deferred, path))
429            })))
430        } else {
431            Ok(None)
432        }
433    }
434
435    /// Encode an iterator of value references
436    #[instrument(level = "trace", skip(items, enc))]
437    fn encode_iter_ref<'a, I>(
438        items: I,
439        enc: &mut Self::Encoder,
440        dst: &mut BytesMut,
441    ) -> Result<
442        Option<DeferredFn<Outgoing>>,
443        <Self::Encoder as tokio_util::codec::Encoder<&'a Self>>::Error,
444    >
445    where
446        I: IntoIterator<Item = &'a Self>,
447        I::IntoIter: ExactSizeIterator,
448        Self::Encoder: tokio_util::codec::Encoder<&'a Self>,
449    {
450        let items = items.into_iter();
451        dst.reserve(items.len());
452        let mut deferred = Vec::with_capacity(items.len());
453        for item in items {
454            enc.encode(item, dst)?;
455            deferred.push(enc.take_deferred());
456        }
457        if deferred.iter().any(Option::is_some) {
458            Ok(Some(Box::new(move |w, path| {
459                Box::pin(handle_deferred_tx(w, deferred, path))
460            })))
461        } else {
462            Ok(None)
463        }
464    }
465
466    /// Encode a list of owned values
467    #[instrument(level = "trace", skip(items, enc), fields(ty = "list"))]
468    fn encode_list_own(
469        items: Vec<Self>,
470        enc: &mut Self::Encoder,
471        dst: &mut BytesMut,
472    ) -> Result<
473        Option<DeferredFn<Outgoing>>,
474        <Self::Encoder as tokio_util::codec::Encoder<Self>>::Error,
475    > {
476        let n = u32::try_from(items.len())
477            .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidInput, err))?;
478        dst.reserve(5 + items.len());
479        Leb128Encoder.encode(n, dst)?;
480        Self::encode_iter_own(items, enc, dst)
481    }
482
483    /// Encode a list of value references
484    #[instrument(level = "trace", skip(items, enc), fields(ty = "list"))]
485    fn encode_list_ref<'a>(
486        items: &'a [Self],
487        enc: &mut Self::Encoder,
488        dst: &mut BytesMut,
489    ) -> Result<
490        Option<DeferredFn<Outgoing>>,
491        <Self::Encoder as tokio_util::codec::Encoder<&'a Self>>::Error,
492    >
493    where
494        Self::Encoder: tokio_util::codec::Encoder<&'a Self>,
495    {
496        let n = u32::try_from(items.len())
497            .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidInput, err))?;
498        dst.reserve(5 + items.len());
499        Leb128Encoder.encode(n, dst)?;
500        Self::encode_iter_ref(items, enc, dst)
501    }
502}
503
504/// Defines value decoding
505pub trait Decode: Sized {
506    /// Decoder used to decode value
507    type Decoder: tokio_util::codec::Decoder<Item = Self>
508        + Deferred<BufferedIncoming>
509        + Default
510        + Send
511        + 'static;
512    /// Decoder used to decode lists of value
513    type ListDecoder: tokio_util::codec::Decoder<Item = Vec<Self>> + Default + 'static;
514}
515
516impl<T, W> Deferred<W> for OptionEncoder<T>
517where
518    T: Deferred<W>,
519{
520    fn take_deferred(&mut self) -> Option<DeferredFn<W>> {
521        self.0.take_deferred()
522    }
523}
524
525impl<T> Encode for Option<T>
526where
527    T: Encode,
528{
529    type Encoder = OptionEncoder<T::Encoder>;
530}
531
532impl<'a, T> Encode for &'a Option<T>
533where
534    T: Encode,
535    T::Encoder: tokio_util::codec::Encoder<&'a T>,
536{
537    type Encoder = OptionEncoder<T::Encoder>;
538}
539
540impl<T, W> Deferred<W> for OptionDecoder<T>
541where
542    T: Deferred<W> + Default,
543{
544    fn take_deferred(&mut self) -> Option<DeferredFn<W>> {
545        mem::take(self).into_inner().take_deferred()
546    }
547}
548
549impl<T> Decode for Option<T>
550where
551    T: Decode,
552{
553    type Decoder = OptionDecoder<T::Decoder>;
554    type ListDecoder = ListDecoder<Self::Decoder>;
555}
556
557impl<O, E, W> Deferred<W> for ResultEncoder<O, E>
558where
559    O: Deferred<W>,
560    E: Deferred<W>,
561{
562    fn take_deferred(&mut self) -> Option<DeferredFn<W>> {
563        match (self.ok.take_deferred(), self.err.take_deferred()) {
564            (None, None) => None,
565            (Some(ok), None) => Some(ok),
566            (None, Some(err)) => Some(err),
567            (Some(ok), Some(_)) => {
568                if cfg!(debug_assertions) {
569                    panic!("both `result::ok` and `result::err` deferred function set");
570                } else {
571                    Some(ok)
572                }
573            }
574        }
575    }
576}
577
578impl<O, E> Encode for Result<O, E>
579where
580    O: Encode,
581    E: Encode,
582    std::io::Error: From<<O::Encoder as tokio_util::codec::Encoder<O>>::Error>,
583    std::io::Error: From<<E::Encoder as tokio_util::codec::Encoder<E>>::Error>,
584{
585    type Encoder = ResultEncoder<O::Encoder, E::Encoder>;
586}
587
588impl<'a, O, E> Encode for &'a Result<O, E>
589where
590    O: Encode,
591    O::Encoder: tokio_util::codec::Encoder<&'a O>,
592    E: Encode,
593    E::Encoder: tokio_util::codec::Encoder<&'a E>,
594    std::io::Error: From<<O::Encoder as tokio_util::codec::Encoder<&'a O>>::Error>,
595    std::io::Error: From<<E::Encoder as tokio_util::codec::Encoder<&'a E>>::Error>,
596{
597    type Encoder = ResultEncoder<O::Encoder, E::Encoder>;
598}
599
600impl<O, E, W> Deferred<W> for ResultDecoder<O, E>
601where
602    O: Deferred<W> + Default,
603    E: Deferred<W> + Default,
604{
605    fn take_deferred(&mut self) -> Option<DeferredFn<W>> {
606        let (mut ok, mut err) = mem::take(self).into_inner();
607        match (ok.take_deferred(), err.take_deferred()) {
608            (None, None) => None,
609            (Some(ok), None) => Some(ok),
610            (None, Some(err)) => Some(err),
611            (Some(ok), Some(_)) => {
612                if cfg!(debug_assertions) {
613                    panic!("both `result::ok` and `result::err` deferred function set");
614                } else {
615                    Some(ok)
616                }
617            }
618        }
619    }
620}
621
622impl<O, E> Decode for Result<O, E>
623where
624    O: Decode,
625    E: Decode,
626    std::io::Error: From<<O::Decoder as tokio_util::codec::Decoder>::Error>,
627    std::io::Error: From<<E::Decoder as tokio_util::codec::Decoder>::Error>,
628{
629    type Decoder = ResultDecoder<O::Decoder, E::Decoder>;
630    type ListDecoder = ListDecoder<Self::Decoder>;
631}
632
633/// Encoder for `list<T>`
634#[derive(Default)]
635pub struct ListEncoder {
636    deferred: Option<DeferredFn<Outgoing>>,
637}
638
639impl Deferred<Outgoing> for ListEncoder {
640    fn take_deferred(&mut self) -> Option<DeferredFn<Outgoing>> {
641        self.deferred.take()
642    }
643}
644
645impl<T> tokio_util::codec::Encoder<Vec<T>> for ListEncoder
646where
647    T: Encode,
648{
649    type Error = <T::Encoder as tokio_util::codec::Encoder<T>>::Error;
650
651    fn encode(&mut self, items: Vec<T>, dst: &mut BytesMut) -> Result<(), Self::Error> {
652        let mut enc = T::Encoder::default();
653        self.deferred = T::encode_list_own(items, &mut enc, dst)?;
654        Ok(())
655    }
656}
657
658impl<'a, T> tokio_util::codec::Encoder<&'a Vec<T>> for ListEncoder
659where
660    T: Encode,
661    T::Encoder: tokio_util::codec::Encoder<&'a T>,
662{
663    type Error = <T::Encoder as tokio_util::codec::Encoder<&'a T>>::Error;
664
665    fn encode(&mut self, items: &'a Vec<T>, dst: &mut BytesMut) -> Result<(), Self::Error> {
666        let mut enc = T::Encoder::default();
667        self.deferred = T::encode_list_ref(items, &mut enc, dst)?;
668        Ok(())
669    }
670}
671
672impl<'a, 'b, T> tokio_util::codec::Encoder<&'a &'b Vec<T>> for ListEncoder
673where
674    T: Encode,
675    T::Encoder: tokio_util::codec::Encoder<&'b T>,
676{
677    type Error = <T::Encoder as tokio_util::codec::Encoder<&'b T>>::Error;
678
679    fn encode(&mut self, items: &'a &'b Vec<T>, dst: &mut BytesMut) -> Result<(), Self::Error> {
680        let mut enc = T::Encoder::default();
681        self.deferred = T::encode_list_ref(items, &mut enc, dst)?;
682        Ok(())
683    }
684}
685
686impl<'a, T> tokio_util::codec::Encoder<&'a [T]> for ListEncoder
687where
688    T: Encode,
689    T::Encoder: tokio_util::codec::Encoder<&'a T>,
690{
691    type Error = <T::Encoder as tokio_util::codec::Encoder<&'a T>>::Error;
692
693    fn encode(&mut self, items: &'a [T], dst: &mut BytesMut) -> Result<(), Self::Error> {
694        let mut enc = T::Encoder::default();
695        self.deferred = T::encode_list_ref(items, &mut enc, dst)?;
696        Ok(())
697    }
698}
699
700impl<'a, 'b, T> tokio_util::codec::Encoder<&'a &'b [T]> for ListEncoder
701where
702    T: Encode,
703    T::Encoder: tokio_util::codec::Encoder<&'b T>,
704{
705    type Error = <T::Encoder as tokio_util::codec::Encoder<&'b T>>::Error;
706
707    fn encode(&mut self, items: &'a &'b [T], dst: &mut BytesMut) -> Result<(), Self::Error> {
708        let mut enc = T::Encoder::default();
709        self.deferred = T::encode_list_ref(items, &mut enc, dst)?;
710        Ok(())
711    }
712}
713
714impl<T> Encode for Vec<T>
715where
716    T: Encode,
717{
718    type Encoder = ListEncoder;
719}
720
721impl<'a, T> Encode for &'a Vec<T>
722where
723    T: Encode,
724    T::Encoder: tokio_util::codec::Encoder<&'a T>,
725{
726    type Encoder = ListEncoder;
727}
728
729impl<'a, T> Encode for &'a [T]
730where
731    T: Encode,
732    T::Encoder: tokio_util::codec::Encoder<&'a T>,
733{
734    type Encoder = ListEncoder;
735}
736
737/// Decoder for `list<T>`
738pub struct ListDecoder<T>
739where
740    T: tokio_util::codec::Decoder,
741{
742    dec: T,
743    ret: Vec<T::Item>,
744    cap: usize,
745    deferred: Vec<Option<DeferredFn<BufferedIncoming>>>,
746}
747
748impl<T> ListDecoder<T>
749where
750    T: tokio_util::codec::Decoder,
751{
752    /// Constructs a new list decoder
753    pub fn new(dec: T) -> Self {
754        Self {
755            dec,
756            ret: Vec::default(),
757            cap: 0,
758            deferred: vec![],
759        }
760    }
761}
762
763impl<T> Default for ListDecoder<T>
764where
765    T: tokio_util::codec::Decoder + Default,
766{
767    fn default() -> Self {
768        Self::new(T::default())
769    }
770}
771
772impl<T> Deferred<BufferedIncoming> for ListDecoder<T>
773where
774    T: tokio_util::codec::Decoder,
775{
776    fn take_deferred(&mut self) -> Option<DeferredFn<BufferedIncoming>> {
777        let deferred = mem::take(&mut self.deferred);
778        if deferred.iter().any(Option::is_some) {
779            Some(Box::new(|r, path| {
780                Box::pin(handle_deferred_rx(r, deferred, path))
781            }))
782        } else {
783            None
784        }
785    }
786}
787
788impl<T> tokio_util::codec::Decoder for ListDecoder<T>
789where
790    T: tokio_util::codec::Decoder + Deferred<BufferedIncoming>,
791{
792    type Item = Vec<T::Item>;
793    type Error = T::Error;
794
795    #[instrument(level = "trace", skip(self), fields(ty = "list"))]
796    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
797        if self.cap == 0 {
798            let Some(len) = Leb128DecoderU32.decode(src)? else {
799                return Ok(None);
800            };
801            if len == 0 {
802                return Ok(Some(Vec::default()));
803            }
804            let len = len
805                .try_into()
806                .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidInput, err))?;
807            self.ret = Vec::with_capacity(len);
808            self.deferred = Vec::with_capacity(len);
809            self.cap = len;
810        }
811        while self.cap > 0 {
812            let Some(v) = self.dec.decode(src)? else {
813                return Ok(None);
814            };
815            self.ret.push(v);
816            self.deferred.push(self.dec.take_deferred());
817            self.cap -= 1;
818        }
819        Ok(Some(mem::take(&mut self.ret)))
820    }
821}
822
823impl<T> Decode for Vec<T>
824where
825    T: Decode + Send,
826    T::ListDecoder: Deferred<BufferedIncoming> + Send,
827{
828    type Decoder = T::ListDecoder;
829    type ListDecoder = ListDecoder<Self::Decoder>;
830}
831
832macro_rules! impl_copy_codec {
833    ($t:ty, $c:tt) => {
834        impl Encode for $t {
835            type Encoder = $c;
836
837            #[instrument(level = "trace", skip(items))]
838            fn encode_iter_own<I>(
839                items: I,
840                enc: &mut Self::Encoder,
841                dst: &mut BytesMut,
842            ) -> Result<
843                Option<DeferredFn<Outgoing>>,
844                <Self::Encoder as tokio_util::codec::Encoder<Self>>::Error,
845            >
846            where
847                I: IntoIterator<Item = Self>,
848                I::IntoIter: ExactSizeIterator,
849            {
850                let items = items.into_iter();
851                dst.reserve(items.len());
852                for item in items {
853                    enc.encode(item, dst)?;
854                }
855                Ok(None)
856            }
857
858            #[instrument(level = "trace", skip(items))]
859            fn encode_iter_ref<'a, I>(
860                items: I,
861                enc: &mut Self::Encoder,
862                dst: &mut BytesMut,
863            ) -> Result<
864                Option<DeferredFn<Outgoing>>,
865                <Self::Encoder as tokio_util::codec::Encoder<&'a Self>>::Error,
866            >
867            where
868                I: IntoIterator<Item = &'a Self>,
869                I::IntoIter: ExactSizeIterator,
870            {
871                let items = items.into_iter();
872                dst.reserve(items.len());
873                for item in items {
874                    enc.encode(*item, dst)?;
875                }
876                Ok(None)
877            }
878        }
879
880        impl<'b> Encode for &'b $t {
881            type Encoder = $c;
882
883            #[instrument(level = "trace", skip(items))]
884            fn encode_iter_own<I>(
885                items: I,
886                enc: &mut Self::Encoder,
887                dst: &mut BytesMut,
888            ) -> Result<
889                Option<DeferredFn<Outgoing>>,
890                <Self::Encoder as tokio_util::codec::Encoder<Self>>::Error,
891            >
892            where
893                I: IntoIterator<Item = Self>,
894                I::IntoIter: ExactSizeIterator,
895            {
896                let items = items.into_iter();
897                dst.reserve(items.len());
898                for item in items {
899                    enc.encode(*item, dst)?;
900                }
901                Ok(None)
902            }
903
904            #[instrument(level = "trace", skip(items))]
905            fn encode_iter_ref<'a, I>(
906                items: I,
907                enc: &mut Self::Encoder,
908                dst: &mut BytesMut,
909            ) -> Result<
910                Option<DeferredFn<Outgoing>>,
911                <Self::Encoder as tokio_util::codec::Encoder<&'a Self>>::Error,
912            >
913            where
914                I: IntoIterator<Item = &'a Self>,
915                I::IntoIter: ExactSizeIterator,
916                'b: 'a,
917            {
918                let items = items.into_iter();
919                dst.reserve(items.len());
920                for item in items {
921                    enc.encode(item, dst)?;
922                }
923                Ok(None)
924            }
925        }
926
927        impl Decode for $t {
928            type Decoder = $c;
929            type ListDecoder = CoreVecDecoder<Self::Decoder>;
930        }
931    };
932}
933
934// The Component Model canonical ABI mandates a single canonical `NaN`
935// representation for floating point values. Encoding canonicalizes `NaN`s to
936// match; decoding is lenient and accepts any `NaN` representation.
937//
938// See `canonicalize_nan{32,64}` in
939// <https://github.com/WebAssembly/component-model/blob/main/design/mvp/canonical-abi/definitions.py>.
940const CANONICAL_NAN_F32: u32 = 0x7fc0_0000;
941const CANONICAL_NAN_F64: u64 = 0x7ff8_0000_0000_0000;
942
943/// Defines a floating-point codec that canonicalizes `NaN` values on encode to
944/// match the Component Model canonical ABI, delegating the actual byte encoding
945/// and decoding to the wrapped `wasm-tokio` codec.
946macro_rules! impl_canonical_nan_codec {
947    ($name:ident, $inner:ty, $t:ty, $canon:expr_2021) => {
948        #[doc = concat!("Canonicalizes `NaN`s on encode, wrapping [`", stringify!($inner), "`].")]
949        #[derive(Debug, Default)]
950        pub struct $name($inner);
951
952        impl tokio_util::codec::Encoder<$t> for $name {
953            type Error = std::io::Error;
954
955            fn encode(&mut self, item: $t, dst: &mut BytesMut) -> Result<(), Self::Error> {
956                let item = if item.is_nan() {
957                    <$t>::from_bits($canon)
958                } else {
959                    item
960                };
961                self.0.encode(item, dst)
962            }
963        }
964
965        impl tokio_util::codec::Encoder<&$t> for $name {
966            type Error = std::io::Error;
967
968            fn encode(&mut self, item: &$t, dst: &mut BytesMut) -> Result<(), Self::Error> {
969                tokio_util::codec::Encoder::<$t>::encode(self, *item, dst)
970            }
971        }
972
973        impl tokio_util::codec::Encoder<&&$t> for $name {
974            type Error = std::io::Error;
975
976            fn encode(&mut self, item: &&$t, dst: &mut BytesMut) -> Result<(), Self::Error> {
977                tokio_util::codec::Encoder::<$t>::encode(self, **item, dst)
978            }
979        }
980
981        impl tokio_util::codec::Decoder for $name {
982            type Item = $t;
983            type Error = std::io::Error;
984
985            fn decode(&mut self, src: &mut BytesMut) -> Result<Option<$t>, Self::Error> {
986                self.0.decode(src)
987            }
988        }
989
990        impl_deferred_sync!($name);
991        impl_deferred_sync!(CoreVecDecoder<$name>);
992    };
993}
994
995impl_canonical_nan_codec!(CanonicalNanF32Codec, F32Codec, f32, CANONICAL_NAN_F32);
996impl_canonical_nan_codec!(CanonicalNanF64Codec, F64Codec, f64, CANONICAL_NAN_F64);
997
998impl_copy_codec!(bool, BoolCodec);
999impl_copy_codec!(i8, S8Codec);
1000impl_copy_codec!(i16, S16Codec);
1001impl_copy_codec!(u16, U16Codec);
1002impl_copy_codec!(i32, S32Codec);
1003impl_copy_codec!(u32, U32Codec);
1004impl_copy_codec!(i64, S64Codec);
1005impl_copy_codec!(u64, U64Codec);
1006impl_copy_codec!(f32, CanonicalNanF32Codec);
1007impl_copy_codec!(f64, CanonicalNanF64Codec);
1008impl_copy_codec!(char, Utf8Codec);
1009
1010impl Encode for u8 {
1011    type Encoder = U8Codec;
1012
1013    #[instrument(level = "trace", skip(items))]
1014    fn encode_iter_own<I>(
1015        items: I,
1016        enc: &mut Self::Encoder,
1017        dst: &mut BytesMut,
1018    ) -> Result<
1019        Option<DeferredFn<Outgoing>>,
1020        <Self::Encoder as tokio_util::codec::Encoder<Self>>::Error,
1021    >
1022    where
1023        I: IntoIterator<Item = Self>,
1024        I::IntoIter: ExactSizeIterator,
1025    {
1026        let items = items.into_iter();
1027        dst.reserve(items.len());
1028        dst.extend(items);
1029        Ok(None)
1030    }
1031
1032    #[instrument(level = "trace", skip(items))]
1033    fn encode_iter_ref<'a, I>(
1034        items: I,
1035        enc: &mut Self::Encoder,
1036        dst: &mut BytesMut,
1037    ) -> Result<
1038        Option<DeferredFn<Outgoing>>,
1039        <Self::Encoder as tokio_util::codec::Encoder<&'a Self>>::Error,
1040    >
1041    where
1042        I: IntoIterator<Item = &'a Self>,
1043        I::IntoIter: ExactSizeIterator,
1044    {
1045        let items = items.into_iter();
1046        dst.reserve(items.len());
1047        dst.extend(items);
1048        Ok(None)
1049    }
1050
1051    #[instrument(level = "trace", skip(items), fields(ty = "list<u8>"))]
1052    fn encode_list_own(
1053        items: Vec<Self>,
1054        enc: &mut Self::Encoder,
1055        dst: &mut BytesMut,
1056    ) -> Result<
1057        Option<DeferredFn<Outgoing>>,
1058        <Self::Encoder as tokio_util::codec::Encoder<Self>>::Error,
1059    > {
1060        CoreVecEncoderBytes.encode(items, dst)?;
1061        Ok(None)
1062    }
1063
1064    #[instrument(level = "trace", skip(items), fields(ty = "list<u8>"))]
1065    fn encode_list_ref<'a>(
1066        items: &'a [Self],
1067        enc: &mut Self::Encoder,
1068        dst: &mut BytesMut,
1069    ) -> Result<
1070        Option<DeferredFn<Outgoing>>,
1071        <Self::Encoder as tokio_util::codec::Encoder<&'a Self>>::Error,
1072    >
1073    where
1074        Self::Encoder: tokio_util::codec::Encoder<&'a Self>,
1075    {
1076        CoreVecEncoderBytes.encode(items, dst)?;
1077        Ok(None)
1078    }
1079}
1080
1081impl<'b> Encode for &'b u8 {
1082    type Encoder = U8Codec;
1083
1084    #[instrument(level = "trace", skip(items))]
1085    fn encode_iter_own<I>(
1086        items: I,
1087        enc: &mut Self::Encoder,
1088        dst: &mut BytesMut,
1089    ) -> Result<
1090        Option<DeferredFn<Outgoing>>,
1091        <Self::Encoder as tokio_util::codec::Encoder<Self>>::Error,
1092    >
1093    where
1094        I: IntoIterator<Item = Self>,
1095        I::IntoIter: ExactSizeIterator,
1096    {
1097        let items = items.into_iter();
1098        dst.reserve(items.len());
1099        dst.extend(items);
1100        Ok(None)
1101    }
1102
1103    #[instrument(level = "trace", skip(items))]
1104    fn encode_iter_ref<'a, I>(
1105        items: I,
1106        enc: &mut Self::Encoder,
1107        dst: &mut BytesMut,
1108    ) -> Result<
1109        Option<DeferredFn<Outgoing>>,
1110        <Self::Encoder as tokio_util::codec::Encoder<&'a Self>>::Error,
1111    >
1112    where
1113        I: IntoIterator<Item = &'a Self>,
1114        I::IntoIter: ExactSizeIterator,
1115        'b: 'a,
1116    {
1117        let items = items.into_iter();
1118        dst.reserve(items.len());
1119        dst.extend(items.map(|b| **b));
1120        Ok(None)
1121    }
1122}
1123
1124/// Decoder for `list<u8>`
1125#[derive(Debug, Default)]
1126#[repr(transparent)]
1127pub struct ListDecoderU8(CoreVecDecoderBytes);
1128
1129impl tokio_util::codec::Decoder for ListDecoderU8 {
1130    type Item = Vec<u8>;
1131    type Error = <CoreVecDecoderBytes as tokio_util::codec::Decoder>::Error;
1132
1133    #[instrument(level = "trace", skip(self), fields(ty = "list<u8>"))]
1134    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1135        let Some(buf) = self.0.decode(src)? else {
1136            return Ok(None);
1137        };
1138        Ok(Some(buf.into()))
1139    }
1140}
1141
1142impl Decode for u8 {
1143    type Decoder = U8Codec;
1144    type ListDecoder = ListDecoderU8;
1145}
1146
1147impl Encode for &str {
1148    type Encoder = CoreNameEncoder;
1149}
1150
1151impl Encode for &&str {
1152    type Encoder = CoreNameEncoder;
1153}
1154
1155impl Encode for String {
1156    type Encoder = CoreNameEncoder;
1157}
1158
1159impl Encode for &String {
1160    type Encoder = CoreNameEncoder;
1161}
1162
1163impl Decode for String {
1164    type Decoder = CoreNameDecoder;
1165    type ListDecoder = CoreVecDecoder<Self::Decoder>;
1166}
1167
1168impl Encode for Bytes {
1169    type Encoder = CoreVecEncoderBytes;
1170}
1171
1172impl Encode for &Bytes {
1173    type Encoder = CoreVecEncoderBytes;
1174}
1175
1176impl Decode for Bytes {
1177    type Decoder = CoreVecDecoderBytes;
1178    type ListDecoder = CoreVecDecoder<Self::Decoder>;
1179}
1180
1181/// Encoder for `resource` types
1182#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
1183#[repr(transparent)]
1184pub struct ResourceEncoder;
1185
1186impl<T: ?Sized> tokio_util::codec::Encoder<ResourceOwn<T>> for ResourceEncoder {
1187    type Error = std::io::Error;
1188
1189    #[instrument(level = "trace", skip(self, item), ret, fields(ty = "own"))]
1190    fn encode(&mut self, item: ResourceOwn<T>, dst: &mut BytesMut) -> std::io::Result<()> {
1191        CoreVecEncoderBytes.encode(item.repr, dst)
1192    }
1193}
1194
1195impl<T: ?Sized> tokio_util::codec::Encoder<&ResourceOwn<T>> for ResourceEncoder {
1196    type Error = std::io::Error;
1197
1198    #[instrument(level = "trace", skip(self, item), ret, fields(ty = "own"))]
1199    fn encode(&mut self, item: &ResourceOwn<T>, dst: &mut BytesMut) -> std::io::Result<()> {
1200        CoreVecEncoderBytes.encode(&item.repr, dst)
1201    }
1202}
1203
1204impl<T: ?Sized> Encode for ResourceOwn<T> {
1205    type Encoder = ResourceEncoder;
1206}
1207
1208impl<T: ?Sized> Encode for &ResourceOwn<T> {
1209    type Encoder = ResourceEncoder;
1210}
1211
1212impl<T: ?Sized> tokio_util::codec::Encoder<ResourceBorrow<T>> for ResourceEncoder {
1213    type Error = std::io::Error;
1214
1215    #[instrument(level = "trace", skip(self, item), ret, fields(ty = "borrow"))]
1216    fn encode(&mut self, item: ResourceBorrow<T>, dst: &mut BytesMut) -> std::io::Result<()> {
1217        CoreVecEncoderBytes.encode(item.repr, dst)
1218    }
1219}
1220
1221impl<T: ?Sized> tokio_util::codec::Encoder<&ResourceBorrow<T>> for ResourceEncoder {
1222    type Error = std::io::Error;
1223
1224    #[instrument(level = "trace", skip(self, item), ret, fields(ty = "borrow"))]
1225    fn encode(&mut self, item: &ResourceBorrow<T>, dst: &mut BytesMut) -> std::io::Result<()> {
1226        CoreVecEncoderBytes.encode(&item.repr, dst)
1227    }
1228}
1229
1230impl<T: ?Sized> Encode for ResourceBorrow<T> {
1231    type Encoder = ResourceEncoder;
1232}
1233
1234impl<T: ?Sized> Encode for &ResourceBorrow<T> {
1235    type Encoder = ResourceEncoder;
1236}
1237
1238/// Decoder for borrowed resource types
1239#[derive(Debug)]
1240#[repr(transparent)]
1241pub struct ResourceBorrowDecoder<T: ?Sized> {
1242    dec: CoreVecDecoderBytes,
1243    _ty: PhantomData<T>,
1244}
1245
1246impl<T: ?Sized> Default for ResourceBorrowDecoder<T> {
1247    fn default() -> Self {
1248        Self {
1249            dec: CoreVecDecoderBytes::default(),
1250            _ty: PhantomData,
1251        }
1252    }
1253}
1254
1255impl<T: ?Sized> Deferred<BufferedIncoming> for ResourceBorrowDecoder<T> {
1256    fn take_deferred(&mut self) -> Option<DeferredFn<BufferedIncoming>> {
1257        None
1258    }
1259}
1260
1261impl<T: ?Sized> Deferred<BufferedIncoming> for CoreVecDecoder<ResourceBorrowDecoder<T>> {
1262    fn take_deferred(&mut self) -> Option<DeferredFn<BufferedIncoming>> {
1263        None
1264    }
1265}
1266
1267impl<T: ?Sized + Send + 'static> Decode for ResourceBorrow<T> {
1268    type Decoder = ResourceBorrowDecoder<T>;
1269    type ListDecoder = CoreVecDecoder<Self::Decoder>;
1270}
1271
1272impl<T: ?Sized> tokio_util::codec::Decoder for ResourceBorrowDecoder<T> {
1273    type Item = ResourceBorrow<T>;
1274    type Error = std::io::Error;
1275
1276    #[instrument(level = "trace", skip(self), fields(ty = "borrow"))]
1277    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1278        let repr = self.dec.decode(src)?;
1279        Ok(repr.map(Self::Item::from))
1280    }
1281}
1282
1283/// Decoder for owned resource types
1284#[derive(Debug)]
1285#[repr(transparent)]
1286pub struct ResourceOwnDecoder<T: ?Sized> {
1287    dec: CoreVecDecoderBytes,
1288    _ty: PhantomData<T>,
1289}
1290
1291impl<T: ?Sized> Default for ResourceOwnDecoder<T> {
1292    fn default() -> Self {
1293        Self {
1294            dec: CoreVecDecoderBytes::default(),
1295            _ty: PhantomData,
1296        }
1297    }
1298}
1299
1300impl<T: ?Sized> Deferred<BufferedIncoming> for ResourceOwnDecoder<T> {
1301    fn take_deferred(&mut self) -> Option<DeferredFn<BufferedIncoming>> {
1302        None
1303    }
1304}
1305
1306impl<T: ?Sized> Deferred<BufferedIncoming> for CoreVecDecoder<ResourceOwnDecoder<T>> {
1307    fn take_deferred(&mut self) -> Option<DeferredFn<BufferedIncoming>> {
1308        None
1309    }
1310}
1311
1312impl<T: ?Sized + Send + 'static> Decode for ResourceOwn<T> {
1313    type Decoder = ResourceOwnDecoder<T>;
1314    type ListDecoder = CoreVecDecoder<Self::Decoder>;
1315}
1316
1317impl<T: ?Sized> tokio_util::codec::Decoder for ResourceOwnDecoder<T> {
1318    type Item = ResourceOwn<T>;
1319    type Error = std::io::Error;
1320
1321    #[instrument(level = "trace", skip(self), fields(ty = "own"))]
1322    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1323        let repr = self.dec.decode(src)?;
1324        Ok(repr.map(Self::Item::from))
1325    }
1326}
1327
1328/// Codec for `()`
1329#[derive(Copy, Clone, Debug, Default, Eq, PartialEq)]
1330#[repr(transparent)]
1331pub struct UnitCodec;
1332
1333impl tokio_util::codec::Encoder<()> for UnitCodec {
1334    type Error = std::io::Error;
1335
1336    #[instrument(level = "trace", skip(self), ret)]
1337    fn encode(&mut self, (): (), dst: &mut BytesMut) -> std::io::Result<()> {
1338        Ok(())
1339    }
1340}
1341
1342impl tokio_util::codec::Encoder<&()> for UnitCodec {
1343    type Error = std::io::Error;
1344
1345    #[instrument(level = "trace", skip(self), ret)]
1346    fn encode(&mut self, (): &(), dst: &mut BytesMut) -> std::io::Result<()> {
1347        Ok(())
1348    }
1349}
1350
1351impl tokio_util::codec::Decoder for UnitCodec {
1352    type Item = ();
1353    type Error = std::io::Error;
1354
1355    #[instrument(level = "trace", skip(self))]
1356    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1357        Ok(Some(()))
1358    }
1359}
1360
1361/// Marker trait for [Encode] tuple types
1362pub trait TupleEncode: Encode {}
1363
1364/// Marker trait for [Decode] tuple types
1365pub trait TupleDecode: Decode {}
1366
1367impl Encode for () {
1368    type Encoder = UnitCodec;
1369}
1370
1371impl TupleEncode for () {}
1372
1373impl Decode for () {
1374    type Decoder = UnitCodec;
1375    type ListDecoder = CoreVecDecoder<Self::Decoder>;
1376}
1377
1378impl TupleDecode for () {}
1379
1380macro_rules! impl_tuple_codec {
1381    ($($vn:ident),+; $($vt:ident),+; $($cn:ident),+; $($ct:ident),+) => {
1382        impl<$($ct),+> Deferred<Outgoing> for TupleEncoder::<($($ct),+,)>
1383        where
1384            $($ct: Deferred<Outgoing> + Default + 'static),+
1385        {
1386            fn take_deferred(&mut self) -> Option<DeferredFn<Outgoing>> {
1387                let Self(($(mut $cn),+,)) = mem::take(self);
1388                let deferred = [ $($cn.take_deferred()),+ ];
1389                if deferred.iter().any(Option::is_some) {
1390                    Some(Box::new(|r, path| Box::pin(handle_deferred_tx(r, deferred, path))))
1391                } else {
1392                    None
1393                }
1394            }
1395        }
1396
1397        impl<E, $($vt),+> Encode for ($($vt),+,)
1398        where
1399            E: From<std::io::Error>,
1400            $(
1401                $vt: Encode,
1402                $vt::Encoder: tokio_util::codec::Encoder<$vt, Error = E> + 'static,
1403            )+
1404        {
1405            type Encoder = TupleEncoder::<($($vt::Encoder),+,)>;
1406        }
1407
1408        impl<E, $($vt),+> TupleEncode for ($($vt),+,)
1409        where
1410            E: From<std::io::Error>,
1411            $(
1412                $vt: Encode,
1413                $vt::Encoder: tokio_util::codec::Encoder<$vt, Error = E> + 'static,
1414            )+
1415        {
1416        }
1417
1418        impl<'a, E, $($vt),+> Encode for &'a ($($vt),+,)
1419        where
1420            E: From<std::io::Error>,
1421            $(
1422                $vt: Encode,
1423                $vt::Encoder: tokio_util::codec::Encoder<&'a $vt, Error = E> + 'static,
1424            )+
1425        {
1426            type Encoder = TupleEncoder::<($($vt::Encoder),+,)>;
1427        }
1428
1429        impl<$($vt),+> Deferred<BufferedIncoming> for TupleDecoder::<($($vt::Decoder),+,), ($(Option<$vt>),+,)>
1430        where
1431            $($vt: Decode),+
1432        {
1433            fn take_deferred(&mut self) -> Option<DeferredFn<BufferedIncoming>> {
1434                let ($(mut $cn),+,) = mem::take(self).into_inner();
1435                let deferred = [ $($cn.take_deferred()),+ ];
1436                if deferred.iter().any(Option::is_some) {
1437                    Some(Box::new(|r, path| Box::pin(handle_deferred_rx(r, deferred, path))))
1438                } else {
1439                    None
1440                }
1441            }
1442        }
1443
1444        impl<E, $($vt),+> Decode for ($($vt),+,)
1445        where
1446            E: From<std::io::Error>,
1447            $(
1448                $vt: Decode + Send + 'static,
1449                $vt::Decoder: tokio_util::codec::Decoder<Error = E> + Send + 'static,
1450            )+
1451        {
1452            type Decoder = TupleDecoder::<($($vt::Decoder),+,), ($(Option<$vt>),+,)>;
1453            type ListDecoder = ListDecoder<Self::Decoder>;
1454        }
1455
1456        impl<E, $($vt),+> TupleDecode for ($($vt),+,)
1457        where
1458            E: From<std::io::Error>,
1459            $(
1460                $vt: Decode + Send + 'static,
1461                $vt::Decoder: tokio_util::codec::Decoder<Error = E> + Send + 'static,
1462            )+
1463        {
1464        }
1465    };
1466}
1467
1468impl_tuple_codec!(
1469    v0;
1470    V0;
1471    c0;
1472    C0
1473);
1474
1475impl_tuple_codec!(
1476    v0, v1;
1477    V0, V1;
1478    c0, c1;
1479    C0, C1
1480);
1481
1482impl_tuple_codec!(
1483    v0, v1, v2;
1484    V0, V1, V2;
1485    c0, c1, c2;
1486    C0, C1, C2
1487);
1488
1489impl_tuple_codec!(
1490    v0, v1, v2, v3;
1491    V0, V1, V2, V3;
1492    c0, c1, c2, c3;
1493    C0, C1, C2, C3
1494);
1495
1496impl_tuple_codec!(
1497    v0, v1, v2, v3, v4;
1498    V0, V1, V2, V3, V4;
1499    c0, c1, c2, c3, c4;
1500    C0, C1, C2, C3, C4
1501);
1502
1503impl_tuple_codec!(
1504    v0, v1, v2, v3, v4, v5;
1505    V0, V1, V2, V3, V4, V5;
1506    c0, c1, c2, c3, c4, c5;
1507    C0, C1, C2, C3, C4, C5
1508);
1509
1510impl_tuple_codec!(
1511    v0, v1, v2, v3, v4, v5, v6;
1512    V0, V1, V2, V3, V4, V5, V6;
1513    c0, c1, c2, c3, c4, c5, c6;
1514    C0, C1, C2, C3, C4, C5, C6
1515);
1516
1517impl_tuple_codec!(
1518    v0, v1, v2, v3, v4, v5, v6, v7;
1519    V0, V1, V2, V3, V4, V5, V6, V7;
1520    c0, c1, c2, c3, c4, c5, c6, c7;
1521    C0, C1, C2, C3, C4, C5, C6, C7
1522);
1523
1524impl_tuple_codec!(
1525    v0, v1, v2, v3, v4, v5, v6, v7, v8;
1526    V0, V1, V2, V3, V4, V5, V6, V7, V8;
1527    c0, c1, c2, c3, c4, c5, c6, c7, c8;
1528    C0, C1, C2, C3, C4, C5, C6, C7, C8
1529);
1530
1531impl_tuple_codec!(
1532    v0, v1, v2, v3, v4, v5, v6, v7, v8, v9;
1533    V0, V1, V2, V3, V4, V5, V6, V7, V8, V9;
1534    c0, c1, c2, c3, c4, c5, c6, c7, c8, c9;
1535    C0, C1, C2, C3, C4, C5, C6, C7, C8, C9
1536);
1537
1538impl_tuple_codec!(
1539    v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10;
1540    V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10;
1541    c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10;
1542    C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10
1543);
1544
1545impl_tuple_codec!(
1546    v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11;
1547    V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10, V11;
1548    c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11;
1549    C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10, C11
1550);
1551
1552impl_tuple_codec!(
1553    v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12;
1554    V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10, V11, V12;
1555    c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12;
1556    C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10, C11, C12
1557);
1558
1559impl_tuple_codec!(
1560    v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13;
1561    V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10, V11, V12, V13;
1562    c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13;
1563    C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10, C11, C12, C13
1564);
1565
1566impl_tuple_codec!(
1567    v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14;
1568    V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10, V11, V12, V13, V14;
1569    c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13, c14;
1570    C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10, C11, C12, C13, C14
1571);
1572
1573impl_tuple_codec!(
1574    v0, v1, v2, v3, v4, v5, v6, v7, v8, v9, v10, v11, v12, v13, v14, v15;
1575    V0, V1, V2, V3, V4, V5, V6, V7, V8, V9, V10, V11, V12, V13, V14, V15;
1576    c0, c1, c2, c3, c4, c5, c6, c7, c8, c9, c10, c11, c12, c13, c14, c15;
1577    C0, C1, C2, C3, C4, C5, C6, C7, C8, C9, C10, C11, C12, C13, C14, C15
1578);
1579
1580/// Encoder for `future<T>`
1581#[derive(Default)]
1582pub struct FutureEncoder {
1583    deferred: Option<DeferredFn<Outgoing>>,
1584}
1585
1586impl Deferred<Outgoing> for FutureEncoder {
1587    fn take_deferred(&mut self) -> Option<DeferredFn<Outgoing>> {
1588        self.deferred.take()
1589    }
1590}
1591
1592impl<T, Fut> tokio_util::codec::Encoder<Fut> for FutureEncoder
1593where
1594    T: Encode,
1595    Fut: Future<Output = T> + Send + 'static,
1596    std::io::Error: From<<T::Encoder as tokio_util::codec::Encoder<T>>::Error>,
1597{
1598    type Error = std::io::Error;
1599
1600    #[instrument(level = "trace", skip(self, item), fields(ty = "future"))]
1601    fn encode(&mut self, item: Fut, dst: &mut BytesMut) -> std::io::Result<()> {
1602        // TODO: Check if future is resolved
1603        dst.reserve(1);
1604        dst.put_u8(0x00);
1605        let span = Span::current();
1606        self.deferred = Some(Box::new(|mut w, path| {
1607            Box::pin(
1608                async move {
1609                    if !path.is_empty() {
1610                        w = w.index(&path).map_err(std::io::Error::other)?;
1611                    }
1612                    let item = item.await;
1613                    let mut enc = T::Encoder::default();
1614                    let mut buf = BytesMut::default();
1615                    enc.encode(item, &mut buf)?;
1616                    w.write_all(&buf).await?;
1617                    match enc.take_deferred() {
1618                        Some(f) => f(w, Vec::default()).await,
1619                        _ => Ok(()),
1620                    }
1621                }
1622                .instrument(span),
1623            )
1624        }));
1625        Ok(())
1626    }
1627}
1628
1629impl<T> Encode for Pin<Box<dyn Future<Output = T> + Send>>
1630where
1631    T: Encode + 'static,
1632    std::io::Error: From<<T::Encoder as tokio_util::codec::Encoder<T>>::Error>,
1633{
1634    type Encoder = FutureEncoder;
1635}
1636
1637/// Decoder for `future<T>`
1638pub struct FutureDecoder<T>
1639where
1640    T: Decode,
1641{
1642    dec: OptionDecoder<T::Decoder>,
1643    deferred: Option<DeferredFn<BufferedIncoming>>,
1644    _ty: PhantomData<T>,
1645}
1646
1647impl<T> Default for FutureDecoder<T>
1648where
1649    T: Decode,
1650{
1651    fn default() -> Self {
1652        Self {
1653            dec: OptionDecoder::default(),
1654            deferred: None,
1655            _ty: PhantomData,
1656        }
1657    }
1658}
1659
1660impl<T> Deferred<BufferedIncoming> for FutureDecoder<T>
1661where
1662    T: Decode,
1663{
1664    fn take_deferred(&mut self) -> Option<DeferredFn<BufferedIncoming>> {
1665        self.deferred.take()
1666    }
1667}
1668
1669impl<T> tokio_util::codec::Decoder for FutureDecoder<T>
1670where
1671    T: Decode + Send + 'static,
1672    std::io::Error: From<<T::Decoder as tokio_util::codec::Decoder>::Error>,
1673{
1674    type Item = Pin<Box<dyn Future<Output = T> + Send>>;
1675    type Error = <T::Decoder as tokio_util::codec::Decoder>::Error;
1676
1677    #[instrument(level = "trace", skip(self), fields(ty = "future"))]
1678    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
1679        let Some(item) = self.dec.decode(src)? else {
1680            return Ok(None);
1681        };
1682        if let Some(item) = item {
1683            self.deferred = self.dec.take_deferred();
1684            return Ok(Some(Box::pin(async { item })));
1685        }
1686
1687        // future is pending
1688        let (tx, rx) = oneshot::channel();
1689        let dec = mem::take(&mut self.dec).into_inner();
1690        let span = Span::current();
1691        self.deferred = Some(Box::new(|mut r, path| {
1692            Box::pin(
1693                async move {
1694                    if !path.is_empty() {
1695                        r = r.index(&path).map_err(std::io::Error::other)?;
1696                    }
1697                    let mut dec = FramedRead::new(r, dec);
1698                    trace!(?path, "receiving future element");
1699                    let Some(item) = dec.next().await else {
1700                        return Err(std::io::ErrorKind::UnexpectedEof.into());
1701                    };
1702                    let item = item?;
1703                    if tx.send(item).is_err() {
1704                        debug!("future receiver closed, discard data");
1705                        return Ok(());
1706                    }
1707                    if let Some(rx) = dec.decoder_mut().take_deferred() {
1708                        let buf = mem::take(dec.read_buffer_mut());
1709                        let mut r = dec.into_inner();
1710                        if r.buffer.is_empty() {
1711                            r.buffer = buf;
1712                        } else {
1713                            r.buffer.unsplit(buf);
1714                        }
1715                        rx(r, Vec::default()).await?;
1716                    }
1717                    Ok(())
1718                }
1719                .instrument(span),
1720            )
1721        }));
1722        Ok(Some(Box::pin(async {
1723            let Ok(ret) = rx.await else {
1724                error!("future I/O dropped");
1725                return pending().await;
1726            };
1727            ret
1728        })))
1729    }
1730}
1731
1732impl<T> Decode for Pin<Box<dyn Future<Output = T> + Send>>
1733where
1734    T: Decode + Send + 'static,
1735    std::io::Error: From<<T::Decoder as tokio_util::codec::Decoder>::Error>,
1736{
1737    type Decoder = FutureDecoder<T>;
1738    type ListDecoder = ListDecoder<Self::Decoder>;
1739}
1740
1741/// Encoder for `stream<T>`
1742#[derive(Default)]
1743pub struct StreamEncoder {
1744    deferred: Option<DeferredFn<Outgoing>>,
1745}
1746
1747impl Deferred<Outgoing> for StreamEncoder {
1748    fn take_deferred(&mut self) -> Option<DeferredFn<Outgoing>> {
1749        self.deferred.take()
1750    }
1751}
1752
1753impl<T, S> tokio_util::codec::Encoder<S> for StreamEncoder
1754where
1755    T: Encode + Send + 'static,
1756    S: Stream<Item = Vec<T>> + Send + Unpin + 'static,
1757    std::io::Error: From<<T::Encoder as tokio_util::codec::Encoder<T>>::Error>,
1758{
1759    type Error = std::io::Error;
1760
1761    #[instrument(level = "trace", skip(self, items), fields(ty = "stream"))]
1762    fn encode(&mut self, mut items: S, dst: &mut BytesMut) -> std::io::Result<()> {
1763        // TODO: Check if stream is resolved
1764        dst.reserve(1);
1765        dst.put_u8(0x00);
1766        let span = Span::current();
1767        self.deferred = Some(Box::new(|mut w, path| {
1768            Box::pin(async move {
1769                if !path.is_empty() {
1770                    w = w.index(&path).map_err(std::io::Error::other)?;
1771                }
1772                let mut enc = T::Encoder::default();
1773                let mut buf = BytesMut::default();
1774                let mut tasks = JoinSet::new();
1775                let mut i = 0_u64;
1776                loop {
1777                    select! {
1778                        chunk = items.next() => {
1779                            let Some(chunk) = chunk else {
1780                                trace!("writing stream end");
1781                                buf.reserve(1);
1782                                buf.put_u8(0x00);
1783                                w.write_all(&buf).await?;
1784                                while let Some(res) = tasks.join_next().await {
1785                                    trace!(?res, "receiver task finished");
1786                                    res??;
1787                                }
1788                                return Ok(())
1789                            };
1790                            let n = u32::try_from(chunk.len()).map_err(|err| {
1791                                std::io::Error::new(std::io::ErrorKind::InvalidInput, err)
1792                            })?;
1793                            let end = i.checked_add(n.into()).ok_or_else(|| {
1794                                std::io::Error::new(
1795                                    std::io::ErrorKind::InvalidInput,
1796                                    "stream element index would overflow u64",
1797                                )
1798                            })?;
1799                            trace!(n, "encoding chunk length");
1800                            Leb128Encoder.encode(n, &mut buf)?;
1801                            trace!(i, buf = format!("{buf:02x?}"), "writing stream chunk items");
1802
1803                            buf.reserve(chunk.len());
1804                            for (i, item) in zip(i.., chunk) {
1805                                enc.encode(item, &mut buf)?;
1806                                if let Some(f) = enc.take_deferred() {
1807                                    let i = i
1808                                        .try_into()
1809                                        .map_err(|err| std::io::Error::new(std::io::ErrorKind::InvalidInput, err))?;
1810                                    let w = w.index(&[i]).map_err(std::io::Error::other)?;
1811                                    trace!("spawning transmit task");
1812                                    tasks.spawn(f(w, Vec::default()));
1813                                }
1814                            }
1815                            i = end;
1816                        }
1817                        Some(res) = tasks.join_next() => {
1818                            trace!(?res, "receiver task finished");
1819                            res??;
1820                        }
1821                        res = w.write(&buf), if !buf.is_empty() => {
1822                            let n = res?;
1823                            trace!(?buf, n, "wrote bytes from buffer");
1824                            buf.advance(n);
1825                        }
1826                    }
1827                }
1828            }.instrument(span))
1829        }));
1830        Ok(())
1831    }
1832}
1833
1834impl<T> Encode for Pin<Box<dyn Stream<Item = Vec<T>> + Send>>
1835where
1836    T: Encode + Send + 'static,
1837    std::io::Error: From<<T::Encoder as tokio_util::codec::Encoder<T>>::Error>,
1838{
1839    type Encoder = StreamEncoder;
1840}
1841
1842/// Encoder for `stream<list<u8>>`
1843#[derive(Default)]
1844pub struct StreamEncoderBytes {
1845    deferred: Option<DeferredFn<Outgoing>>,
1846}
1847
1848impl Deferred<Outgoing> for StreamEncoderBytes {
1849    fn take_deferred(&mut self) -> Option<DeferredFn<Outgoing>> {
1850        self.deferred.take()
1851    }
1852}
1853
1854impl<S> tokio_util::codec::Encoder<S> for StreamEncoderBytes
1855where
1856    S: Stream<Item = Bytes> + Send + Unpin + 'static,
1857{
1858    type Error = std::io::Error;
1859
1860    #[instrument(level = "trace", skip(self, items), fields(ty = "stream<u8>"))]
1861    fn encode(&mut self, mut items: S, dst: &mut BytesMut) -> std::io::Result<()> {
1862        // TODO: Check if reader is resolved
1863        dst.reserve(1);
1864        dst.put_u8(0x00);
1865        self.deferred = Some(Box::new(|mut w, path| {
1866            Box::pin(async move {
1867                if !path.is_empty() {
1868                    w = w.index(&path).map_err(std::io::Error::other)?;
1869                }
1870                let mut buf = BytesMut::default();
1871                loop {
1872                    select! {
1873                        chunk = items.next() => {
1874                            let Some(chunk) = chunk else {
1875                                trace!("writing stream end");
1876                                buf.reserve(1);
1877                                buf.put_u8(0x00);
1878                                return w.write_all(&buf).await
1879                            };
1880                            let n = u32::try_from(chunk.len()).map_err(|err| {
1881                                std::io::Error::new(std::io::ErrorKind::InvalidInput, err)
1882                            })?;
1883                            trace!(n, "encoding chunk length");
1884                            Leb128Encoder.encode(n, &mut buf)?;
1885                            buf.extend_from_slice(&chunk);
1886                        }
1887                        res = w.write(&buf), if !buf.is_empty() => {
1888                            let n = res?;
1889                            buf.advance(n);
1890                        }
1891                    }
1892                }
1893            })
1894        }));
1895        Ok(())
1896    }
1897}
1898
1899impl Encode for Pin<Box<dyn Stream<Item = Bytes> + Send>> {
1900    type Encoder = StreamEncoderBytes;
1901}
1902
1903/// Encoder for `stream<list<u8>>` with [`AsyncRead`] support
1904#[derive(Default)]
1905pub struct StreamEncoderRead {
1906    deferred: Option<DeferredFn<Outgoing>>,
1907}
1908
1909impl Deferred<Outgoing> for StreamEncoderRead {
1910    fn take_deferred(&mut self) -> Option<DeferredFn<Outgoing>> {
1911        self.deferred.take()
1912    }
1913}
1914
1915impl<S> tokio_util::codec::Encoder<S> for StreamEncoderRead
1916where
1917    S: AsyncRead + Send + Unpin + 'static,
1918{
1919    type Error = std::io::Error;
1920
1921    #[instrument(level = "trace", skip(self, items), fields(ty = "stream<u8>"))]
1922    fn encode(&mut self, mut items: S, dst: &mut BytesMut) -> std::io::Result<()> {
1923        // TODO: Check if reader is resolved
1924        dst.reserve(1);
1925        dst.put_u8(0x00);
1926        self.deferred = Some(Box::new(|mut w, path| {
1927            Box::pin(async move {
1928                if !path.is_empty() {
1929                    w = w.index(&path).map_err(std::io::Error::other)?;
1930                }
1931                let mut buf = BytesMut::default();
1932                let mut chunk = BytesMut::default();
1933                loop {
1934                    select! {
1935                        res = items.read_buf(&mut chunk) => {
1936                            let n = res?;
1937                            if n == 0 {
1938                                trace!("writing stream end");
1939                                buf.reserve(1);
1940                                buf.put_u8(0x00);
1941                                return w.write_all(&buf).await
1942                            }
1943                            let n = u32::try_from(n).map_err(|err| {
1944                                std::io::Error::new(std::io::ErrorKind::InvalidInput, err)
1945                            })?;
1946                            trace!(n, "encoding chunk length");
1947                            Leb128Encoder.encode(n, &mut buf)?;
1948                            buf.extend_from_slice(&chunk);
1949                            chunk.clear();
1950                        }
1951                        res = w.write(&buf), if !buf.is_empty() => {
1952                            let n = res?;
1953                            buf.advance(n);
1954                        }
1955                    }
1956                }
1957            })
1958        }));
1959        Ok(())
1960    }
1961}
1962
1963impl Encode for Pin<Box<dyn AsyncRead + Send>> {
1964    type Encoder = StreamEncoderRead;
1965}
1966
1967impl<T> Encode for std::io::Cursor<T>
1968where
1969    T: AsRef<[u8]> + Send + Unpin + 'static,
1970{
1971    type Encoder = StreamEncoderRead;
1972}
1973
1974impl Encode for tokio::io::Empty {
1975    type Encoder = StreamEncoderRead;
1976}
1977
1978#[cfg(feature = "io-std")]
1979impl Encode for tokio::io::Stdin {
1980    type Encoder = StreamEncoderRead;
1981}
1982
1983#[cfg(feature = "fs")]
1984impl Encode for tokio::fs::File {
1985    type Encoder = StreamEncoderRead;
1986}
1987
1988#[cfg(feature = "net")]
1989impl Encode for tokio::net::TcpStream {
1990    type Encoder = StreamEncoderRead;
1991}
1992
1993#[cfg(all(unix, feature = "net"))]
1994impl Encode for tokio::net::UnixStream {
1995    type Encoder = StreamEncoderRead;
1996}
1997
1998#[cfg(all(unix, feature = "net"))]
1999impl Encode for tokio::net::unix::pipe::Receiver {
2000    type Encoder = StreamEncoderRead;
2001}
2002
2003/// Decoder for `stream<T>`
2004pub struct StreamDecoder<T>
2005where
2006    T: Decode,
2007{
2008    dec: T::ListDecoder,
2009    deferred: Option<DeferredFn<BufferedIncoming>>,
2010    _ty: PhantomData<T>,
2011}
2012
2013impl<T> Default for StreamDecoder<T>
2014where
2015    T: Decode,
2016{
2017    fn default() -> Self {
2018        Self {
2019            dec: T::ListDecoder::default(),
2020            deferred: None,
2021            _ty: PhantomData,
2022        }
2023    }
2024}
2025
2026impl<T> Deferred<BufferedIncoming> for StreamDecoder<T>
2027where
2028    T: Decode,
2029{
2030    fn take_deferred(&mut self) -> Option<DeferredFn<BufferedIncoming>> {
2031        self.deferred.take()
2032    }
2033}
2034
2035#[instrument(level = "trace", skip(dec, r, tx), ret)]
2036async fn handle_deferred_stream<C, T>(
2037    dec: C,
2038    mut r: BufferedIncoming,
2039    path: Vec<usize>,
2040    tx: mpsc::Sender<Vec<T>>,
2041) -> std::io::Result<()>
2042where
2043    C: tokio_util::codec::Decoder<Item = T> + Deferred<BufferedIncoming>,
2044    std::io::Error: From<C::Error>,
2045{
2046    let dec = ListDecoder::new(dec);
2047    if !path.is_empty() {
2048        r = r.index(&path).map_err(std::io::Error::other)?;
2049    }
2050    let mut framed = FramedRead::new(r, dec);
2051    let mut tasks = JoinSet::new();
2052    let mut i = 0_usize;
2053    loop {
2054        trace!("receiving pending stream chunk");
2055        select! {
2056            Some(chunk) = framed.next() => {
2057                let chunk = chunk?;
2058                if chunk.is_empty() {
2059                    trace!("received stream end");
2060                    while let Some(res) = tasks.join_next().await {
2061                        res??;
2062                    }
2063                    return Ok(())
2064                }
2065                let end = i.checked_add(chunk.len()).ok_or_else(|| {
2066                    std::io::Error::new(
2067                        std::io::ErrorKind::InvalidInput,
2068                        "stream element index would overflow usize",
2069                    )
2070                })?;
2071                trace!(i, end, "received stream chunk");
2072                if tx.send(chunk).await.is_err() {
2073                    debug!("stream receiver closed, discard data");
2074                    return Ok(())
2075                }
2076                for (i, deferred) in zip(i.., mem::take(&mut framed.decoder_mut().deferred)) {
2077                    if let Some(deferred) = deferred {
2078                        let r = framed.get_ref().index(&[i]).map_err(std::io::Error::other)?;
2079                        trace!("spawning receive task");
2080                        tasks.spawn(deferred(r, Vec::default()));
2081                    }
2082                }
2083                i = end;
2084            },
2085            Some(res) = tasks.join_next() => {
2086                trace!(?res, "receiver task finished");
2087                res??;
2088            }
2089            else => {
2090                return Ok(());
2091            }
2092        }
2093    }
2094}
2095
2096impl<T> tokio_util::codec::Decoder for StreamDecoder<T>
2097where
2098    T: Decode + Send + 'static,
2099    T::ListDecoder: Deferred<BufferedIncoming>,
2100    <T::Decoder as tokio_util::codec::Decoder>::Error: Send,
2101    std::io::Error: From<<T::Decoder as tokio_util::codec::Decoder>::Error>,
2102{
2103    type Item = Pin<Box<dyn Stream<Item = Vec<T>> + Send>>;
2104    type Error = <<T as Decode>::ListDecoder as tokio_util::codec::Decoder>::Error;
2105
2106    #[instrument(level = "trace", skip(self), fields(ty = "stream"))]
2107    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
2108        let Some(chunk) = self.dec.decode(src)? else {
2109            return Ok(None);
2110        };
2111        if !chunk.is_empty() {
2112            self.deferred = self.dec.take_deferred();
2113            return Ok(Some(Box::pin(stream::iter([chunk]))));
2114        }
2115
2116        // stream is pending
2117        let (tx, rx) = mpsc::channel(128);
2118        self.deferred = Some(Box::new(|r, path| {
2119            Box::pin(
2120                async move { handle_deferred_stream(T::Decoder::default(), r, path, tx).await },
2121            )
2122        }));
2123        Ok(Some(Box::pin(ReceiverStream::new(rx))))
2124    }
2125}
2126
2127impl<T> Decode for Pin<Box<dyn Stream<Item = Vec<T>> + Send>>
2128where
2129    T: Decode + Send + 'static,
2130    T::ListDecoder: Deferred<BufferedIncoming> + Send,
2131    <T::Decoder as tokio_util::codec::Decoder>::Error: Send,
2132    std::io::Error: From<<T::Decoder as tokio_util::codec::Decoder>::Error>,
2133{
2134    type Decoder = StreamDecoder<T>;
2135    type ListDecoder = ListDecoder<Self::Decoder>;
2136}
2137
2138/// Decoder for `stream<list<u8>>`
2139#[derive(Default)]
2140pub struct StreamDecoderBytes {
2141    dec: CoreVecDecoderBytes,
2142    deferred: Option<DeferredFn<BufferedIncoming>>,
2143}
2144
2145impl Deferred<BufferedIncoming> for StreamDecoderBytes {
2146    fn take_deferred(&mut self) -> Option<DeferredFn<BufferedIncoming>> {
2147        self.deferred.take()
2148    }
2149}
2150
2151impl tokio_util::codec::Decoder for StreamDecoderBytes {
2152    type Item = Pin<Box<dyn Stream<Item = Bytes> + Send>>;
2153    type Error = std::io::Error;
2154
2155    #[instrument(level = "trace", skip(self), fields(ty = "stream<u8>"))]
2156    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
2157        let Some(chunk) = self.dec.decode(src)? else {
2158            return Ok(None);
2159        };
2160        if !chunk.is_empty() {
2161            return Ok(Some(Box::pin(stream::iter([chunk]))));
2162        }
2163
2164        // stream is pending
2165        let (tx, rx) = mpsc::channel(128);
2166        let dec = mem::take(&mut self.dec);
2167        let span = Span::current();
2168        self.deferred = Some(Box::new(|mut r, path| {
2169            Box::pin(
2170                async move {
2171                    if !path.is_empty() {
2172                        r = r.index(&path).map_err(std::io::Error::other)?;
2173                    }
2174                    let mut framed = FramedRead::new(r, dec);
2175                    trace!(?path, "receiving pending byte stream chunk");
2176                    while let Some(chunk) = framed.next().await {
2177                        let chunk = chunk?;
2178                        if chunk.is_empty() {
2179                            trace!("received stream end");
2180                            return Ok(());
2181                        }
2182                        trace!(?chunk, "received pending byte stream chunk");
2183                        if tx.send(chunk).await.is_err() {
2184                            debug!("stream receiver closed, discard data");
2185                            return Ok(());
2186                        }
2187                    }
2188                    Ok(())
2189                }
2190                .instrument(span),
2191            )
2192        }));
2193        Ok(Some(Box::pin(ReceiverStream::new(rx))))
2194    }
2195}
2196
2197impl Decode for Pin<Box<dyn Stream<Item = Bytes> + Send>> {
2198    type Decoder = StreamDecoderBytes;
2199    type ListDecoder = ListDecoder<Self::Decoder>;
2200}
2201
2202/// Decoder for `stream<list<u8>>` with [`AsyncRead`] support
2203#[derive(Default)]
2204pub struct StreamDecoderRead {
2205    dec: CoreVecDecoderBytes,
2206    deferred: Option<DeferredFn<BufferedIncoming>>,
2207}
2208
2209impl Deferred<BufferedIncoming> for StreamDecoderRead {
2210    fn take_deferred(&mut self) -> Option<DeferredFn<BufferedIncoming>> {
2211        self.deferred.take()
2212    }
2213}
2214
2215impl tokio_util::codec::Decoder for StreamDecoderRead {
2216    type Item = Pin<Box<dyn AsyncRead + Send>>;
2217    type Error = std::io::Error;
2218
2219    #[instrument(level = "trace", skip(self), fields(ty = "stream<u8>"))]
2220    fn decode(&mut self, src: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
2221        let Some(chunk) = self.dec.decode(src)? else {
2222            return Ok(None);
2223        };
2224        if !chunk.is_empty() {
2225            return Ok(Some(Box::pin(std::io::Cursor::new(chunk))));
2226        }
2227
2228        // stream is pending
2229        let (tx, rx) = mpsc::channel(128);
2230        let dec = mem::take(&mut self.dec);
2231        self.deferred = Some(Box::new(|mut r, path| {
2232            Box::pin(async move {
2233                if !path.is_empty() {
2234                    r = r.index(&path).map_err(std::io::Error::other)?;
2235                }
2236                let mut framed = FramedRead::new(r, dec);
2237                trace!("receiving pending byte stream chunk");
2238                while let Some(chunk) = framed.next().await {
2239                    let chunk = chunk?;
2240                    if chunk.is_empty() {
2241                        trace!("received stream end");
2242                        return Ok(());
2243                    }
2244                    trace!(?chunk, "received byte stream chunk");
2245                    if tx.send(std::io::Result::Ok(chunk)).await.is_err() {
2246                        debug!("stream receiver closed, discard data");
2247                        return Ok(());
2248                    }
2249                }
2250                Ok(())
2251            })
2252        }));
2253        Ok(Some(Box::pin(StreamReader::new(ReceiverStream::new(rx)))))
2254    }
2255}
2256
2257impl Decode for Pin<Box<dyn AsyncRead + Send>> {
2258    type Decoder = StreamDecoderRead;
2259    type ListDecoder = ListDecoder<Self::Decoder>;
2260}
2261
2262#[cfg(test)]
2263mod tests {
2264    use anyhow::bail;
2265
2266    use super::*;
2267
2268    #[test_log::test(tokio::test)]
2269    async fn codec() -> anyhow::Result<()> {
2270        let mut buf = BytesMut::new();
2271        let mut enc = <(u8, u32) as Encode>::Encoder::default();
2272        enc.encode((0x42, 0x42), &mut buf)?;
2273        if let Some(_f) = Deferred::<Outgoing>::take_deferred(&mut enc) {
2274            bail!("no deferred write should have been returned");
2275        }
2276        assert_eq!(buf.as_ref(), b"\x42\x42");
2277        Ok(())
2278    }
2279
2280    #[test]
2281    fn canonical_nan_f32() {
2282        let mut enc = <f32 as Encode>::Encoder::default();
2283
2284        // A non-canonical (e.g. signalling) `NaN` is canonicalized on encode.
2285        let mut buf = BytesMut::new();
2286        enc.encode(f32::from_bits(0x7f80_0001), &mut buf).unwrap();
2287        assert_eq!(buf.as_ref(), CANONICAL_NAN_F32.to_le_bytes());
2288
2289        // A negative `NaN` is canonicalized to the (positive) canonical `NaN`.
2290        let mut buf = BytesMut::new();
2291        enc.encode(f32::from_bits(0xffc0_0000), &mut buf).unwrap();
2292        assert_eq!(buf.as_ref(), CANONICAL_NAN_F32.to_le_bytes());
2293
2294        // Non-`NaN` values are encoded unchanged.
2295        let mut buf = BytesMut::new();
2296        enc.encode(1.5_f32, &mut buf).unwrap();
2297        assert_eq!(buf.as_ref(), 1.5_f32.to_bits().to_le_bytes());
2298    }
2299
2300    #[test]
2301    fn canonical_nan_f64() {
2302        let mut enc = <f64 as Encode>::Encoder::default();
2303
2304        let mut buf = BytesMut::new();
2305        enc.encode(f64::from_bits(0x7ff0_0000_0000_0001), &mut buf)
2306            .unwrap();
2307        assert_eq!(buf.as_ref(), CANONICAL_NAN_F64.to_le_bytes());
2308
2309        let mut buf = BytesMut::new();
2310        enc.encode(f64::from_bits(0xfff8_0000_0000_0000), &mut buf)
2311            .unwrap();
2312        assert_eq!(buf.as_ref(), CANONICAL_NAN_F64.to_le_bytes());
2313
2314        let mut buf = BytesMut::new();
2315        enc.encode(1.5_f64, &mut buf).unwrap();
2316        assert_eq!(buf.as_ref(), 1.5_f64.to_bits().to_le_bytes());
2317    }
2318}
wrpc_transport/value.rs

wrpc_transport/
value.rs
