src/encoder/aio.rs

   1 //! Asynchronous `pxar` format handling.
   2
   3 use std::io;
   4 use std::path::Path;
   5 use std::pin::Pin;
   6 use std::task::{Context, Poll};
   7
   8 use crate::encoder::{self, LinkOffset, SeqWrite};
   9 use crate::format;
  10 use crate::Metadata;
  11
  12 /// Asynchronous `pxar` encoder.
  13 ///
  14 /// This is the `async` version of the `pxar` encoder.
  15 #[repr(transparent)]
  16 pub struct Encoder<'a, T: SeqWrite + 'a> {
  17     inner: encoder::EncoderImpl<'a, T>,
  18 }
  19
  20 #[cfg(feature = "tokio-io")]
  21 impl<'a, T: tokio::io::AsyncWrite + 'a> Encoder<'a, TokioWriter<T>> {
  22     /// Encode a `pxar` archive into a `tokio::io::AsyncWrite` output.
  23     #[inline]
  24     pub async fn from_tokio(
  25         output: T,
  26         metadata: &Metadata,
  27     ) -> io::Result<Encoder<'a, TokioWriter<T>>> {
  28         Encoder::new(TokioWriter::new(output), metadata).await
  29     }
  30 }
  31
  32 #[cfg(feature = "tokio-fs")]
  33 impl<'a> Encoder<'a, TokioWriter<tokio::fs::File>> {
  34     /// Convenience shortcut for `File::create` followed by `Encoder::from_tokio`.
  35     pub async fn create<'b, P: AsRef<Path>>(
  36         path: P,
  37         metadata: &'b Metadata,
  38     ) -> io::Result<Encoder<'a, TokioWriter<tokio::fs::File>>> {
  39         Encoder::new(
  40             TokioWriter::new(tokio::fs::File::create(path.as_ref()).await?),
  41             metadata,
  42         )
  43         .await
  44     }
  45 }
  46
  47 impl<'a, T: SeqWrite + 'a> Encoder<'a, T> {
  48     /// Create an asynchronous encoder for an output implementing our internal write interface.
  49     pub async fn new(output: T, metadata: &Metadata) -> io::Result<Encoder<'a, T>> {
  50         Ok(Self {
  51             inner: encoder::EncoderImpl::new(output, metadata).await?,
  52         })
  53     }
  54
  55     /// Create a new regular file in the archive. This returns a `File` object to which the
  56     /// contents have to be written out *completely*. Failing to do so will put the encoder into an
  57     /// error state.
  58     pub async fn create_file<'b, P: AsRef<Path>>(
  59         &'b mut self,
  60         metadata: &Metadata,
  61         file_name: P,
  62         file_size: u64,
  63     ) -> io::Result<File<'b>>
  64     where
  65         'a: 'b,
  66     {
  67         Ok(File {
  68             inner: self
  69                 .inner
  70                 .create_file(metadata, file_name.as_ref(), file_size)
  71                 .await?,
  72         })
  73     }
  74
  75     // /// Convenience shortcut to add a *regular* file by path including its contents to the archive.
  76     // pub async fn add_file<P, F>(
  77     //     &mut self,
  78     //     metadata: &Metadata,
  79     //     file_name: P,
  80     //     file_size: u64,
  81     //     content: &mut dyn tokio::io::Read,
  82     // ) -> io::Result<()>
  83     // where
  84     //     P: AsRef<Path>,
  85     //     F: AsAsyncReader,
  86     // {
  87     //     self.inner.add_file(
  88     //         metadata,
  89     //         file_name.as_ref(),
  90     //         file_size,
  91     //         content.as_async_reader(),
  92     //     ).await
  93     // }
  94
  95     /// Create a new subdirectory. Note that the subdirectory has to be finished by calling the
  96     /// `finish()` method, otherwise the entire archive will be in an error state.
  97     pub async fn create_directory<'b, P: AsRef<Path>>(
  98         &'b mut self,
  99         file_name: P,
 100         metadata: &Metadata,
 101     ) -> io::Result<Encoder<'b, &'b mut dyn SeqWrite>>
 102     where
 103         'a: 'b,
 104     {
 105         Ok(Encoder {
 106             inner: self
 107                 .inner
 108                 .create_directory(file_name.as_ref(), metadata)
 109                 .await?,
 110         })
 111     }
 112
 113     /// Finish this directory. This is mandatory, otherwise the `Drop` handler will `panic!`.
 114     pub async fn finish(self) -> io::Result<T> {
 115         self.inner.finish().await
 116     }
 117
 118     /// Cancel this directory and get back the contained writer.
 119     pub fn into_writer(self) -> T {
 120         self.inner.into_writer()
 121     }
 122
 123     /// Add a symbolic link to the archive.
 124     pub async fn add_symlink<PF: AsRef<Path>, PT: AsRef<Path>>(
 125         &mut self,
 126         metadata: &Metadata,
 127         file_name: PF,
 128         target: PT,
 129     ) -> io::Result<()> {
 130         self.inner
 131             .add_symlink(metadata, file_name.as_ref(), target.as_ref())
 132             .await
 133     }
 134
 135     /// Add a hard link to the archive.
 136     pub async fn add_hardlink<PF: AsRef<Path>, PT: AsRef<Path>>(
 137         &mut self,
 138         file_name: PF,
 139         target: PT,
 140         offset: LinkOffset,
 141     ) -> io::Result<()> {
 142         self.inner
 143             .add_hardlink(file_name.as_ref(), target.as_ref(), offset)
 144             .await
 145     }
 146
 147     /// Add a device node to the archive.
 148     pub async fn add_device<P: AsRef<Path>>(
 149         &mut self,
 150         metadata: &Metadata,
 151         file_name: P,
 152         device: format::Device,
 153     ) -> io::Result<()> {
 154         self.inner
 155             .add_device(metadata, file_name.as_ref(), device)
 156             .await
 157     }
 158
 159     /// Add a device node to the archive.
 160     pub async fn add_fifo<P: AsRef<Path>>(
 161         &mut self,
 162         metadata: &Metadata,
 163         file_name: P,
 164     ) -> io::Result<()> {
 165         self.inner.add_fifo(metadata, file_name.as_ref()).await
 166     }
 167
 168     /// Add a device node to the archive.
 169     pub async fn add_socket<P: AsRef<Path>>(
 170         &mut self,
 171         metadata: &Metadata,
 172         file_name: P,
 173     ) -> io::Result<()> {
 174         self.inner.add_socket(metadata, file_name.as_ref()).await
 175     }
 176 }
 177
 178 #[repr(transparent)]
 179 pub struct File<'a> {
 180     inner: encoder::FileImpl<'a>,
 181 }
 182
 183 impl<'a> File<'a> {
 184     /// Get the file offset to be able to reference it with `add_hardlink`.
 185     pub fn file_offset(&self) -> LinkOffset {
 186         self.inner.file_offset()
 187     }
 188
 189     /// Write file data for the current file entry in a pxar archive.
 190     pub async fn write(&mut self, data: &[u8]) -> io::Result<usize> {
 191         self.inner.write(data).await
 192     }
 193
 194     /// Completely write file data for the current file entry in a pxar archive.
 195     pub async fn write_all(&mut self, data: &[u8]) -> io::Result<()> {
 196         self.inner.write_all(data).await
 197     }
 198 }
 199
 200 #[cfg(feature = "tokio-io")]
 201 impl<'a> tokio::io::AsyncWrite for File<'a> {
 202     fn poll_write(self: Pin<&mut Self>, cx: &mut Context, data: &[u8]) -> Poll<io::Result<usize>> {
 203         unsafe { self.map_unchecked_mut(|this| &mut this.inner) }.poll_write(cx, data)
 204     }
 205
 206     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
 207         unsafe { self.map_unchecked_mut(|this| &mut this.inner) }.poll_flush(cx)
 208     }
 209
 210     fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
 211         unsafe { self.map_unchecked_mut(|this| &mut this.inner) }.poll_close(cx)
 212     }
 213 }
 214
 215 /// Pxar encoder write adapter for `tokio::io::AsyncWrite`.
 216 #[cfg(feature = "tokio-io")]
 217 mod tokio_writer {
 218     use std::io;
 219     use std::pin::Pin;
 220     use std::task::{Context, Poll};
 221
 222     use crate::encoder::SeqWrite;
 223
 224     pub struct TokioWriter<T> {
 225         inner: Option<T>,
 226     }
 227
 228     impl<T: tokio::io::AsyncWrite> TokioWriter<T> {
 229         pub fn new(inner: T) -> Self {
 230             Self { inner: Some(inner) }
 231         }
 232
 233         fn inner_mut(&mut self) -> io::Result<Pin<&mut T>> {
 234             let inner = self
 235                 .inner
 236                 .as_mut()
 237                 .ok_or_else(|| io_format_err!("write after close"))?;
 238             Ok(unsafe { Pin::new_unchecked(inner) })
 239         }
 240
 241         fn inner(self: Pin<&mut Self>) -> io::Result<Pin<&mut T>> {
 242             unsafe { self.get_unchecked_mut() }.inner_mut()
 243         }
 244     }
 245
 246     impl<T: tokio::io::AsyncWrite> SeqWrite for TokioWriter<T> {
 247         fn poll_seq_write(
 248             self: Pin<&mut Self>,
 249             cx: &mut Context,
 250             buf: &[u8],
 251         ) -> Poll<io::Result<usize>> {
 252             let this = unsafe { self.get_unchecked_mut() };
 253             this.inner_mut()?.poll_write(cx, buf)
 254         }
 255
 256         fn poll_flush(self: Pin<&mut Self>, cx: &mut Context) -> Poll<io::Result<()>> {
 257             self.inner()?.poll_flush(cx)
 258         }
 259     }
 260 }
 261
 262 #[cfg(feature = "tokio-io")]
 263 pub use tokio_writer::TokioWriter;
 264
 265 #[cfg(test)]
 266 mod test {
 267     use std::io;
 268     use std::pin::Pin;
 269     use std::task::{Context, Poll};
 270
 271     use super::Encoder;
 272     use crate::Metadata;
 273
 274     struct DummyOutput;
 275
 276     impl super::SeqWrite for DummyOutput {
 277         fn poll_seq_write(
 278             self: Pin<&mut Self>,
 279             _cx: &mut Context,
 280             _buf: &[u8],
 281         ) -> Poll<io::Result<usize>> {
 282             unreachable!();
 283         }
 284
 285         fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context) -> Poll<io::Result<()>> {
 286             unreachable!();
 287         }
 288     }
 289
 290     #[test]
 291     /// Assert that `Encoder` is `Send`
 292     fn send_test() {
 293         let test = async {
 294             let mut encoder = Encoder::new(
 295                 DummyOutput,
 296                 &Metadata::dir_builder(0o700).build(),
 297             )
 298             .await
 299             .unwrap();
 300             encoder
 301                 .create_directory("baba", &Metadata::dir_builder(0o700).build())
 302                 .await
 303                 .unwrap();
 304         };
 305
 306         fn test_send<T: Send>(_: T) {}
 307         test_send(test);
 308     }
 309 }