objectstore_client/
put.rs

1use std::fmt;
2use std::io::Cursor;
3use std::{borrow::Cow, collections::BTreeMap};
4
5use async_compression::tokio::bufread::ZstdEncoder;
6use bytes::Bytes;
7use futures_util::StreamExt;
8use objectstore_types::Metadata;
9use reqwest::Body;
10use serde::Deserialize;
11use tokio::io::AsyncRead;
12use tokio_util::io::{ReaderStream, StreamReader};
13
14pub use objectstore_types::{Compression, ExpirationPolicy};
15
16use crate::{ClientStream, Session};
17
18/// The response returned from the service after uploading an object.
19#[derive(Debug, Deserialize)]
20pub struct PutResponse {
21    /// The key of the object, as stored.
22    pub key: String,
23}
24
25pub(crate) enum PutBody {
26    Buffer(Bytes),
27    Stream(ClientStream),
28}
29
30impl fmt::Debug for PutBody {
31    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
32        f.debug_tuple("PutBody").finish_non_exhaustive()
33    }
34}
35
36impl Session {
37    fn put_body(&self, body: PutBody) -> PutBuilder {
38        let metadata = Metadata {
39            expiration_policy: self.scope.usecase().expiration(),
40            compression: Some(self.scope.usecase().compression()),
41            ..Default::default()
42        };
43
44        PutBuilder {
45            session: self.clone(),
46            metadata,
47            key: None,
48            body,
49        }
50    }
51
52    /// Creates a PUT request for a [`Bytes`]-like type.
53    pub fn put(&self, body: impl Into<Bytes>) -> PutBuilder {
54        self.put_body(PutBody::Buffer(body.into()))
55    }
56
57    /// Creates a PUT request with a stream.
58    pub fn put_stream(&self, body: ClientStream) -> PutBuilder {
59        self.put_body(PutBody::Stream(body))
60    }
61
62    /// Creates a PUT request with an [`AsyncRead`] type.
63    pub fn put_read<R>(&self, body: R) -> PutBuilder
64    where
65        R: AsyncRead + Send + Sync + 'static,
66    {
67        let stream = ReaderStream::new(body).boxed();
68        self.put_body(PutBody::Stream(stream))
69    }
70}
71
72/// A PUT request builder.
73#[derive(Debug)]
74pub struct PutBuilder {
75    session: Session,
76    metadata: Metadata,
77    key: Option<String>,
78    body: PutBody,
79}
80
81impl PutBuilder {
82    /// Sets an explicit object key.
83    ///
84    /// If a key is specified, the object will be stored under that key. Otherwise, the objectstore
85    /// server will automatically assign a random key, which is then returned from this request.
86    pub fn key(mut self, key: impl Into<String>) -> Self {
87        self.key = Some(key.into());
88        self
89    }
90
91    /// Sets an explicit compression algorithm to be used for this payload.
92    ///
93    /// [`None`] should be used if no compression should be performed by the client,
94    /// either because the payload is uncompressible (such as a media format), or if the user
95    /// will handle any kind of compression, without the clients knowledge.
96    pub fn compression(mut self, compression: impl Into<Option<Compression>>) -> Self {
97        self.metadata.compression = compression.into();
98        self
99    }
100
101    /// Sets the expiration policy of the object to be uploaded.
102    pub fn expiration_policy(mut self, expiration_policy: ExpirationPolicy) -> Self {
103        self.metadata.expiration_policy = expiration_policy;
104        self
105    }
106
107    /// Sets the content type of the object to be uploaded.
108    pub fn content_type(mut self, content_type: impl Into<Cow<'static, str>>) -> Self {
109        self.metadata.content_type = content_type.into();
110        self
111    }
112
113    /// This sets the custom metadata to the provided map.
114    ///
115    /// It will clear any previously set metadata.
116    pub fn set_metadata(mut self, metadata: impl Into<BTreeMap<String, String>>) -> Self {
117        self.metadata.custom = metadata.into();
118        self
119    }
120
121    /// Appends they `key`/`value` to the custom metadata of this object.
122    pub fn append_metadata(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
123        self.metadata.custom.insert(key.into(), value.into());
124        self
125    }
126}
127
128// TODO: instead of a separate `send` method, it would be nice to just implement `IntoFuture`.
129// However, `IntoFuture` needs to define the resulting future as an associated type,
130// and "impl trait in associated type position" is not yet stable :-(
131impl PutBuilder {
132    /// Sends the built PUT request to the upstream service.
133    pub async fn send(self) -> crate::Result<PutResponse> {
134        let mut builder = self.session.request(
135            reqwest::Method::PUT,
136            self.key.as_deref().unwrap_or_default(),
137        )?;
138
139        let body = match (self.metadata.compression, self.body) {
140            (Some(Compression::Zstd), PutBody::Buffer(bytes)) => {
141                let cursor = Cursor::new(bytes);
142                let encoder = ZstdEncoder::new(cursor);
143                let stream = ReaderStream::new(encoder);
144                Body::wrap_stream(stream)
145            }
146            (Some(Compression::Zstd), PutBody::Stream(stream)) => {
147                let stream = StreamReader::new(stream);
148                let encoder = ZstdEncoder::new(stream);
149                let stream = ReaderStream::new(encoder);
150                Body::wrap_stream(stream)
151            }
152            (None, PutBody::Buffer(bytes)) => bytes.into(),
153            (None, PutBody::Stream(stream)) => Body::wrap_stream(stream),
154            // _ => todo!("compression algorithms other than `zstd` are currently not supported"),
155        };
156
157        builder = builder.headers(self.metadata.to_headers("", false)?);
158
159        let response = builder.body(body).send().await?;
160        Ok(response.error_for_status()?.json().await?)
161    }
162}
objectstore_client/put.rs

objectstore_client/
put.rs
