pub async fn write_stream<E, T>(
        &self,
        inv: crate::Invoice,
        parcels: HashMap<String, T>,
    ) -> Result<()>
    where
        E: std::error::Error,
        T: Stream<Item = std::result::Result<bytes::Bytes, E>> + Unpin,
    {
        validate_shas(&inv, parcels.keys())?;

        // Should create all the directories needed down to the parcels directory
        tokio::fs::create_dir_all(self.base_path.join(PARCEL_DIR)).await?;

        write_invoice(&self.base_path, &inv).await?;

        let parcel_writes = parcels.into_iter().map(|(sha, mut stream)| async move {
            let path = self.base_path.join(PARCEL_DIR).join(format!("{}.dat", sha));

            let mut file = tokio::fs::OpenOptions::new()
                .write(true)
                .create_new(true) // Make sure we aren't overwriting
                .open(&path)
                .await?;

            debug!("Writing parcel to {}", path.display());

            while let Some(b) = stream.next().await {
                let b =
                    b.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;
                file.write_all(&b).await?;
            }
            file.flush().await?;

            debug!("Finished writing parcel to {}", path.display());
            Ok(())
        });
        futures::future::join_all(parcel_writes)
            .await
            .into_iter()
            .collect::<std::io::Result<Vec<_>>>()?;

        Ok(())
    }
/// Writes the given invoice and collection of parcel streams
Writes the given invoice and collection of parcel streams
Writes the given invoice and collection of parcel streams
pub async fn write_stream<E, T>(
        &self,
        inv: crate::Invoice,
        parcels: HashMap<String, T>,
    ) -> Result<()>
    where
        E: std::error::Error,
        T: Stream<Item = std::result::Result<bytes::Bytes, E>> + Unpin,
    {
        validate_shas(&inv, parcels.keys())?;
        // Should create all the directories needed down to the parcels directory
        tokio::fs::create_dir_all(self.base_path.join(PARCEL_DIR)).await?;
        write_invoice(&self.base_path, &inv).await?;
        let parcel_writes = parcels.into_iter().map(|(sha, mut stream)| async move {
            let path = self.base_path.join(PARCEL_DIR).join(format!("{}.dat", sha));
            let mut file = tokio::fs::OpenOptions::new()
                .write(true)
                .create_new(true) // Make sure we aren't overwriting
                .open(&path)
                .await?;
            debug!("Writing parcel to {}", path.display());
            while let Some(b) = stream.next().await {
                let b =
                    b.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;
                file.write_all(&b).await?;
            }
            file.flush().await?;
            debug!("Finished writing parcel to {}", path.display());
            Ok(())
        });
        futures::future::join_all(parcel_writes)
            .await
            .into_iter()
            .collect::<std::io::Result<Vec<_>>>()?;
        Ok(())
    }
pub async fn write_stream < E , T > ( & self , inv : crate :: Invoice , parcels : HashMap < String , T > , ) -> Result < ( ) > where E : std :: error :: Error , T : Stream < Item = std :: result :: Result < bytes :: Bytes , E > > + Unpin , { validate_shas ( & inv , parcels . keys ( ) ) ? ; // Should create all the directories needed down to the parcels directory tokio :: fs :: create_dir_all ( self . base_path . join ( PARCEL_DIR ) ) . await ? ; write_invoice ( & self . base_path , & inv ) . await ? ; let parcel_writes = parcels . into_iter ( ) . map ( | ( sha , mut stream ) | async move { let path = self . base_path . join ( PARCEL_DIR ) . join ( format ! ( " " sha ) ) ; let mut file = tokio :: fs :: OpenOptions :: new ( ) . write ( true ) . create_new ( true ) // Make sure we aren't overwriting . open ( & path ) . await ? ; debug ! ( " " path display ( ) ) ; while let Some ( b ) = stream . next ( ) . await { let b = b . map_err ( | e | std :: io :: Error :: new ( std :: io :: ErrorKind :: Other , e . to_string ( ) ) ) ? ; file . write_all ( & b ) . await ? ; } file . flush ( ) . await ? ; debug ! ( " " path display ( ) ) ; Ok ( ( ) ) } ) ; futures :: future :: join_all ( parcel_writes ) . await . into_iter ( ) . collect :: < std :: io :: Result < Vec < _ > > > ( ) ? ; Ok ( ( ) ) }
Writes the given invoice and collection of parcel streams
Writes the given invoice and collection of parcel streams
&self
inv
crate::Invoice
parcels
HashMap<String, T>
&self
inv
crate::Invoice
parcels
HashMap<String, T>