syncfast 0.2.0

rsync/rdiff/zsync clone
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

//! This module contains the transfer protocol handlers.

pub mod fs;
pub mod locations;
pub mod ssh;
mod utils;

use log::info;
use futures::join;
use futures::sink::Sink;
use futures::stream::{LocalBoxStream, StreamExt};
use std::pin::Pin;

use crate::{Error, HashDigest};

pub enum SourceEvent {
    FileEntry(Vec<u8>, usize, HashDigest),
    EndFiles,
    FileStart(Vec<u8>),
    FileBlock(HashDigest, usize),
    FileEnd,
    BlockData(HashDigest, Vec<u8>),
}

impl std::fmt::Debug for SourceEvent {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        match self {
            &SourceEvent::FileEntry(ref path, size, ref hash) => write!(
                f,
                "FileEntry({}, {}, {})",
                String::from_utf8_lossy(&path),
                size,
                hash,
            ),
            &SourceEvent::EndFiles => write!(f, "EndFiles"),
            &SourceEvent::FileStart(ref path) => write!(
                f,
                "FileStart({})",
                String::from_utf8_lossy(&path),
            ),
            &SourceEvent::FileBlock(ref hash, size) => write!(
                f,
                "FileBlock({}, {})",
                hash,
                size,
            ),
            &SourceEvent::FileEnd => write!(f, "FileEnd"),
            &SourceEvent::BlockData(ref hash, ref data) => write!(
                f,
                "BlockData({}, <{} bytes>)",
                hash,
                data.len(),
            ),
        }
    }
}

pub enum DestinationEvent {
    GetFile(Vec<u8>),
    GetBlock(HashDigest),
    Complete,
}

impl std::fmt::Debug for DestinationEvent {
    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
        match self {
            &DestinationEvent::GetFile(ref path) => write!(
                f,
                "GetFile({})",
                String::from_utf8_lossy(&path),
            ),
            &DestinationEvent::GetBlock(ref hash) => write!(f, "GetBlock({})", hash),
            &DestinationEvent::Complete => write!(f, "Complete"),
        }
    }
}

/// The source, representing where the files are coming from.
///
/// This is relative to a single process, e.g. the sending side has a source
/// that reads from files, and the receiving side has a source that reads from
/// the network.
pub struct Source {
    stream: LocalBoxStream<'static, Result<SourceEvent, Error>>,
    sink: Pin<Box<dyn Sink<DestinationEvent, Error=Error>>>,
}

/// The destination, representing where the files are being sent.
///
/// This is relative to a single process, e.g. the sending side has a
/// destination encapsulating some network protocol, and the receiving side has
/// a destination that actually updates files.
pub struct Destination {
    stream: LocalBoxStream<'static, Result<DestinationEvent, Error>>,
    sink: Pin<Box<dyn Sink<SourceEvent, Error=Error>>>,
}

pub async fn do_sync(
    source: Source,
    destination: Destination,
) -> Result<(), Error> {
    info!("Starting sync...");
    let Source { stream: source_from, sink: source_to } = source;
    let Destination { stream: destination_from, sink: destination_to } = destination;
    info!("Streams opened");

    // Concurrently forward streams into sinks
    let (r1, r2) = join!(
        source_from.forward(destination_to),
        destination_from.forward(source_to),
    );
    r1?;
    r2?;
    info!("Sync complete");

    Ok(())
}