Struct s3_sync::Bucket

pub struct Bucket { /* private fields */ }

Represents S3 bucket.

Implementations

impl Bucket

pub fn from_name(name: impl Into<String>) -> Bucket

Creates Bucket from bucket name.

pub fn name(&self) -> &str

Gets bucket name.

Examples found in repository

src/lib.rs (line 200)

    pub fn bucket_name(&self) -> &str {
        self.bucket.name()
    }

    /// Gets key.
    pub fn key(&self) -> &str {
        &self.key
    }

    /// Gets present bucket.
    pub fn bucket(&self) -> &Present<Bucket> {
        self.bucket
    }

    /// Makes an assumption that the object does exist on S3 without making an API call to verify.
    pub fn assume_present(self) -> Present<BucketKey<'b>> {
        Present(self)
    }
}

impl<'b> From<Object<'b>> for BucketKey<'b> {
    fn from(obj: Object<'b>) -> BucketKey<'b> {
        obj.bucket_key.0
    }
}

impl<'b> From<Object<'b>> for Present<BucketKey<'b>> {
    fn from(obj: Object<'b>) -> Present<BucketKey<'b>> {
        obj.bucket_key
    }
}

impl<'b> Borrow<Present<BucketKey<'b>>> for Object<'b> {
    fn borrow(&self) -> &Present<BucketKey<'b>> {
        self.bucket_key()
    }
}

impl<'b> From<Present<BucketKey<'b>>> for BucketKey<'b> {
    fn from(obj: Present<BucketKey<'b>>) -> BucketKey<'b> {
        obj.0
    }
}

impl<'b> From<Absent<BucketKey<'b>>> for BucketKey<'b> {
    fn from(obj: Absent<BucketKey<'b>>) -> BucketKey<'b> {
        obj.0
    }
}

/// Represents an existing object and its metadata in S3.
#[derive(Debug, PartialEq, Eq)]
pub struct Object<'b> {
    pub bucket_key: Present<BucketKey<'b>>,
    pub size: i64,
    pub e_tag: String,
    pub last_modified: LastModified,
}

impl<'b> Object<'b> {
    fn from_head(
        bucket_key: BucketKey<'b>,
        res: HeadObjectOutput,
    ) -> Result<Object, S3SyncError> {
        Ok(Object {
            bucket_key: Present(bucket_key),
            e_tag: res.e_tag.ok_or(S3SyncError::MissingObjectMetaData("e_tag"))?,
            last_modified: res.last_modified.map(LastModified::Rfc2822)
                .ok_or(S3SyncError::MissingObjectMetaData("last_modified"))?,
            size: res.content_length.ok_or(S3SyncError::MissingObjectMetaData("content_length"))?,
        })
    }

    fn from_s3_object(bucket: &Present<Bucket>, object: S3Object) -> Result<Object, S3SyncError> {
        Ok(Object {
            bucket_key: Present(BucketKey::from_key(bucket, object.key
                .ok_or(S3SyncError::MissingObjectMetaData("key"))?)),
            e_tag: object.e_tag.ok_or(S3SyncError::MissingObjectMetaData("e_tag"))?,
            last_modified: object.last_modified.map(LastModified::Rfc3339)
                .ok_or(S3SyncError::MissingObjectMetaData("last_modified"))?,
            size: object.size.ok_or(S3SyncError::MissingObjectMetaData("size"))?,
        })
    }

    /// Gets [BucketKey] pointing to this object.
    pub fn bucket_key(&self) -> &Present<BucketKey<'b>> {
        &self.bucket_key
    }

    /// Unwraps inner [BucketKey] pointing to this object.
    pub fn unwrap_bucket_key(self) -> Present<BucketKey<'b>> {
        self.bucket_key
    }

    /// Gets object size.
    pub fn size(&self) -> i64 {
        self.size
    }

    /// Gets object ETag.
    pub fn e_tag(&self) -> &str {
        &self.e_tag
    }

    /// Gets object last modified time.
    pub fn last_modified(&self) -> &LastModified {
        &self.last_modified
    }
}

/// Existing pair of bucket name and key strings.
pub trait PresentBucketKeyName {
    /// Gets bucket name.
    fn bucket_name(&self) -> &str;
    /// Gets key.
    fn key(&self) -> &str;
}

impl<'b> PresentBucketKeyName for Present<BucketKey<'b>> {
    fn bucket_name(&self) -> &str {
        self.bucket.name()
    }

    fn key(&self) -> &str {
        &self.key
    }
}

impl<'b> PresentBucketKeyName for Object<'b> {
    fn bucket_name(&self) -> &str {
        self.bucket_key.bucket_name()
    }

    fn key(&self) -> &str {
        self.bucket_key.key()
    }
}

/// Represents last modified time and date.
///
/// Depending on the source of the information it will be in different format.
#[derive(Debug, PartialEq, Eq)]
pub enum LastModified {
    /// Date and time in RFC2822 format.
    Rfc2822(String),
    /// Date and time in RFC3339 format.
    Rfc3339(String),
}

impl fmt::Display for LastModified {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.as_str())
    }
}

impl LastModified {
    /// Gets string representation that may be in different format.
    pub fn as_str(&self) -> &str {
        match self {
            LastModified::Rfc2822(dt) => &dt,
            LastModified::Rfc3339(dt) => &dt,
        }
    }

    #[cfg(feature = "chrono")]
    /// Returns parsed date and time.
    pub fn parse(&self) -> Result<DateTime<FixedOffset>, S3SyncError> {
        Ok(match &self {
            LastModified::Rfc2822(lm) => DateTime::parse_from_rfc2822(lm),
            LastModified::Rfc3339(lm) => DateTime::parse_from_rfc3339(lm),
        }?)
    }
}

struct PaginationIter<RQ, RS, SSA, GSA, FF, E>
where RQ: Clone, SSA: Fn(&mut RQ, String), GSA: Fn(&RS) -> Option<String>, FF: Fn(RQ) -> Result<RS, E> {
    request: RQ,
    // function that returns request parametrised to fetch next page
    set_start_after: SSA,
    get_start_after: GSA,
    fetch: FF,
    done: bool
}

impl<RQ, RS, SSA, GSA, FF, E> Iterator for PaginationIter<RQ, RS, SSA, GSA, FF, E>
where RQ: Clone, SSA: Fn(&mut RQ, String), GSA: Fn(&RS) -> Option<String>, FF: Fn(RQ) -> Result<RS, E> {
    type Item = Result<RS, E>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.done {
            return None
        }
        Some((self.fetch)(self.request.clone()).map(|response| {
            if let Some(start_after) = (self.get_start_after)(&response) {
                (self.set_start_after)(&mut self.request, start_after);
            } else {
                self.done = true;
            }
            response
        }))
    }
}

/// Information about status of the ongoing transfer.
#[derive(Debug)]
pub enum TransferStatus {
    /// Initialization successful.
    Init,
    /// Transfer is ongoing.
    Progress(TransferStats),
    /// Transfer successfully complete.
    Done(TransferStats),
    /// There was an error.
    Failed(String),
}

impl Default for TransferStatus {
    fn default() -> Self {
        TransferStatus::Init
    }
}

impl TransferStatus {
    fn update(&mut self, stats: TransferStats) {
        match self {
            TransferStatus::Init => {
                *self = TransferStatus::Progress(stats);
            }
            TransferStatus::Progress(ref mut s) => {
                s.buffers += stats.buffers;
                s.bytes = stats.bytes;
                s.bytes_total += stats.bytes_total;
            },
            _ => panic!("TransferStats in bad state for .done(): {:?}", self),
        }
    }

    fn done(self) -> Self {
        match self {
            TransferStatus::Progress(stats) => TransferStatus::Done(stats),
            _ => panic!("TransferStats in bad state for .done(): {:?}", self),
        }
    }

    fn failed(self, err: String) -> Self {
        match self {
            TransferStatus::Init |
            TransferStatus::Progress(_) => TransferStatus::Failed(err),
            _ => panic!("TransferStats in bad state for .failed(): {:?}", self),
        }
    }
}

/// Information about transfer progress.
#[derive(Debug, Default)]
pub struct TransferStats {
    /// Number of buffers or parts transferred.
    pub buffers: u16,
    /// Number of bytes transferred since last progress.
    pub bytes: u64,
    /// Total number of bytes transferred.
    pub bytes_total: u64,
}

/// Meta information about object body.
#[derive(Debug)]
pub struct ObjectBodyMeta {
    /// A standard MIME type describing the format of the object data.
    pub content_type: String,
    /// Specifies presentational information for the object.
    pub content_disposition: Option<String>,
    /// The language the content is in.
    pub content_language: Option<String>,
}

impl Default for ObjectBodyMeta {
    fn default() -> ObjectBodyMeta {
        ObjectBodyMeta {
            content_type: "application/octet-stream".to_owned(),
            content_disposition: None,
            content_language: None,
        }
    }
}

/// Method used for checking if given object exists
#[derive(Debug)]
pub enum CheckObjectImpl {
    /// Required `GetObject` AWS permission
    Head,
    /// Required `ListBucket` AWS permission
    List,
}

#[derive(Debug)]
pub struct Settings {
    /// Size of multipart upload part.
    ///
    /// Note: On AWS S3 the part size is must be between 5MiB to 5GiB
    pub part_size: usize,
    /// Timeout for non data related operations.
    pub timeout: Duration,
    /// Timeout for data upload/download operations.
    pub data_timeout: Duration,
    /// Maximum number of multipart uploads (for calculations of [S3::max_upload_size()]) (AWS limit is 10k)
    pub max_multipart_upload_parts: usize,
    /// Maximum number of objects that can be deleted with one API call (AWS limit is 1k)
    pub max_delete_objects: usize,
}

impl Default for Settings {
    fn default() -> Settings {
        Settings {
            part_size: 10 * 1024 * 1024, // note that max part count is 10k so we can upload up to 100_000 MiB
            timeout: Duration::from_secs(10),
            data_timeout: Duration::from_secs(300),
            max_multipart_upload_parts: 10_000, // ASW S3 limit
            max_delete_objects: 1_000, // ASW S3 limit
        }
    }
}

/// Wrapper of Rusoto S3 client that adds some high level imperative and declarative operations on
/// S3 buckets and objects.
pub struct S3 {
    client: S3Client,
    on_upload_progress: Option<RefCell<Box<dyn FnMut(&TransferStatus)>>>,
    part_size: usize,
    timeout: Duration,
    data_timeout: Duration,
    max_multipart_upload_parts: usize,
    max_delete_objects: usize,
}

impl fmt::Debug for S3 {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("S3")
         .field("part_size", &self.part_size)
         .field("timeout", &self.timeout)
         .field("data_timeout", &self.data_timeout)
         .finish()
    }
}

impl Default for S3 {
    fn default() -> S3 {
        S3::new(None, None, None)
    }
}

impl S3 {
    /// Creates high level S3 client.
    ///
    /// * `region` - the AWS region to connect to; when `None` autodetects the region value (see [Region] for detail)
    /// * `region_endpoint` - use dedicated AWS endpoint within the region
    /// * `settings` - use specific client setting
    pub fn new(
        region: impl Into<Option<Region>>,
        region_endpoint: impl Into<Option<String>>,
        settings: impl Into<Option<Settings>>
    ) -> S3 {
        let region = match (region.into(), region_endpoint.into()) {
            (Some(region), Some(endpoint)) => Region::Custom { name: region.name().to_owned(), endpoint },
            (None, Some(endpoint)) => Region::Custom { name: Region::default().name().to_owned(), endpoint },
            (Some(region), None) => region,
            _ => Region::default(),
        };
        let settings = settings.into().unwrap_or_default();

        S3 {
            client: S3Client::new(region),
            on_upload_progress: None,
            part_size: settings.part_size,
            timeout: settings.timeout,
            data_timeout: settings.data_timeout,
            max_multipart_upload_parts: settings.max_multipart_upload_parts,
            max_delete_objects: settings.max_delete_objects,
        }
    }

    /// Creates high level S3 client with given region and default settings.
    pub fn with_region(region: Region) -> S3 {
        S3::new(region, None, None)
    }

    /// Gets maximum size of the multipart upload part.
    ///
    /// Useful to set up other I/O buffers accordingly.
    pub fn part_size(&self) -> usize {
        self.part_size
    }

    /// Returns maximum size of data in bytes that can be uploaded to a single object with current settings.
    pub fn max_upload_size(&self) -> usize {
        self.part_size * self.max_multipart_upload_parts
    }

    /// Set callback on body upload progress.
    pub fn on_upload_progress(
        &mut self,
        callback: impl FnMut(&TransferStatus) + 'static
    ) -> Option<Box<dyn FnMut(&TransferStatus)>> {
        let ret = self.on_upload_progress.take();
        self.on_upload_progress = Some(RefCell::new(Box::new(callback)));
        ret.map(|c| c.into_inner())
    }

    /// Calls `f` with [S3] client that has [S3::on_upload_progress()] set to `callback` and restores
    /// callback to previous state on return.
    pub fn with_on_upload_progress<O>(
        &mut self,
        callback: impl FnMut(&TransferStatus) + 'static,
        f: impl FnOnce(&mut Self) -> O
    ) -> O {
        let old = self.on_upload_progress(callback);
        let ret = f(self);
        old.map(|callback| self.on_upload_progress(callback));
        ret
    }

    fn notify_upload_progress(&self, status: &TransferStatus) {
        self.on_upload_progress.as_ref().map(|c| {
            c.try_borrow_mut().expect("S3 upload_progress closure already borrowed mutable").as_mut()(status);
        });
    }

    /// Checks if given bucket exists.
    pub fn check_bucket_exists(&self, bucket: Bucket) -> Result<Either<Present<Bucket>, Absent<Bucket>>, S3SyncError> {
        let res = self.client.head_bucket(HeadBucketRequest {
            bucket: bucket.name.clone(),
            .. Default::default()
        }).with_timeout(self.timeout).sync();
        trace!("Head bucket response: {:?}", res);

        match res {
            Ok(_) => Ok(Left(Present(bucket))),
            Err(RusotoError::Service(HeadBucketError::NoSuchBucket(_))) => Ok(Right(Absent(bucket))),
            Err(RusotoError::Unknown(BufferedHttpResponse { status, .. })) if status.as_u16() == 404 => Ok(Right(Absent(bucket))),
            Err(err) => Err(err.into())
        }
    }

    /// Checks if given object exists.
    ///
    /// * `implementation` - select implementation of this function
    ///
    /// Note:
    ///
    /// * [Object::last_modified()] value will be in different format depending on implementation.
    pub fn check_object_exists<'s, 'b>(&'s self, bucket_key: BucketKey<'b>, implementation: CheckObjectImpl)
        -> Result<Either<Object<'b>, Absent<BucketKey<'b>>>, S3SyncError> {
        match implementation {
            CheckObjectImpl::List => self.check_object_exists_list(bucket_key),
            CheckObjectImpl::Head => self.check_object_exists_head(bucket_key),
        }
    }

    /// Checks if given object exists by issuing `HeadObject` API request.
    ///
    /// Note:
    ///
    /// * Requires `GetObject` AWS permission.
    /// * The [Object::last_modified()] value will be in RFC 2822 format.
    pub fn check_object_exists_head<'s, 'b>(&'s self, bucket_key: BucketKey<'b>)
        -> Result<Either<Object<'b>, Absent<BucketKey<'b>>>, S3SyncError> {
        let res = self.client.head_object(HeadObjectRequest {
            bucket: bucket_key.bucket.name.clone(),
            key: bucket_key.key.clone(),
            .. Default::default()
        }).with_timeout(self.timeout).sync();
        trace!("Head response: {:?}", res);

        match res {
            Ok(res) => Ok(Left(Object::from_head(bucket_key, res)?)),
            Err(RusotoError::Service(HeadObjectError::NoSuchKey(_))) =>
                Ok(Right(Absent(bucket_key))),
            Err(RusotoError::Unknown(BufferedHttpResponse { status, .. })) if status.as_u16() == 404 =>
                Ok(Right(Absent(bucket_key))),
            Err(err) => Err(err.into())
        }
    }

    /// Checks if given object exists by listing objects with `ListObjcetsV2` API request.
    ///
    /// Note:
    ///
    /// * Requires `ListBucket` AWS permission.
    /// * The [Object::last_modified()] value will be in RFC 3339 format.
    pub fn check_object_exists_list<'s, 'b>(&'s self, bucket_key: BucketKey<'b>)
        -> Result<Either<Object<'b>, Absent<BucketKey<'b>>>, S3SyncError> {
        let request = ListObjectsV2Request {
            bucket: bucket_key.bucket.name().to_owned(),
            prefix: Some(bucket_key.key.clone()),
            max_keys: Some(1),
            .. Default::default()
        };

        let res = self.client.list_objects_v2(request).with_timeout(self.timeout).sync()?;
        let first_obj = res.contents.
            and_then(|list| list.into_iter().next());
        match first_obj {
            Some(obj) if obj.key.as_deref().expect("S3 object has no key!") == bucket_key.key =>
                Ok(Left(Object::from_s3_object(bucket_key.bucket, obj)?)),
            _ => Ok(Right(Absent(bucket_key)))
        }
    }

    /// Provides iterator of objects in existing bucket that have key of given prefix.
    ///
    /// Note:
    ///
    /// * Requires `ListBucket` AWS permission.
    /// * The [Object::last_modified()] value will be in RFC 3339 format.
    pub fn list_objects<'b, 's: 'b>(&'s self, bucket: &'b Present<Bucket>, prefix: String)
        -> impl Iterator<Item = Result<Object<'b>, S3SyncError>> + Captures1<'s> + Captures2<'b> {
        let client = &self.client;
        let pages = PaginationIter {
            request: ListObjectsV2Request {
                bucket: bucket.name().to_owned(),
                prefix: Some(prefix),
                .. Default::default()
            },
            set_start_after: |request: &mut ListObjectsV2Request, start_after| {
                request.start_after = Some(start_after);
            },
            get_start_after: |response: &ListObjectsV2Output| {
                response.contents.as_ref()
                    .and_then(|objects| objects.last().and_then(|last| last.key.as_ref().map(|r| r.clone())))
            },
            fetch: move |request: ListObjectsV2Request| {
                client.list_objects_v2(request).with_timeout(self.timeout).sync().map_err(Into::into)
            },
            done: false
        };

        pages.flat_map(move |response| {
            let mut error = None;
            let mut objects = None;
            match response {
                Err(err) => error = Some(err),
                Ok(output) => objects = output.contents.map(|objects| objects.into_iter()),
            }

            unfold((), move |_| {
                if let Some(error) = error.take() {
                    Some(Err(error))
                } else {
                    objects.as_mut().and_then(|obj| obj.next()).map(|o| Ok(Object::from_s3_object(bucket, o)?))
                }
            })
        })
    }

Trait Implementations

impl Clone for Bucket

fn clone(&self) -> Bucket

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Bucket

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Ord for Bucket

fn cmp(&self, other: &Bucket) -> Ordering

This method returns an Ordering between self and other.

fn max(self, other: Self) -> Selfwhere
    Self: Sized,

Compares and returns the maximum of two values.

fn min(self, other: Self) -> Selfwhere
    Self: Sized,

Compares and returns the minimum of two values.

fn clamp(self, min: Self, max: Self) -> Selfwhere
    Self: Sized + PartialOrd<Self>,

Restrict a value to a certain interval.

impl PartialEq<Bucket> for Bucket

fn eq(&self, other: &Bucket) -> bool

This method tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

This method tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl PartialOrd<Bucket> for Bucket

fn partial_cmp(&self, other: &Bucket) -> Option<Ordering>

This method returns an ordering between self and other values if one exists.

fn lt(&self, other: &Rhs) -> bool

This method tests less than (for self and other) and is used by the < operator.

fn le(&self, other: &Rhs) -> bool

This method tests less than or equal to (for self and other) and is used by the <= operator.

fn gt(&self, other: &Rhs) -> bool

This method tests greater than (for self and other) and is used by the > operator.

fn ge(&self, other: &Rhs) -> bool

This method tests greater than or equal to (for self and other) and is used by the >= operator.

impl Eq for Bucket

impl External for Bucket

impl StructuralEq for Bucket

impl StructuralPartialEq for Bucket