diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index a8c40ca92..9f61045fe 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -22,6 +22,7 @@ use crate::storage::concurrency::{ use crate::storage::head_prefix::{head_prefix_not_found_message, probe_prefix_has_children}; use crate::storage::helper::OperationHelper; use crate::storage::options::{filter_object_metadata, get_content_sha256}; +use crate::storage::readers::InMemoryAsyncReader; use crate::storage::sse::{ DecryptionRequest, EncryptionRequest, PrepareEncryptionRequest, check_encryption_metadata, sse_decryption, sse_encryption, sse_prepare_encryption, strip_managed_encryption_metadata, @@ -144,10 +145,7 @@ use std::{ sync::{Arc, LazyLock}, }; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; -use tokio::{ - io::{AsyncRead, AsyncSeek}, - sync::mpsc, -}; +use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tokio_tar::Archive; use tokio_util::io::{ReaderStream, StreamReader}; @@ -182,44 +180,6 @@ pub(crate) struct ListObjectUnorderedQuery { pub(crate) allow_unordered: Option, } -pub(crate) struct InMemoryAsyncReader { - cursor: std::io::Cursor>, -} - -impl InMemoryAsyncReader { - pub(crate) fn new(data: Vec) -> Self { - Self { - cursor: std::io::Cursor::new(data), - } - } -} - -impl AsyncRead for InMemoryAsyncReader { - fn poll_read( - mut self: std::pin::Pin<&mut Self>, - _cx: &mut std::task::Context<'_>, - buf: &mut tokio::io::ReadBuf<'_>, - ) -> std::task::Poll> { - let unfilled = buf.initialize_unfilled(); - let bytes_read = std::io::Read::read(&mut self.cursor, unfilled)?; - buf.advance(bytes_read); - std::task::Poll::Ready(Ok(())) - } -} - -impl AsyncSeek for InMemoryAsyncReader { - fn start_seek(mut self: std::pin::Pin<&mut Self>, position: std::io::SeekFrom) -> std::io::Result<()> { - // std::io::Cursor natively supports negative SeekCurrent offsets - // It will automatically handle validation and return an error if the final position would be negative - std::io::Seek::seek(&mut self.cursor, position)?; - Ok(()) - } - - fn poll_complete(self: std::pin::Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> std::task::Poll> { - std::task::Poll::Ready(Ok(self.cursor.position())) - } -} - impl FS { pub fn new() -> Self { // let store: ECStore = ECStore::new(address, endpoint_pools).await?; diff --git a/rustfs/src/storage/mod.rs b/rustfs/src/storage/mod.rs index a15cb31aa..ea161e8eb 100644 --- a/rustfs/src/storage/mod.rs +++ b/rustfs/src/storage/mod.rs @@ -21,6 +21,8 @@ mod ecfs_extend; pub(crate) mod entity; pub(crate) mod helper; pub mod options; +pub(crate) mod readers; +pub(crate) mod s3_api; pub mod tonic_service; pub(crate) use ecfs_extend::*; #[cfg(test)] diff --git a/rustfs/src/storage/readers.rs b/rustfs/src/storage/readers.rs new file mode 100644 index 000000000..0d19e7609 --- /dev/null +++ b/rustfs/src/storage/readers.rs @@ -0,0 +1,55 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use tokio::io::{AsyncRead, AsyncSeek}; + +/// Seekable in-memory async reader used by internal S3 API fast paths (e.g., GET/HEAD) +/// and by SSE flows that need a rewindable in-memory stream. +pub(crate) struct InMemoryAsyncReader { + cursor: std::io::Cursor>, +} + +impl InMemoryAsyncReader { + pub(crate) fn new(data: Vec) -> Self { + Self { + cursor: std::io::Cursor::new(data), + } + } +} + +impl AsyncRead for InMemoryAsyncReader { + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + buf: &mut tokio::io::ReadBuf<'_>, + ) -> std::task::Poll> { + let unfilled = buf.initialize_unfilled(); + let bytes_read = std::io::Read::read(&mut self.cursor, unfilled)?; + buf.advance(bytes_read); + std::task::Poll::Ready(Ok(())) + } +} + +impl AsyncSeek for InMemoryAsyncReader { + fn start_seek(mut self: std::pin::Pin<&mut Self>, position: std::io::SeekFrom) -> std::io::Result<()> { + // std::io::Cursor natively supports negative SeekCurrent offsets + // It will automatically handle validation and return an error if the final position would be negative + std::io::Seek::seek(&mut self.cursor, position)?; + Ok(()) + } + + fn poll_complete(self: std::pin::Pin<&mut Self>, _cx: &mut std::task::Context<'_>) -> std::task::Poll> { + std::task::Poll::Ready(Ok(self.cursor.position())) + } +} diff --git a/rustfs/src/storage/s3_api/mod.rs b/rustfs/src/storage/s3_api/mod.rs new file mode 100644 index 000000000..35921e90c --- /dev/null +++ b/rustfs/src/storage/s3_api/mod.rs @@ -0,0 +1,34 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Facade modules for incremental S3 API extraction from `ecfs.rs`. +//! +//! This file intentionally starts as skeleton-only. Behavior remains in place +//! until each helper is moved with dedicated small refactor steps. + +pub(crate) mod acl {} +pub(crate) mod bucket {} +pub(crate) mod encryption {} +pub(crate) mod multipart {} +/// Object helper facade placeholder. +/// +/// Read-path helpers shared across storage components should live in neutral +/// modules (for example, `storage::readers`) and be consumed from there. +/// Object-specific extraction steps can be added here incrementally. +pub(crate) mod object {} +pub(crate) mod replication {} +pub(crate) mod response {} +pub(crate) mod restore {} +pub(crate) mod select {} +pub(crate) mod validation {} diff --git a/rustfs/src/storage/sse.rs b/rustfs/src/storage/sse.rs index 6e3e0bdd9..bcac64c38 100644 --- a/rustfs/src/storage/sse.rs +++ b/rustfs/src/storage/sse.rs @@ -94,7 +94,7 @@ use tokio::io::AsyncRead; use tracing::{debug, error}; use crate::error::ApiError; -use crate::storage::ecfs::InMemoryAsyncReader; +use crate::storage::readers::InMemoryAsyncReader; use rustfs_ecstore::bucket::metadata_sys; use rustfs_ecstore::error::Error; use s3s::dto::{SSECustomerAlgorithm, SSECustomerKey, SSECustomerKeyMD5, SSEKMSKeyId};