aws_multipart_upload/
uri.rs

1//! Provides `ObjectUri` functionality.
2//!
3//! The only thing required to create a new upload is the URI, or bucket and
4//! key, of the object to be uploaded. An iterator of `ObjectUri` therefore
5//! creates a sequence of multipart uploads by using the `next` value to create
6//! a next upload when one completes.
7//!
8//! [`ObjectUriIter`] can be given such an iterator and then be handed to an
9//! upload, which will use it to create, build, complete, then create a next
10//! upload. This will continue indefinitely, or as long as the iterator can
11//! produce the `next` value.  This allows an upload to work in a stream where
12//! the aim is to consume the stream by writing it to S3 for perpetuity.
13//!
14//! In case there should be only one upload, [`OneTimeUse`] only returns the
15//! value it was created with.
16//!
17//! # Example
18//!
19//! This is an iterator of `ObjectUri`s that writes to a prefix based on the
20//! current date and time.
21//!
22//! ```rust
23//! use aws_multipart_upload::uri::{KeyPrefix, ObjectUriIterExt};
24//! use aws_multipart_upload::ObjectUriIter;
25//!
26//! const BUCKET: &str = "my-bucket";
27//! const PREFIX: &str = "static/object/prefix";
28//!
29//! let iter_pfx = std::iter::repeat_with(|| KeyPrefix::from(PREFIX));
30//! let iter = iter_pfx.map_key(BUCKET, |prefix| {
31//!     let now = chrono::Utc::now();
32//!     let now_str = now.format("%Y/%m/%d/%H").to_string();
33//!     let us = now.timestamp_micros();
34//!     let root = format!("{now_str}/{us}.csv");
35//!     prefix.to_key(&root)
36//! });
37//!
38//! let mut uri = ObjectUriIter::new(iter);
39//! let new_uri = uri.next().unwrap();
40//!
41//! println!("{new_uri}");
42//! // "s3://my-bucket/static/object/prefix/2025/11/11/11/01/1763683634194850.csv"
43//! ```
44//! [`ObjectUriIter`]: super::ObjectUriIter
45use std::borrow::Cow;
46use std::fmt::{self, Formatter};
47use std::ops::Deref;
48use std::sync::{Arc, Mutex};
49
50use crate::client::UploadClient;
51use crate::client::request::{CreateRequest, SendCreateUpload};
52
53/// The destination bucket for this upload when it is complete.
54#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
55pub struct Bucket(Cow<'static, str>);
56
57impl Bucket {
58    /// Create a new `Bucket`.
59    pub fn new<T: Into<Cow<'static, str>>>(bucket: T) -> Self {
60        let bucket: Cow<'static, str> = bucket.into();
61        match bucket.strip_suffix("/") {
62            Some(v) => v.into(),
63            _ => Self(bucket),
64        }
65    }
66
67    pub(crate) fn is_empty(&self) -> bool {
68        self.0.is_empty()
69    }
70}
71
72impl Deref for Bucket {
73    type Target = str;
74
75    fn deref(&self) -> &str {
76        &self.0
77    }
78}
79
80impl AsRef<str> for Bucket {
81    fn as_ref(&self) -> &str {
82        self
83    }
84}
85
86impl fmt::Display for Bucket {
87    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
88        self.0.fmt(f)
89    }
90}
91
92impl From<&str> for Bucket {
93    fn from(value: &str) -> Self {
94        Self::new(value.to_string())
95    }
96}
97
98impl From<String> for Bucket {
99    fn from(value: String) -> Self {
100        Self(Cow::Owned(value))
101    }
102}
103
104impl From<Cow<'static, str>> for Bucket {
105    fn from(value: Cow<'static, str>) -> Self {
106        Self(value)
107    }
108}
109
110/// The key within the associated bucket for this object.
111#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
112pub struct Key(Cow<'static, str>);
113
114impl Key {
115    /// Create a new object `Key`.
116    pub fn new<T: Into<Cow<'static, str>>>(key: T) -> Self {
117        Self(key.into())
118    }
119
120    pub(crate) fn is_empty(&self) -> bool {
121        self.0.is_empty()
122    }
123}
124
125impl Deref for Key {
126    type Target = str;
127
128    fn deref(&self) -> &str {
129        &self.0
130    }
131}
132
133impl AsRef<str> for Key {
134    fn as_ref(&self) -> &str {
135        self
136    }
137}
138
139impl fmt::Display for Key {
140    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
141        self.0.fmt(f)
142    }
143}
144
145impl From<&str> for Key {
146    fn from(value: &str) -> Self {
147        Self::new(value.to_string())
148    }
149}
150
151impl From<String> for Key {
152    fn from(value: String) -> Self {
153        Self(Cow::Owned(value))
154    }
155}
156
157impl From<Cow<'static, str>> for Key {
158    fn from(value: Cow<'static, str>) -> Self {
159        Self(value)
160    }
161}
162
163/// A prefix of S3 object keys.
164#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
165pub struct KeyPrefix(Cow<'static, str>);
166
167impl KeyPrefix {
168    /// Create a new object key prefix.
169    ///
170    /// Normalized to end with a single `'/'` and have no leading `'/'`.
171    pub fn new<T: Into<Cow<'static, str>>>(prefix: T) -> Self {
172        let raw: Cow<'static, str> = prefix.into();
173        let trimmed = raw.trim_matches('/');
174        Self(format!("{trimmed}/").into())
175    }
176
177    /// A root prefix `"/"`.
178    pub fn root() -> Self {
179        KeyPrefix("/".into())
180    }
181
182    /// Extend this prefix by another.
183    pub fn append(&self, other: &KeyPrefix) -> Self {
184        format!("{self}{other}").into()
185    }
186
187    /// Create an object [`Key`] with this prefix and the given suffix.
188    pub fn to_key(&self, suffix: &str) -> Key {
189        format!("{self}{suffix}").into()
190    }
191}
192
193impl Deref for KeyPrefix {
194    type Target = str;
195
196    fn deref(&self) -> &str {
197        &self.0
198    }
199}
200
201impl AsRef<str> for KeyPrefix {
202    fn as_ref(&self) -> &str {
203        self
204    }
205}
206
207impl fmt::Display for KeyPrefix {
208    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
209        self.0.fmt(f)
210    }
211}
212
213impl From<&str> for KeyPrefix {
214    fn from(value: &str) -> Self {
215        Self::new(value.to_string())
216    }
217}
218
219impl From<String> for KeyPrefix {
220    fn from(value: String) -> Self {
221        Self::new(value)
222    }
223}
224
225impl From<Cow<'static, str>> for KeyPrefix {
226    fn from(value: Cow<'static, str>) -> Self {
227        Self(value)
228    }
229}
230
231/// The address of an uploaded object in S3.
232#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
233pub struct ObjectUri {
234    /// The S3 bucket for the object.
235    ///
236    /// This should be the plain bucket name, e.g., "my-s3-bucket".
237    pub bucket: Bucket,
238    /// The full key of this object within the bucket.
239    pub key: Key,
240}
241
242impl ObjectUri {
243    /// Create a new `ObjectUri` from bucket and object key.
244    pub fn new<B: Into<Bucket>, K: Into<Key>>(bucket: B, key: K) -> Self {
245        Self { bucket: bucket.into(), key: key.into() }
246    }
247
248    /// Returns a reference to the bucket.
249    pub fn bucket(&self) -> &Bucket {
250        &self.bucket
251    }
252
253    /// Returns a reference to the object key.
254    pub fn key(&self) -> &Key {
255        &self.key
256    }
257
258    pub(crate) fn is_empty(&self) -> bool {
259        self.bucket.is_empty() || self.key.is_empty()
260    }
261}
262
263impl fmt::Display for ObjectUri {
264    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
265        write!(f, "s3://{}/{}", &self.bucket, &self.key)
266    }
267}
268
269impl<T: Into<Bucket>, U: Into<Key>> From<(T, U)> for ObjectUri {
270    fn from((b, k): (T, U)) -> Self {
271        ObjectUri::new(b.into(), k.into())
272    }
273}
274
275/// Produce an `ObjectUri` for a new upload from an iterator.
276#[derive(Clone)]
277pub struct ObjectUriIter {
278    inner: Arc<Mutex<ObjectUriIterInner>>,
279}
280
281impl ObjectUriIter {
282    /// Create a new `ObjectUriIter` from an arbitrary iterator of `ObjectUri`.
283    pub fn new<I>(iter: I) -> Self
284    where
285        I: Iterator<Item = ObjectUri> + Send + Sync + 'static,
286    {
287        let inner = ObjectUriIterInner(Box::new(iter));
288        Self { inner: Arc::new(Mutex::new(inner)) }
289    }
290
291    /// Construct the request future to create a new multipart upload using the
292    /// next `ObjectUri` produced by this `ObjectUriIter` value.
293    pub fn next_upload(
294        &mut self,
295        client: &UploadClient,
296    ) -> Option<SendCreateUpload> {
297        let uri = self.next()?;
298        let req = CreateRequest::new(uri);
299        let fut = SendCreateUpload::new(client, req);
300        Some(fut)
301    }
302}
303
304impl Default for ObjectUriIter {
305    fn default() -> Self {
306        Self::new(EmptyUri.into_iter())
307    }
308}
309
310impl Iterator for ObjectUriIter {
311    type Item = ObjectUri;
312
313    fn next(&mut self) -> Option<Self::Item> {
314        let mut inner = self.inner.lock().unwrap();
315        inner.next()
316    }
317}
318
319impl fmt::Debug for ObjectUriIter {
320    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
321        f.debug_struct("ObjectUriIter").finish()
322    }
323}
324
325/// Adds the method `map_key` to iterators over `KeyPrefix`.
326pub trait ObjectUriIterExt: Iterator {
327    /// Returns an iterator of `ObjectUri` by applying the function `F` to each
328    /// `KeyPrefix` to produce the object `Key`.
329    fn map_key<B, F>(self, bucket: B, f: F) -> MapKey<Self, F>
330    where
331        Self: Iterator<Item = KeyPrefix> + Sized,
332        F: FnMut(KeyPrefix) -> Key,
333        B: Into<Bucket>,
334    {
335        MapKey::new(self, bucket, f)
336    }
337}
338
339impl<I: Iterator> ObjectUriIterExt for I {}
340
341/// Iterator for [`map_key`](ObjectUriIterExt::map_key).
342pub struct MapKey<I, F> {
343    bucket: Bucket,
344    inner: I,
345    f: F,
346}
347
348impl<I, F> MapKey<I, F> {
349    fn new<B: Into<Bucket>>(inner: I, bucket: B, f: F) -> Self {
350        Self { inner, bucket: bucket.into(), f }
351    }
352}
353
354impl<I, F> Iterator for MapKey<I, F>
355where
356    I: Iterator<Item = KeyPrefix>,
357    F: FnMut(KeyPrefix) -> Key,
358{
359    type Item = ObjectUri;
360
361    fn next(&mut self) -> Option<Self::Item> {
362        let prefix = self.inner.next()?;
363        let key = (self.f)(prefix);
364        let uri = ObjectUri::new(self.bucket.clone(), key);
365        Some(uri)
366    }
367}
368
369/// An empty iterator of `ObjectUri`s.
370#[derive(Debug, Clone, Copy, Default)]
371pub struct EmptyUri;
372
373impl IntoIterator for EmptyUri {
374    type IntoIter = std::iter::Empty<ObjectUri>;
375    type Item = ObjectUri;
376
377    fn into_iter(self) -> Self::IntoIter {
378        std::iter::empty()
379    }
380}
381
382/// Iterator that is exhausted after one `ObjectUri`.
383#[derive(Debug, Clone, Default)]
384pub struct OneTimeUse(Option<ObjectUri>);
385
386impl OneTimeUse {
387    /// Use the given `uri` as the one produced.
388    pub fn new(uri: ObjectUri) -> Self {
389        Self(Some(uri))
390    }
391}
392
393impl Iterator for OneTimeUse {
394    type Item = ObjectUri;
395
396    fn next(&mut self) -> Option<Self::Item> {
397        self.0.take()
398    }
399}
400
401struct ObjectUriIterInner(Box<dyn Iterator<Item = ObjectUri> + Send + Sync>);
402
403impl ObjectUriIterInner {
404    fn next(&mut self) -> Option<ObjectUri> {
405        self.0.next()
406    }
407}
aws_multipart_upload/uri.rs

aws_multipart_upload/
uri.rs
