s3sync 1.58.6

Reliable, flexible, and fast synchronization tool for S3.
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

use anyhow::{Context, Result, anyhow};
use async_channel::{Receiver, Sender};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use crate::Config;
use crate::storage::Storage;
use crate::types::token::PipelineCancellationToken;
use crate::types::{S3syncObject, SyncStatistics};

pub struct Stage {
    pub config: Config,
    pub source: Option<Storage>,
    pub target: Option<Storage>,
    pub receiver: Option<Receiver<S3syncObject>>,
    pub sender: Option<Sender<S3syncObject>>,
    pub cancellation_token: PipelineCancellationToken,
    pub has_warning: Arc<AtomicBool>,
}

#[derive(Debug, Clone, PartialEq)]
pub enum SendResult {
    Success,
    Closed,
}

impl Stage {
    pub fn new(
        config: Config,
        source: Option<Storage>,
        target: Option<Storage>,
        receiver: Option<Receiver<S3syncObject>>,
        sender: Option<Sender<S3syncObject>>,
        cancellation_token: PipelineCancellationToken,
        has_warning: Arc<AtomicBool>,
    ) -> Self {
        Self {
            config,
            source,
            target,
            receiver,
            sender,
            cancellation_token,
            has_warning,
        }
    }

    pub async fn send(&self, object: S3syncObject) -> Result<SendResult> {
        let result = self
            .sender
            .as_ref()
            .unwrap()
            .send(object)
            .await
            .context("async_channel::Sender::send() failed.");

        if let Err(e) = result {
            return if !self.is_channel_closed() {
                Err(anyhow!(e))
            } else {
                Ok(SendResult::Closed)
            };
        }

        Ok(SendResult::Success)
    }

    pub fn is_channel_closed(&self) -> bool {
        self.sender.as_ref().unwrap().is_closed()
    }

    pub async fn send_stats(&self, stats: SyncStatistics) {
        let _ = self
            .target
            .as_ref()
            .unwrap()
            .get_stats_sender()
            .send(stats)
            .await;
    }

    pub fn set_warning(&self) {
        self.has_warning.store(true, Ordering::SeqCst);
    }
}