iota_storage/object_store/http/
s3.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5use std::{fmt, sync::Arc};
6
7use anyhow::Result;
8use async_trait::async_trait;
9use bytes::Bytes;
10use iota_config::object_storage_config::{CONNECT_TIMEOUT, TRANSFER_STALL_TIMEOUT};
11use object_store::{GetResult, path::Path};
12use percent_encoding::{PercentEncode, utf8_percent_encode};
13use reqwest::{Client, ClientBuilder};
14
15use crate::object_store::{
16    ObjectStoreGetExt, collect_get_result_with_progress,
17    http::{DEFAULT_USER_AGENT, STRICT_PATH_ENCODE_SET, exists, get, size},
18};
19
20#[derive(Debug)]
21pub(crate) struct S3Client {
22    endpoint: String,
23    client: Client,
24}
25
26impl S3Client {
27    pub fn new(endpoint: &str) -> Result<Self> {
28        let mut builder = ClientBuilder::new();
29        builder = builder
30            .user_agent(DEFAULT_USER_AGENT)
31            .pool_idle_timeout(None)
32            .connect_timeout(CONNECT_TIMEOUT)
33            .read_timeout(TRANSFER_STALL_TIMEOUT);
34        let client = builder.https_only(false).build()?;
35
36        Ok(Self {
37            endpoint: endpoint.to_string(),
38            client,
39        })
40    }
41    async fn get(&self, location: &Path) -> Result<GetResult> {
42        let url = self.path_url(location);
43        get(&url, "s3", location, &self.client).await
44    }
45    async fn exists(&self, location: &Path) -> Result<bool> {
46        let url = self.path_url(location);
47        exists(&url, "s3", location, &self.client).await
48    }
49    async fn size(&self, location: &Path) -> Result<u64> {
50        let url = self.path_url(location);
51        size(&url, "s3", location, &self.client).await
52    }
53    fn path_url(&self, path: &Path) -> String {
54        format!("{}/{}", self.endpoint, Self::encode_path(path))
55    }
56    fn encode_path(path: &Path) -> PercentEncode<'_> {
57        utf8_percent_encode(path.as_ref(), &STRICT_PATH_ENCODE_SET)
58    }
59}
60
61/// Interface for [Amazon S3](https://aws.amazon.com/s3/).
62#[derive(Debug)]
63pub struct AmazonS3 {
64    client: Arc<S3Client>,
65}
66
67impl AmazonS3 {
68    pub fn new(endpoint: &str) -> Result<Self> {
69        let s3_client = S3Client::new(endpoint)?;
70        Ok(AmazonS3 {
71            client: Arc::new(s3_client),
72        })
73    }
74}
75
76impl fmt::Display for AmazonS3 {
77    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
78        write!(f, "s3:{}", self.client.endpoint)
79    }
80}
81
82#[async_trait]
83impl ObjectStoreGetExt for AmazonS3 {
84    async fn get_bytes(&self, location: &Path) -> Result<Bytes> {
85        let result = self.client.get(location).await?;
86        let bytes = result.bytes().await?;
87        Ok(bytes)
88    }
89
90    async fn get_bytes_with_progress(
91        &self,
92        location: &Path,
93        on_bytes: &(dyn Fn(u64) + Send + Sync),
94    ) -> Result<Bytes> {
95        let result = self.client.get(location).await?;
96        collect_get_result_with_progress(result, location, on_bytes).await
97    }
98
99    async fn exists(&self, location: &Path) -> Result<bool> {
100        self.client.exists(location).await
101    }
102
103    async fn object_size(&self, location: &Path) -> Result<u64> {
104        self.client.size(location).await
105    }
106}
iota_storage/object_store/http/s3.rs

iota_storage/object_store/http/
s3.rs
