fae22d07f7d386b89f14ac353b079c9eef71f948

│ ●   fae22d0   Edit project name   bob      z6Mkt67…v4N1tRk   active     now     │

$ rad id accept fae22d0 -q

│ ●   fae22d0   Edit project name   bob      z6Mkt67…v4N1tRk   accepted   now     │

✓ Identity revision 9d9031417f1d86a6c0ed5ec2c4bf5820dca0eec9 created

│ Revision 9d9031417f1d86a6c0ed5ec2c4bf5820dca0eec9                      │

$ rad id accept 9d9031417f1d86a6c0ed5ec2c4bf5820dca0eec9 --repo rad:z42hL2jL4XNk6K8oHQaSWfMgCL7ji --no-confirm

✓ Revision 9d9031417f1d86a6c0ed5ec2c4bf5820dca0eec9 accepted

│ Revision 9d9031417f1d86a6c0ed5ec2c4bf5820dca0eec9                      │

1483814d54ad1321fc4ddb1f8bf7d90454de1790

│ ●   1483814   Make private       alice    (you)             active     now     │

│ ●   9d90314   Add Eve            bob      z6Mkt67…v4N1tRk   accepted   now     │

$ rad id redact 1483814d54ad1321fc4ddb1f8bf7d90454de1790

$ rad id edit --title "Boo!" --description "Boo!" 1483814d54ad1321fc4ddb1f8bf7d90454de1790

$ rad id edit --title "Make private" --description "Privacy is cool." 1483814d54ad1321fc4ddb1f8bf7d90454de1790

✓ Revision 1483814d54ad1321fc4ddb1f8bf7d90454de1790 edited

$ rad id show 1483814d54ad1321fc4ddb1f8bf7d90454de1790

│ Revision 1483814d54ad1321fc4ddb1f8bf7d90454de1790                      │

$ rad id redact 1483814d54ad1321fc4ddb1f8bf7d90454de1790

✓ Revision 1483814d54ad1321fc4ddb1f8bf7d90454de1790 redacted

$ rad id show 1483814d54ad1321fc4ddb1f8bf7d90454de1790

✗ Error: revision `1483814d54ad1321fc4ddb1f8bf7d90454de1790` not found

✓ Identity revision ea60049b8265f60f3dcca21798ce50ef67779421 created

│ Revision ea60049b8265f60f3dcca21798ce50ef67779421                      │

✗ Fetching rad:z2ug5mwNKZB8KGpBDRTrWHAMbvHCu from z6MknSL…StBU8Vi.. error: connection reset

✓ Patch 6d260fc8388e74d8fefb5dabc5a798e125ec3cf9 opened

## bob/feature...rad/patches/6d260fc8388e74d8fefb5dabc5a798e125ec3cf9

$ rad patch checkout 6d260fc8388e74d8fefb5dabc5a798e125ec3cf9

✓ Switched to branch patch/6d260fc

✓ Branch patch/6d260fc setup to track rad/patches/6d260fc8388e74d8fefb5dabc5a798e125ec3cf9

✓ Patch 6d260fc updated to 750081b35a3f831f428653bd2240eb4674ccae71

   bdcdb30..cad2666  bob/feature -> patches/6d260fc8388e74d8fefb5dabc5a798e125ec3cf9

$ rad patch show 6d260fc

│ Patch    6d260fc8388e74d8fefb5dabc5a798e125ec3cf9                   │

│ ↑ updated to 750081b35a3f831f428653bd2240eb4674ccae71 (cad2666) now │

cad2666a8a2250e4dee175ed5044be2c251ff08b	refs/heads/patches/6d260fc8388e74d8fefb5dabc5a798e125ec3cf9

## patch/6d260fc...rad/patches/6d260fc8388e74d8fefb5dabc5a798e125ec3cf9 [behind 1]

│ Issue   d457c205de780d37d1c16efbe5333aed4662a6c3        │

│ ●   d457c20   flux capacitor underpowered   bob      (you)                        now    │

$ rad issue comment d457c205de780d37d1c16efbe5333aed4662a6c3 --message 'The flux capacitor needs 1.21 Gigawatts' -q

2e8b9538aedaea418e90e90d19e61edf3f022933

✓ Patch 4bfb6fe940f815e3fcce6a2796e051df85db9fe1 opened

│ ●  4bfb6fe  Define power requirements  bob     (you)  3e674d1  +0  -0  now     │

$ rad patch show 4bfb6fe940f815e3fcce6a2796e051df85db9fe1

│ Patch     4bfb6fe940f815e3fcce6a2796e051df85db9fe1 │

3e674d1a1df90807e934f9ae5da2591dd6848a33	refs/heads/patches/4bfb6fe940f815e3fcce6a2796e051df85db9fe1

✓ Patch 4bfb6fe updated to 7782e60eb51b6e852abb184b092249327354c625

   3e674d1..27857ec  flux-capacitor-power -> patches/4bfb6fe940f815e3fcce6a2796e051df85db9fe1

$ rad patch comment 4bfb6fe940f815e3fcce6a2796e051df85db9fe1 --message 'I cannot wait to get back to the 90s!' -q

a3ba1df99fbd874affc790c95cd18607940f1a90

 * [new branch]      patches/4bfb6fe940f815e3fcce6a2796e051df85db9fe1 -> bob/patches/4bfb6fe940f815e3fcce6a2796e051df85db9fe1

  bob/patches/4bfb6fe940f815e3fcce6a2796e051df85db9fe1

$ rad patch show 4bfb6fe

│ Patch    4bfb6fe940f815e3fcce6a2796e051df85db9fe1                   │

│ ↑ updated to 7782e60eb51b6e852abb184b092249327354c625 (27857ec) now │

$ rad patch checkout 4bfb6fe940f815e3fcce6a2796e051df85db9fe1

✓ Switched to branch patch/4bfb6fe

✓ Branch patch/4bfb6fe setup to track rad/patches/4bfb6fe940f815e3fcce6a2796e051df85db9fe1

[patch/4bfb6fe f567f69] Use markdown for requirements

✓ Patch 4bfb6fe updated to fab4fddf6bcae7d55432417cdf5a7d0270d0d7d3

 * [new branch]      patch/4bfb6fe -> patches/4bfb6fe940f815e3fcce6a2796e051df85db9fe1

$ git merge patch/4bfb6fe

✓ Patch 4bfb6fe940f815e3fcce6a2796e051df85db9fe1 merged at revision fab4fdd

$ rad patch show 4bfb6fe

│ Patch    4bfb6fe940f815e3fcce6a2796e051df85db9fe1                   │

│ ↑ updated to 7782e60eb51b6e852abb184b092249327354c625 (27857ec) now │

│ * revised by alice (you) in fab4fdd (f567f69) now                   │

│ ✓ merged by alice (you) at revision fab4fdd (f567f69) now           │

        let tracking = tracking::Store::open(tracking_db)?;

        let tracking = tracking::Config::new(config.policy, config.scope, tracking);

        log::info!(target: "node", "Default tracking policy set to '{}'", &config.policy);

        let mut wire = Wire::new(service, worker_send, signer, proxy, clock);

            id,

                atomic,

pub use radicle::node::tracking::config as tracking;

use crate::storage::{ReadRepository, RefUpdate, RemoteRepository as _};

        result: Result<(Vec<RefUpdate>, HashSet<NodeId>), FetchError>,

            Ok((updated, namespaces)) => {

use crate::storage::{Namespaces, RefUpdate};

use crate::worker::FetchError;

    Fetched(

        Id,

        NodeId,

        Rc<Result<(Vec<RefUpdate>, HashSet<NodeId>), FetchError>>,

    ),

                            Ok((_, remotes)) => {

                                    Namespaces::Trusted(remotes.clone()),

                                Rc::new(Ok((

                                    vec![],

                                    match namespaces {

                                ))),

    alice.fetched(rid1, bob.id, Ok((vec![], Default::default())));

    alice.fetched(rid2, bob.id, Ok((vec![], Default::default())));

    ReadRepository, ReadStorage, RemoteRepository, ValidateRepository, WriteRepository,

// TODO: ignoring as this is being fixed in incoming fetch changes

#[ignore]

fn test_replication_no_delegates() {

    assert_matches!(

        result,

        FetchResult::Failed {

            reason

        } if reason == "no delegates in transfer"

    assert_eq!(

        Some((vec![], HashSet::from_iter([bob.id])))

        FetchResult::Success { .. }

                    namespaces,

                            namespaces,

mod fetch;

mod tunnel;

use std::collections::{BTreeSet, HashSet};

use std::io::{prelude::*, BufReader};

use std::ops::ControlFlow;

use std::{env, io, net, process, time};

use radicle::storage::{Namespaces, ReadRepository, ReadStorage, RefUpdate};

use radicle::{git, storage, Storage};

use channels::{ChannelReader, ChannelWriter};

use tunnel::Tunnel;

    /// Whether to use atomic fetches.

    pub atomic: bool,

    StagingInit(#[from] fetch::error::Init),

    StagingTransition(#[from] fetch::error::Transition),

    StagingTransfer(#[from] fetch::error::Transfer),

    /// Client is initiating a fetch in order to receive the specified

    /// `refspecs` determined by [`Namespaces`].

        /// Namespaces to fetch.

        namespaces: Namespaces,

        result: Result<(Vec<RefUpdate>, HashSet<NodeId>), FetchError>,

    daemon: net::SocketAddr,

    timeout: time::Duration,

    atomic: bool,

        mut channels: Channels,

                namespaces,

                timeout,

                let result = self.fetch(rid, remote, stream, &namespaces, channels, timeout);

                log::debug!(target: "worker", "Worker processing incoming fetch..");

                let (stream_w, stream_r) = channels.split();

                // Nb. two fetches are usually expected: one for the *special* refs,

                // followed by another for the signed refs.

                let result = loop {

                    match self.upload_pack(remote, stream, stream_r, stream_w) {

                        Ok(ControlFlow::Continue(())) => continue,

                        Ok(ControlFlow::Break(rid)) => break Ok(rid),

                        Err(e) => break Err(e),

                log::debug!(target: "worker", "Upload process on stream {stream} exited with result {result:?}");

                FetchResult::Responder { result }

            }

        }

    }

    fn fetch(

        &mut self,

        rid: Id,

        remote: NodeId,

        stream: StreamId,

        namespaces: &Namespaces,

        mut channels: Channels,

        timeout: time::Duration,

    ) -> Result<(Vec<RefUpdate>, HashSet<NodeId>), FetchError> {

        let staging =

            fetch::StagingPhaseInitial::new(&self.storage, rid, self.nid, namespaces.clone())?;

        let refs = if staging.repo.is_cloning() {

            match self._fetch(

                &staging.repo,

                staging.repo.is_cloning(),

                remote,

                staging.refspecs(),

                stream,

                &mut channels,

                timeout,

            ) {

                Ok(_) => {

                    log::debug!(target: "worker", "Initial fetch for {rid} exited successfully")

                }

                Err(e) => {

                    log::error!(target: "worker", "Initial fetch for {rid} failed: {e}");

                    return Err(e);

            }

            // TODO(finto): when cloning we simply fetch the special

            // rad refs from the remote side, however, when the

            // repository already exists we need to `ls-remote` (see

            // below). The result of the ls-remote is a BTreeSet of

            // refs and so we need return an empty set here so that we

            // can pass them into `into_final`. This is seems like a

            // code smell to me due to bad boundaries between the

            // logic in this module and the logic in the fetch module.

            BTreeSet::new()

        } else {

            self.ls_refs(

                &staging.repo,

                staging.ls_remote_refs(),

                remote,

                stream,

                &mut channels,

            )?

        };

        let staging = staging.into_final(refs)?;

        match self._fetch(

            &staging.repo,

            staging.repo.is_cloning(),

            remote,

            staging.refspecs(),

            stream,

            &mut channels,

            timeout,

        ) {

            Ok(()) => log::debug!(target: "worker", "Final fetch for {rid} exited successfully"),

            Err(e) => {

                log::error!(target: "worker", "Final fetch for {rid} failed: {e}");

                return Err(e);

        staging.transfer().map_err(FetchError::from)

    fn upload_pack(

        &mut self,

        remote: NodeId,

        stream: StreamId,

        stream_r: &mut ChannelReader,

        stream_w: &mut ChannelWriter,

    ) -> Result<ControlFlow<()>, UploadError> {

        log::debug!(target: "worker", "Waiting for Git request pktline from {remote}..");

        // Read the request packet line to know what repository we're uploading.

        let (rid, request) = match pktline::Reader::new(stream_r).read_request_pktline() {

            Ok((req, pktline)) => (req.repo, pktline),

            Err(err) if err.kind() == io::ErrorKind::ConnectionReset => {

                log::debug!(

                    target: "worker",

                    "Upload process received stream `close` from {remote}"

                );

                return Ok(ControlFlow::Break(()));

            }

            Err(err) => {

                return Err(UploadError::PacketLine(err));

            }

        };

        log::debug!(target: "worker", "Received Git request pktline for {rid}..");

        let doc = repo.identity_doc()?;

            return Err(UploadError::Unauthorized(remote, rid));

        }

        match self._upload_pack(rid, remote, request, stream, stream_r, stream_w) {

            Ok(()) => {

                log::debug!(target: "worker", "Upload of {rid} to {remote} on stream {stream} exited successfully");

                Ok(ControlFlow::Continue(()))

            }

            Err(e) => Err(e),

    fn _upload_pack(

        request: Vec<u8>,

        stream: StreamId,

        stream_r: &mut ChannelReader,

        stream_w: &mut ChannelWriter,

    ) -> Result<(), UploadError> {

        log::debug!(target: "worker", "Connecting to daemon..");

        // Connect to our local git daemon, running as a child process.

        let daemon = net::TcpStream::connect_timeout(&self.daemon, self.timeout)

            .map_err(UploadError::DaemonConnectionFailed)?;

        let (mut daemon_r, mut daemon_w) = (daemon.try_clone()?, daemon);

        daemon_r.set_read_timeout(Some(self.timeout))?;

        daemon_w.set_write_timeout(Some(self.timeout))?;

        // Write the raw request to the daemon, once we've parsed it.

        daemon_w.write_all(&request)?;

        log::debug!(target: "worker", "Entering Git protocol loop for {rid}..");

        thread::scope(|s| {

            let daemon_to_stream = thread::spawn_scoped(&self.nid, "upload-pack", s, || {

                let mut buffer = [0; u16::MAX as usize + 1];

                loop {

                    match daemon_r.read(&mut buffer) {

                        Ok(0) => break,

                        Ok(n) => {

                            stream_w.send(buffer[..n].to_vec())?;

                            if let Err(e) = self.handle.flush(remote, stream) {

                                log::error!(target: "worker", "Worker channel disconnected; aborting");

                                return Err(e);

                            }

                        }

                        Err(e) => {

                            if e.kind() == io::ErrorKind::UnexpectedEof {

                                log::debug!(target: "worker", "Daemon closed the git connection for {rid}");

                                break;

                            }

                            return Err(e);

                        }

                    }

                }

                Self::eof(remote, stream, stream_w, &mut self.handle)

            });

            let stream_to_daemon = thread::spawn_scoped(&self.nid, "upload-pack", s, move || {

                match stream_r

                    .pipe(&mut daemon_w)

                    .and_then(|()| daemon_w.shutdown(net::Shutdown::Both))

                {

                    Ok(()) => Ok(()),

                    // On macOS, this error is returned if the socket is already closed.

                    // We don't consider that a problem, as it just returns `Ok(())` on

                    // Linux.

                    Err(e) if e.kind() == io::ErrorKind::NotConnected => Ok(()),

                    Err(e) => Err(e),

                }

            });

            stream_to_daemon.join().unwrap()?;

            daemon_to_stream.join().unwrap()?;

            Ok::<(), UploadError>(())

        })

    }

    fn ls_refs(

        &self,

        repo: &fetch::StagedRepository,

        namespaces: impl IntoIterator<Item = git::PatternString>,

        remote: NodeId,

        stream: StreamId,

        channels: &mut Channels,

    ) -> Result<BTreeSet<git::Namespaced<'static>>, FetchError> {

        let tunnel = Tunnel::with(channels, stream, self.nid, remote, self.handle.clone())?;

        let tunnel_addr = tunnel.local_addr();

        let mut cmd = process::Command::new("git");

        cmd.current_dir(repo.path())

            .env_clear()

            .envs(env::vars().filter(|(k, _)| k == "PATH" || k.starts_with("GIT_TRACE")))

            .envs(git::env::GIT_DEFAULT_CONFIG)

            .args(["-c", "protocol.version=2"])

            .arg("ls-remote")

            .arg(format!("git://{tunnel_addr}/{}", repo.id.canonical()));

        for ns in namespaces.into_iter() {

            cmd.arg(ns.as_str());

        }

        cmd.stdout(process::Stdio::piped())

            .stderr(process::Stdio::piped())

            .stdin(process::Stdio::piped());

        log::debug!(target: "worker", "Running command: {:?}", cmd);

        let mut refs = BTreeSet::new();

        let mut child = cmd.spawn()?;

        let stderr = child.stderr.take().unwrap();

        let stdout = child.stdout.take().unwrap();

        // Since `ls-remote` may return a lot of data, we read the child's stdout concurrently, to

        // prevent deadlocks that could arise if we fill the pipe buffer before the process exits.

        thread::scope(|s| {

            thread::spawn_scoped(&self.nid, "ls-refs", s, || {

                for line in BufReader::new(stderr).lines().flatten() {

                    log::debug!(target: "worker", "Git: {}", line);

                }

            });

            thread::spawn_scoped(&self.nid, "ls-refs", s, || {

                for line in BufReader::new(stdout).lines().flatten() {

                    log::debug!(target: "worker", "Git: {}", line);

                    let r = match line.split_whitespace().next_back() {

                        Some(r) => r,

                        None => {

                            log::trace!(target: "worker", "Git: ls-remote returned unexpected format {line}");

                            continue;

                        }

                    };

                    match git::RefString::try_from(r) {

                        Ok(r) => {

                            if let Some(ns) = r.to_namespaced() {

                                refs.insert(ns.to_owned());

                            } else {

                                log::debug!(target: "worker", "Git: non-namespaced ref '{r}'")

                            }

                        }

                        Err(err) => {

                            log::warn!(target: "worker", "Git: invalid refname '{r}' {err}")

                        }

                    }

                }

            });

            tunnel.run(self.timeout)?;

            Ok::<_, FetchError>(())

        })?;

        let result = child.wait()?;

        if result.success() {

            Ok(refs)

        } else {

            Err(FetchError::CommandFailed {

                code: result.code().unwrap_or(1),

            })

        }

    }

    fn _fetch<S>(

        &self,

        repo: &storage::git::Repository,

        is_cloning: bool,

        remote: NodeId,

        specs: S,

        stream: StreamId,

        channels: &mut Channels,

        timeout: time::Duration,

    ) -> Result<(), FetchError>

    where

        S: IntoIterator<Item = fetch::Refspec>,

    {

        let tunnel = Tunnel::with(channels, stream, self.nid, remote, self.handle.clone())?;

        let tunnel_addr = tunnel.local_addr();

        let mut cmd = process::Command::new("git");

        cmd.current_dir(repo.path())

            .env_clear()

            .envs(env::vars().filter(|(k, _)| k == "PATH" || k.starts_with("GIT_TRACE")))

            .envs(git::env::GIT_DEFAULT_CONFIG)

            .args(["-c", "protocol.version=2"])

            .arg("fetch")

            .arg("--verbose");

        if self.atomic {

            // Enable atomic fetch. Only works with Git 2.31 and later.

            cmd.arg("--atomic");

        }

        let namespace = self.nid.to_namespace();

        let mut fetchspecs = specs

            .into_iter()

            // Filter out our own refs, if we aren't cloning.

            .filter(|fs| is_cloning || !fs.dst.starts_with(namespace.as_str()))

            .map(|spec| spec.to_string())

            .collect::<Vec<_>>();

        if !is_cloning {

            // Make sure we don't fetch our own refs via a glob pattern.

            fetchspecs.push(format!("^refs/namespaces/{}/*", self.nid));

        }

        cmd.arg(format!("git://{tunnel_addr}/{}", repo.id.canonical()))

            .args(&fetchspecs)

            .stdout(process::Stdio::piped())

            .stderr(process::Stdio::piped())

            .stdin(process::Stdio::piped());

        log::debug!(target: "worker", "Running command: {:?}", cmd);

        let mut child = cmd.spawn()?;

        let stderr = child.stderr.take().unwrap();

        thread::spawn(&self.nid, "fetch", || {

            for line in BufReader::new(stderr).lines().flatten() {

                log::debug!(target: "worker", "Git: {}", line);

            }

        });

        tunnel.run(timeout)?;

        let result = child.wait()?;

        if result.success() {

            Ok(())

        } else {

            Err(FetchError::CommandFailed {

                code: result.code().unwrap_or(1),

            })

        }

    }

    fn eof(

        remote: NodeId,

        stream: StreamId,

        sender: &mut ChannelWriter,

        handle: &mut Handle,

    ) -> Result<(), io::Error> {

        log::debug!(target: "worker", "Sending end-of-file to remote {remote}..");

        if sender.eof().is_err() {

            log::error!(target: "worker", "Fetch error: error sending end-of-file message: channel disconnected");

            return Err(io::ErrorKind::BrokenPipe.into());

        }

        if let Err(e) = handle.flush(remote, stream) {

            log::error!(target: "worker", "Error flushing worker stream: {e}");

        }

        Ok(())

    pub fn with(nid: NodeId, tasks: chan::Receiver<Task>, handle: Handle, config: Config) -> Self {

                daemon: config.daemon,

                timeout: config.timeout,

                atomic: config.atomic,

pub mod pktline {

    use std::io;

    use std::io::Read;

    use std::str;

    use super::Id;

    pub const HEADER_LEN: usize = 4;

    pub struct Reader<'a, R> {

        stream: &'a mut R,

    }

    impl<'a, R: io::Read> Reader<'a, R> {

        /// Create a new packet-line reader.

        pub fn new(stream: &'a mut R) -> Self {

            Self { stream }

        }

        /// Parse a Git request packet-line.

        ///

        /// Example: `0032git-upload-pack /project.git\0host=myserver.com\0`

        ///

        pub fn read_request_pktline(&mut self) -> io::Result<(GitRequest, Vec<u8>)> {

            let mut pktline = [0u8; 1024];

            let length = self.read_pktline(&mut pktline)?;

            let Some(cmd) = GitRequest::parse(&pktline[4..length]) else {

                return Err(io::ErrorKind::InvalidInput.into());

            };

            Ok((cmd, Vec::from(&pktline[..length])))

        }

        /// Parse a Git packet-line.

        fn read_pktline(&mut self, buf: &mut [u8]) -> io::Result<usize> {

            self.read_exact(&mut buf[..HEADER_LEN])?;

            let length = str::from_utf8(&buf[..HEADER_LEN])

                .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))?;

            let length = usize::from_str_radix(length, 16)

                .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, e.to_string()))?;

            self.read_exact(&mut buf[HEADER_LEN..length])?;

            Ok(length)

        }

    }

    impl<'a, R: io::Read> io::Read for Reader<'a, R> {

        fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {

            self.stream.read(buf)

        }

    }

    #[derive(Debug)]

    pub struct GitRequest {

        pub repo: Id,

        pub path: String,

        pub host: Option<(String, Option<u16>)>,

        pub extra: Vec<(String, Option<String>)>,

    }

    impl GitRequest {

        /// Parse a Git command from a packet-line.

        fn parse(input: &[u8]) -> Option<Self> {

            let input = str::from_utf8(input).ok()?;

            let mut parts = input

                .strip_prefix("git-upload-pack ")?

                .split_terminator('\0');

            let path = parts.next()?.to_owned();

            let repo = path.strip_prefix('/')?.parse().ok()?;

            let host = match parts.next() {

                None | Some("") => None,

                Some(host) => {

                    let host = host.strip_prefix("host=")?;

                    match host.split_once(':') {

                        None => Some((host.to_owned(), None)),

                        Some((host, port)) => {

                            let port = port.parse::<u16>().ok()?;

                            Some((host.to_owned(), Some(port)))

                        }

                    }

                }

            };

            let extra = parts

                .skip_while(|part| part.is_empty())

                .map(|part| match part.split_once('=') {

                    None => (part.to_owned(), None),

                    Some((k, v)) => (k.to_owned(), Some(v.to_owned())),

                })

                .collect();

            Some(Self {

                repo,

                path,

                host,

                extra,

            })

        }

    }

}

use std::io::Read;

    pub fn split(&mut self) -> (&mut ChannelWriter<T>, &mut ChannelReader<T>) {

        (&mut self.sender, &mut self.receiver)

    }

    pub fn pipe<W: io::Write>(&mut self, mut writer: W) -> io::Result<()> {

        loop {

            match self.receiver.recv_timeout(self.timeout) {

                Ok(ChannelEvent::Data(data)) => writer.write_all(data.as_ref())?,

                Ok(ChannelEvent::Eof) => return Ok(()),

                Ok(ChannelEvent::Close) => return Err(io::ErrorKind::ConnectionReset.into()),

                Err(chan::RecvTimeoutError::Timeout) => {

                    return Err(io::Error::new(

                        io::ErrorKind::TimedOut,

                        "error reading from stream: channel timed out",

                    ));

                }

                Err(chan::RecvTimeoutError::Disconnected) => {

                    return Err(io::Error::new(

                        io::ErrorKind::BrokenPipe,

                        "error reading from stream: channel is disconnected",

                    ));

                }

            }

        }

    }

pub struct ChannelWriter<T = Vec<u8>> {