ripsync_core/net/
sender.rs

1//! The **sender**: the side that holds the source tree.
2//!
3//! It streams its file list, then answers the receiver's [`Request`]s one at a
4//! time until the receiver sends [`Msg::Finished`]. It never initiates a read on
5//! its own — it is purely reactive — which is what keeps the single duplex stream
6//! deadlock-free.
7
8use std::io::{Read, Write};
9use std::path::Path;
10
11use crate::delta::encode_with_signature;
12use crate::filter::Filter;
13use crate::net::proto::{Data, Msg, NetEntry, NetKind, NetOptions, Request};
14use crate::net::transport::{recv_msg, send_msg};
15use crate::report::Stats;
16use crate::walk::{EntryKind, walk_controlled};
17use crate::{Error, Result, RunControl};
18
19/// Build a whole-file [`Data`] response, compressing with zstd when negotiated.
20fn whole_data(raw: Vec<u8>, options: NetOptions) -> Data {
21    if options.compress {
22        if let Ok(packed) = zstd::encode_all(raw.as_slice(), options.compress_level) {
23            return Data::Whole {
24                bytes: packed,
25                zstd: true,
26            };
27        }
28    }
29    Data::Whole {
30        bytes: raw,
31        zstd: false,
32    }
33}
34
35/// Run the sender half over `conn` for the source tree at `root`.
36///
37/// # Errors
38///
39/// Returns a walk, protocol, or I/O error, or [`Error::Cancelled`] on cancellation.
40pub fn run_sender<C: Read + Write>(
41    conn: &mut C,
42    root: &Path,
43    filter: &Filter,
44    options: NetOptions,
45    threads: usize,
46    control: &RunControl,
47) -> Result<Stats> {
48    let entries = walk_controlled(root, threads, filter, control)?;
49
50    for entry in &entries {
51        control.checkpoint()?;
52        let kind = match &entry.kind {
53            EntryKind::Dir => NetKind::Dir,
54            EntryKind::File => NetKind::File,
55            EntryKind::Symlink(target) => NetKind::Symlink(target.clone()),
56        };
57        let net = NetEntry {
58            rel: entry.rel.clone(),
59            kind,
60            len: entry.len,
61            mtime_s: entry.mtime.unix_seconds(),
62            mtime_ns: entry.mtime.nanoseconds(),
63            mode: entry.mode,
64        };
65        send_msg(conn, &Msg::Entry(net))?;
66    }
67    send_msg(conn, &Msg::ListDone)?;
68
69    // Reactive loop: one request → one response, until the receiver finishes.
70    loop {
71        control.checkpoint()?;
72        match recv_msg(conn)? {
73            Msg::Request(Request::Whole(rel)) => {
74                let data = match std::fs::read(root.join(&rel)) {
75                    Ok(bytes) => whole_data(bytes, options),
76                    Err(_) => Data::NotFound,
77                };
78                send_msg(conn, &Msg::Data(data))?;
79            }
80            Msg::Request(Request::Delta(rel, sig)) => {
81                let data = match std::fs::read(root.join(&rel)) {
82                    Ok(bytes) => Data::Delta(encode_with_signature(&sig, &bytes)),
83                    Err(_) => Data::NotFound,
84                };
85                send_msg(conn, &Msg::Data(data))?;
86            }
87            Msg::Finished(stats) => return Ok(stats),
88            Msg::Error(e) => return Err(Error::Protocol(format!("peer error: {e}"))),
89            other => {
90                return Err(Error::Protocol(format!(
91                    "sender: unexpected message {other:?}"
92                )));
93            }
94        }
95    }
96}
ripsync_core/net/sender.rs

ripsync_core/net/
sender.rs
