pub fn copy<'a, R: AsyncRead + Unpin + 'a>(
        &'a mut self,
        mut reader: R,
        buf: &'a mut [u8],
    ) -> impl Stream<Item = CopyEvent> + 'a {
        Gen::new(|co| {
            async move {
                let &mut Self { ref mut subscribed, ref mut drop_list } = self;
                let mut read;

                loop {
                    for indice in drop_list.drain(..) {
                        subscribed.remove(indice);
                    }

                    if subscribed.is_empty() {
                        co.yield_(CopyEvent::NoWriters).await;
                        break;
                    }

                    read = match reader.read(buf).await {
                        Ok(read) => read,
                        Err(why) => {
                            co.yield_(CopyEvent::SourceFailure(why)).await;
                            break;
                        }
                    };

                    if read == 0 {
                        break;
                    }

                    let slice = &buf[..read];

                    let mut stream = FuturesUnordered::from_iter(subscribed.iter_mut().map(
                        |(indice, writer)| async move { (indice, writer.write_all(slice).await) },
                    ));

                    while let Some((indice, why)) = stream.next().await {
                        if let Err(why) = why {
                            co.yield_(CopyEvent::Failure(indice, why)).await;
                            drop_list.push(indice);
                        }
                    }

                    co.yield_(CopyEvent::Progress(read)).await;
                }
            }
        })
    }
/// Copies bytes from the source to each writer concurrently.
///
/// # Notes
///
/// When a writer fails, it is removed from the slab.
Copies bytes from the source to each writer concurrently.
Copies bytes from the source to each writer concurrently .
pub fn copy<'a, R: AsyncRead + Unpin + 'a>(
        &'a mut self,
        mut reader: R,
        buf: &'a mut [u8],
    ) -> impl Stream<Item = CopyEvent> + 'a {
        Gen::new(|co| {
            async move {
                let &mut Self { ref mut subscribed, ref mut drop_list } = self;
                let mut read;
                loop {
                    for indice in drop_list.drain(..) {
                        subscribed.remove(indice);
                    }
                    if subscribed.is_empty() {
                        co.yield_(CopyEvent::NoWriters).await;
                        break;
                    }
                    read = match reader.read(buf).await {
                        Ok(read) => read,
                        Err(why) => {
                            co.yield_(CopyEvent::SourceFailure(why)).await;
                            break;
                        }
                    };
                    if read == 0 {
                        break;
                    }
                    let slice = &buf[..read];
                    let mut stream = FuturesUnordered::from_iter(subscribed.iter_mut().map(
                        |(indice, writer)| async move { (indice, writer.write_all(slice).await) },
                    ));
                    while let Some((indice, why)) = stream.next().await {
                        if let Err(why) = why {
                            co.yield_(CopyEvent::Failure(indice, why)).await;
                            drop_list.push(indice);
                        }
                    }
                    co.yield_(CopyEvent::Progress(read)).await;
                }
            }
        })
    }
pub fn copy < ' a , R : AsyncRead + Unpin + ' a > ( & ' a mut self , mut reader : R , buf : & ' a mut [ u8 ] , ) -> impl Stream < Item = CopyEvent > + ' a { Gen :: new ( | co | { async move { let & mut Self { ref mut subscribed , ref mut drop_list } = self ; let mut read ; loop { for indice in drop_list . drain ( .. ) { subscribed . remove ( indice ) ; } if subscribed . is_empty ( ) { co . yield_ ( CopyEvent :: NoWriters ) . await ; break ; } read = match reader . read ( buf ) . await { Ok ( read ) => read , Err ( why ) => { co . yield_ ( CopyEvent :: SourceFailure ( why ) ) . await ; break ; } } ; if read == 0 { break ; } let slice = & buf [ .. read ] ; let mut stream = FuturesUnordered :: from_iter ( subscribed . iter_mut ( ) . map ( | ( indice , writer ) | async move { ( indice , writer . write_all ( slice ) . await ) } , ) ) ; while let Some ( ( indice , why ) ) = stream . next ( ) . await { if let Err ( why ) = why { co . yield_ ( CopyEvent :: Failure ( indice , why ) ) . await ; drop_list . push ( indice ) ; } } co . yield_ ( CopyEvent :: Progress ( read ) ) . await ; } } } ) }
Copies bytes from the source to each writer concurrently.
Copies bytes from the source to each writer concurrently .
&'a mut self
reader
R
buf
&'a mut [u8]
&'a mut self
reader
R
buf
&'a mut [u8]
notes
When a writer fails, it is removed from the slab.
When a writer fails it is removed from the slab .