src/decoder/aio.rs

   1 //! Asynchronous `pxar` format handling.
   2
   3 use std::io;
   4
   5 #[cfg(feature = "tokio-fs")]
   6 use std::path::Path;
   7
   8 use crate::decoder::{self, Contents, SeqRead};
   9 use crate::Entry;
  10
  11 /// Asynchronous `pxar` decoder.
  12 ///
  13 /// This is the `async` version of the `pxar` decoder.
  14 #[repr(transparent)]
  15 pub struct Decoder<T> {
  16     inner: decoder::DecoderImpl<T>,
  17 }
  18
  19 #[cfg(feature = "tokio-io")]
  20 impl<T: tokio::io::AsyncRead> Decoder<TokioReader<T>> {
  21     /// Decode a `pxar` archive from a `tokio::io::AsyncRead` input.
  22     #[inline]
  23     pub async fn from_tokio(input: T) -> io::Result<Self> {
  24         Decoder::new(TokioReader::new(input)).await
  25     }
  26 }
  27
  28 #[cfg(feature = "tokio-fs")]
  29 impl Decoder<TokioReader<tokio::fs::File>> {
  30     /// Decode a `pxar` archive from a `tokio::io::AsyncRead` input.
  31     #[inline]
  32     pub async fn open<P: AsRef<Path>>(path: P) -> io::Result<Self> {
  33         Decoder::from_tokio(tokio::fs::File::open(path.as_ref()).await?).await
  34     }
  35 }
  36
  37 impl<T: SeqRead> Decoder<T> {
  38     /// Create an async decoder from an input implementing our internal read interface.
  39     pub async fn new(input: T) -> io::Result<Self> {
  40         Ok(Self {
  41             inner: decoder::DecoderImpl::new(input).await?,
  42         })
  43     }
  44
  45     /// Internal helper for `Accessor`. In this case we have the low-level state machine, and the
  46     /// layer "above" the `Accessor` propagates the actual type (sync vs async).
  47     pub(crate) fn from_impl(inner: decoder::DecoderImpl<T>) -> Self {
  48         Self { inner }
  49     }
  50
  51     // I would normally agree with clippy, but this is async and we can at most implement Stream,
  52     // which we do with feature flags...
  53     #[allow(clippy::should_implement_trait)]
  54     /// If this is a directory entry, get the next item inside the directory.
  55     pub async fn next(&mut self) -> Option<io::Result<Entry>> {
  56         self.inner.next_do().await.transpose()
  57     }
  58
  59     /// Get a reader for the contents of the current entry, if the entry has contents.
  60     pub fn contents(&mut self) -> Option<Contents<T>> {
  61         self.inner.content_reader()
  62     }
  63
  64     /// Get the size of the current contents, if the entry has contents.
  65     pub fn content_size(&self) -> Option<u64> {
  66         self.inner.content_size()
  67     }
  68
  69     /// Include goodbye tables in iteration.
  70     pub fn enable_goodbye_entries(&mut self, on: bool) {
  71         self.inner.with_goodbye_tables = on;
  72     }
  73 }
  74
  75 #[cfg(feature = "tokio-io")]
  76 mod tok {
  77     use crate::decoder::{Contents, SeqRead};
  78     use std::io;
  79     use std::pin::Pin;
  80     use std::task::{Context, Poll};
  81
  82     /// Read adapter for `futures::io::AsyncRead`
  83     pub struct TokioReader<T> {
  84         inner: T,
  85     }
  86
  87     impl<T: tokio::io::AsyncRead> TokioReader<T> {
  88         pub fn new(inner: T) -> Self {
  89             Self { inner }
  90         }
  91     }
  92
  93     impl<T: tokio::io::AsyncRead> crate::decoder::SeqRead for TokioReader<T> {
  94         fn poll_seq_read(
  95             self: Pin<&mut Self>,
  96             cx: &mut Context,
  97             buf: &mut [u8],
  98         ) -> Poll<io::Result<usize>> {
  99             let mut read_buf = tokio::io::ReadBuf::new(buf);
 100             unsafe {
 101                 self.map_unchecked_mut(|this| &mut this.inner)
 102                     .poll_read(cx, &mut read_buf)
 103                     .map_ok(|_| read_buf.filled().len())
 104             }
 105         }
 106     }
 107
 108     impl<'a, T: crate::decoder::SeqRead> tokio::io::AsyncRead for Contents<'a, T> {
 109         fn poll_read(
 110             self: Pin<&mut Self>,
 111             cx: &mut Context<'_>,
 112             buf: &mut tokio::io::ReadBuf<'_>,
 113         ) -> Poll<io::Result<()>> {
 114             unsafe {
 115                 // Safety: poll_seq_read will *probably* only write to the buffer, so we don't
 116                 // initialize it first, instead we treat is a &[u8] immediately and uphold the
 117                 // ReadBuf invariants in the conditional below.
 118                 let write_buf =
 119                     &mut *(buf.unfilled_mut() as *mut [std::mem::MaybeUninit<u8>] as *mut [u8]);
 120                 let result = self.poll_seq_read(cx, write_buf);
 121                 if let Poll::Ready(Ok(n)) = result {
 122                     // if we've written data, advance both initialized and filled bytes cursor
 123                     buf.assume_init(n);
 124                     buf.advance(n);
 125                 }
 126                 result.map(|_| Ok(()))
 127             }
 128         }
 129     }
 130 }
 131
 132 #[cfg(feature = "tokio-io")]
 133 use tok::TokioReader;