diff --git a/TODO.md b/TODO.md index 15472d274..cbcab874a 100644 --- a/TODO.md +++ b/TODO.md @@ -13,6 +13,7 @@ - [x] 远程rpc - [x] 错误类型判断,程序中判断错误类型，如何统一错误 - [x] 优化xlmeta, 自定义msg数据结构 +- [x] appendFile, createFile, readFile, walk_dir sync io ## 基础功能 diff --git a/common/protos/src/generated/flatbuffers_generated/models.rs b/common/protos/src/generated/flatbuffers_generated/models.rs index e4949fdcf..aa1f6ae2f 100644 --- a/common/protos/src/generated/flatbuffers_generated/models.rs +++ b/common/protos/src/generated/flatbuffers_generated/models.rs @@ -1,9 +1,10 @@ // automatically generated by the FlatBuffers compiler, do not modify + // @generated -use core::cmp::Ordering; use core::mem; +use core::cmp::Ordering; extern crate flatbuffers; use self::flatbuffers::{EndianScalar, Follow}; @@ -11,114 +12,112 @@ use self::flatbuffers::{EndianScalar, Follow}; #[allow(unused_imports, dead_code)] pub mod models { - use core::cmp::Ordering; - use core::mem; + use core::mem; + use core::cmp::Ordering; - extern crate flatbuffers; - use self::flatbuffers::{EndianScalar, Follow}; + extern crate flatbuffers; + use self::flatbuffers::{EndianScalar, Follow}; - pub enum PingBodyOffset {} - #[derive(Copy, Clone, PartialEq)] +pub enum PingBodyOffset {} +#[derive(Copy, Clone, PartialEq)] - pub struct PingBody<'a> { - pub _tab: flatbuffers::Table<'a>, +pub struct PingBody<'a> { + pub _tab: flatbuffers::Table<'a>, +} + +impl<'a> flatbuffers::Follow<'a> for PingBody<'a> { + type Inner = PingBody<'a>; + #[inline] + unsafe fn follow(buf: &'a [u8], loc: usize) -> Self::Inner { + Self { _tab: flatbuffers::Table::new(buf, loc) } + } +} + +impl<'a> PingBody<'a> { + pub const VT_PAYLOAD: flatbuffers::VOffsetT = 4; + + pub const fn get_fully_qualified_name() -> &'static str { + "models.PingBody" + } + + #[inline] + pub unsafe fn init_from_table(table: flatbuffers::Table<'a>) -> Self { + PingBody { _tab: table } + } + #[allow(unused_mut)] + pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr, A: flatbuffers::Allocator + 'bldr>( + _fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr, A>, + args: &'args PingBodyArgs<'args> + ) -> flatbuffers::WIPOffset> { + let mut builder = PingBodyBuilder::new(_fbb); + if let Some(x) = args.payload { builder.add_payload(x); } + builder.finish() + } + + + #[inline] + pub fn payload(&self) -> Option> { + // Safety: + // Created from valid Table for this object + // which contains a valid value in this slot + unsafe { self._tab.get::>>(PingBody::VT_PAYLOAD, None)} + } +} + +impl flatbuffers::Verifiable for PingBody<'_> { + #[inline] + fn run_verifier( + v: &mut flatbuffers::Verifier, pos: usize + ) -> Result<(), flatbuffers::InvalidFlatbuffer> { + use self::flatbuffers::Verifiable; + v.visit_table(pos)? + .visit_field::>>("payload", Self::VT_PAYLOAD, false)? + .finish(); + Ok(()) + } +} +pub struct PingBodyArgs<'a> { + pub payload: Option>>, +} +impl<'a> Default for PingBodyArgs<'a> { + #[inline] + fn default() -> Self { + PingBodyArgs { + payload: None, } + } +} - impl<'a> flatbuffers::Follow<'a> for PingBody<'a> { - type Inner = PingBody<'a>; - #[inline] - unsafe fn follow(buf: &'a [u8], loc: usize) -> Self::Inner { - Self { - _tab: flatbuffers::Table::new(buf, loc), - } - } +pub struct PingBodyBuilder<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> { + fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a, A>, + start_: flatbuffers::WIPOffset, +} +impl<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> PingBodyBuilder<'a, 'b, A> { + #[inline] + pub fn add_payload(&mut self, payload: flatbuffers::WIPOffset>) { + self.fbb_.push_slot_always::>(PingBody::VT_PAYLOAD, payload); + } + #[inline] + pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a, A>) -> PingBodyBuilder<'a, 'b, A> { + let start = _fbb.start_table(); + PingBodyBuilder { + fbb_: _fbb, + start_: start, } + } + #[inline] + pub fn finish(self) -> flatbuffers::WIPOffset> { + let o = self.fbb_.end_table(self.start_); + flatbuffers::WIPOffset::new(o.value()) + } +} - impl<'a> PingBody<'a> { - pub const VT_PAYLOAD: flatbuffers::VOffsetT = 4; +impl core::fmt::Debug for PingBody<'_> { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + let mut ds = f.debug_struct("PingBody"); + ds.field("payload", &self.payload()); + ds.finish() + } +} +} // pub mod models - pub const fn get_fully_qualified_name() -> &'static str { - "models.PingBody" - } - - #[inline] - pub unsafe fn init_from_table(table: flatbuffers::Table<'a>) -> Self { - PingBody { _tab: table } - } - #[allow(unused_mut)] - pub fn create<'bldr: 'args, 'args: 'mut_bldr, 'mut_bldr, A: flatbuffers::Allocator + 'bldr>( - _fbb: &'mut_bldr mut flatbuffers::FlatBufferBuilder<'bldr, A>, - args: &'args PingBodyArgs<'args>, - ) -> flatbuffers::WIPOffset> { - let mut builder = PingBodyBuilder::new(_fbb); - if let Some(x) = args.payload { - builder.add_payload(x); - } - builder.finish() - } - - #[inline] - pub fn payload(&self) -> Option> { - // Safety: - // Created from valid Table for this object - // which contains a valid value in this slot - unsafe { - self._tab - .get::>>(PingBody::VT_PAYLOAD, None) - } - } - } - - impl flatbuffers::Verifiable for PingBody<'_> { - #[inline] - fn run_verifier(v: &mut flatbuffers::Verifier, pos: usize) -> Result<(), flatbuffers::InvalidFlatbuffer> { - use self::flatbuffers::Verifiable; - v.visit_table(pos)? - .visit_field::>>("payload", Self::VT_PAYLOAD, false)? - .finish(); - Ok(()) - } - } - pub struct PingBodyArgs<'a> { - pub payload: Option>>, - } - impl<'a> Default for PingBodyArgs<'a> { - #[inline] - fn default() -> Self { - PingBodyArgs { payload: None } - } - } - - pub struct PingBodyBuilder<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> { - fbb_: &'b mut flatbuffers::FlatBufferBuilder<'a, A>, - start_: flatbuffers::WIPOffset, - } - impl<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> PingBodyBuilder<'a, 'b, A> { - #[inline] - pub fn add_payload(&mut self, payload: flatbuffers::WIPOffset>) { - self.fbb_ - .push_slot_always::>(PingBody::VT_PAYLOAD, payload); - } - #[inline] - pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a, A>) -> PingBodyBuilder<'a, 'b, A> { - let start = _fbb.start_table(); - PingBodyBuilder { - fbb_: _fbb, - start_: start, - } - } - #[inline] - pub fn finish(self) -> flatbuffers::WIPOffset> { - let o = self.fbb_.end_table(self.start_); - flatbuffers::WIPOffset::new(o.value()) - } - } - - impl core::fmt::Debug for PingBody<'_> { - fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { - let mut ds = f.debug_struct("PingBody"); - ds.field("payload", &self.payload()); - ds.finish() - } - } -} // pub mod models diff --git a/common/protos/src/generated/proto_gen/node_service.rs b/common/protos/src/generated/proto_gen/node_service.rs index 1910283c4..76a704908 100644 --- a/common/protos/src/generated/proto_gen/node_service.rs +++ b/common/protos/src/generated/proto_gen/node_service.rs @@ -566,8 +566,8 @@ pub struct DeleteVolumeResponse { /// Generated client implementations. pub mod node_service_client { #![allow(unused_variables, dead_code, missing_docs, clippy::let_unit_value)] - use tonic::codegen::http::Uri; use tonic::codegen::*; + use tonic::codegen::http::Uri; #[derive(Debug, Clone)] pub struct NodeServiceClient { inner: tonic::client::Grpc, @@ -598,15 +598,22 @@ pub mod node_service_client { let inner = tonic::client::Grpc::with_origin(inner, origin); Self { inner } } - pub fn with_interceptor(inner: T, interceptor: F) -> NodeServiceClient> + pub fn with_interceptor( + inner: T, + interceptor: F, + ) -> NodeServiceClient> where F: tonic::service::Interceptor, T::ResponseBody: Default, T: tonic::codegen::Service< http::Request, - Response = http::Response<>::ResponseBody>, + Response = http::Response< + >::ResponseBody, + >, >, - >>::Error: Into + Send + Sync, + , + >>::Error: Into + Send + Sync, { NodeServiceClient::new(InterceptedService::new(inner, interceptor)) } @@ -649,9 +656,16 @@ pub mod node_service_client { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/Ping"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/Ping", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "Ping")); @@ -660,13 +674,23 @@ pub mod node_service_client { pub async fn list_bucket( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ListBucket"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/ListBucket", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "ListBucket")); @@ -675,13 +699,23 @@ pub mod node_service_client { pub async fn make_bucket( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/MakeBucket"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/MakeBucket", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "MakeBucket")); @@ -690,13 +724,23 @@ pub mod node_service_client { pub async fn get_bucket_info( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/GetBucketInfo"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/GetBucketInfo", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "GetBucketInfo")); @@ -705,13 +749,23 @@ pub mod node_service_client { pub async fn delete_bucket( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/DeleteBucket"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/DeleteBucket", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "DeleteBucket")); @@ -720,13 +774,23 @@ pub mod node_service_client { pub async fn read_all( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ReadAll"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/ReadAll", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "ReadAll")); @@ -735,13 +799,23 @@ pub mod node_service_client { pub async fn write_all( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/WriteAll"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/WriteAll", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "WriteAll")); @@ -754,9 +828,16 @@ pub mod node_service_client { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/Delete"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/Delete", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "Delete")); @@ -765,13 +846,23 @@ pub mod node_service_client { pub async fn rename_part( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/RenamePart"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/RenamePart", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "RenamePart")); @@ -780,13 +871,23 @@ pub mod node_service_client { pub async fn rename_file( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/RenameFile"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/RenameFile", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "RenameFile")); @@ -799,9 +900,16 @@ pub mod node_service_client { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/Write"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/Write", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "Write")); @@ -815,9 +923,16 @@ pub mod node_service_client { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ReadAt"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/ReadAt", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "ReadAt")); @@ -826,13 +941,23 @@ pub mod node_service_client { pub async fn list_dir( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ListDir"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/ListDir", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "ListDir")); @@ -841,13 +966,23 @@ pub mod node_service_client { pub async fn walk_dir( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/WalkDir"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/WalkDir", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "WalkDir")); @@ -856,13 +991,23 @@ pub mod node_service_client { pub async fn rename_data( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/RenameData"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/RenameData", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "RenameData")); @@ -871,13 +1016,23 @@ pub mod node_service_client { pub async fn make_volumes( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/MakeVolumes"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/MakeVolumes", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "MakeVolumes")); @@ -886,13 +1041,23 @@ pub mod node_service_client { pub async fn make_volume( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/MakeVolume"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/MakeVolume", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "MakeVolume")); @@ -901,13 +1066,23 @@ pub mod node_service_client { pub async fn list_volumes( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ListVolumes"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/ListVolumes", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "ListVolumes")); @@ -916,13 +1091,23 @@ pub mod node_service_client { pub async fn stat_volume( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/StatVolume"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/StatVolume", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "StatVolume")); @@ -931,13 +1116,23 @@ pub mod node_service_client { pub async fn delete_paths( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/DeletePaths"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/DeletePaths", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "DeletePaths")); @@ -946,13 +1141,23 @@ pub mod node_service_client { pub async fn update_metadata( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/UpdateMetadata"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/UpdateMetadata", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "UpdateMetadata")); @@ -961,13 +1166,23 @@ pub mod node_service_client { pub async fn write_metadata( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/WriteMetadata"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/WriteMetadata", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "WriteMetadata")); @@ -976,13 +1191,23 @@ pub mod node_service_client { pub async fn read_version( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ReadVersion"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/ReadVersion", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "ReadVersion")); @@ -995,9 +1220,16 @@ pub mod node_service_client { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ReadXL"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/ReadXL", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "ReadXL")); @@ -1006,13 +1238,23 @@ pub mod node_service_client { pub async fn delete_version( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/DeleteVersion"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/DeleteVersion", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "DeleteVersion")); @@ -1021,13 +1263,23 @@ pub mod node_service_client { pub async fn delete_versions( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/DeleteVersions"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/DeleteVersions", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "DeleteVersions")); @@ -1036,13 +1288,23 @@ pub mod node_service_client { pub async fn read_multiple( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/ReadMultiple"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/ReadMultiple", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "ReadMultiple")); @@ -1051,13 +1313,23 @@ pub mod node_service_client { pub async fn delete_volume( &mut self, request: impl tonic::IntoRequest, - ) -> std::result::Result, tonic::Status> { + ) -> std::result::Result< + tonic::Response, + tonic::Status, + > { self.inner .ready() .await - .map_err(|e| tonic::Status::new(tonic::Code::Unknown, format!("Service was not ready: {}", e.into())))?; + .map_err(|e| { + tonic::Status::new( + tonic::Code::Unknown, + format!("Service was not ready: {}", e.into()), + ) + })?; let codec = tonic::codec::ProstCodec::default(); - let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/DeleteVolume"); + let path = http::uri::PathAndQuery::from_static( + "/node_service.NodeService/DeleteVolume", + ); let mut req = request.into_request(); req.extensions_mut() .insert(GrpcMethod::new("node_service.NodeService", "DeleteVolume")); @@ -1080,19 +1352,31 @@ pub mod node_service_server { async fn list_bucket( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn make_bucket( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn get_bucket_info( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn delete_bucket( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn read_all( &self, request: tonic::Request, @@ -1100,7 +1384,10 @@ pub mod node_service_server { async fn write_all( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn delete( &self, request: tonic::Request, @@ -1108,11 +1395,17 @@ pub mod node_service_server { async fn rename_part( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn rename_file( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn write( &self, request: tonic::Request, @@ -1133,39 +1426,66 @@ pub mod node_service_server { async fn rename_data( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn make_volumes( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn make_volume( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn list_volumes( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn stat_volume( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn delete_paths( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn update_metadata( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn write_metadata( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn read_version( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn read_xl( &self, request: tonic::Request, @@ -1173,19 +1493,31 @@ pub mod node_service_server { async fn delete_version( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn delete_versions( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn read_multiple( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; async fn delete_volume( &self, request: tonic::Request, - ) -> std::result::Result, tonic::Status>; + ) -> std::result::Result< + tonic::Response, + tonic::Status, + >; } #[derive(Debug)] pub struct NodeServiceServer { @@ -1208,7 +1540,10 @@ pub mod node_service_server { max_encoding_message_size: None, } } - pub fn with_interceptor(inner: T, interceptor: F) -> InterceptedService + pub fn with_interceptor( + inner: T, + interceptor: F, + ) -> InterceptedService where F: tonic::service::Interceptor, { @@ -1252,7 +1587,10 @@ pub mod node_service_server { type Response = http::Response; type Error = std::convert::Infallible; type Future = BoxFuture; - fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll> { + fn poll_ready( + &mut self, + _cx: &mut Context<'_>, + ) -> Poll> { Poll::Ready(Ok(())) } fn call(&mut self, req: http::Request) -> Self::Future { @@ -1260,12 +1598,21 @@ pub mod node_service_server { "/node_service.NodeService/Ping" => { #[allow(non_camel_case_types)] struct PingSvc(pub Arc); - impl tonic::server::UnaryService for PingSvc { + impl tonic::server::UnaryService + for PingSvc { type Response = super::PingResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::ping(&inner, request).await }; + let fut = async move { + ::ping(&inner, request).await + }; Box::pin(fut) } } @@ -1278,8 +1625,14 @@ pub mod node_service_server { let method = PingSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1288,12 +1641,23 @@ pub mod node_service_server { "/node_service.NodeService/ListBucket" => { #[allow(non_camel_case_types)] struct ListBucketSvc(pub Arc); - impl tonic::server::UnaryService for ListBucketSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for ListBucketSvc { type Response = super::ListBucketResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::list_bucket(&inner, request).await }; + let fut = async move { + ::list_bucket(&inner, request).await + }; Box::pin(fut) } } @@ -1306,8 +1670,14 @@ pub mod node_service_server { let method = ListBucketSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1316,12 +1686,23 @@ pub mod node_service_server { "/node_service.NodeService/MakeBucket" => { #[allow(non_camel_case_types)] struct MakeBucketSvc(pub Arc); - impl tonic::server::UnaryService for MakeBucketSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for MakeBucketSvc { type Response = super::MakeBucketResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::make_bucket(&inner, request).await }; + let fut = async move { + ::make_bucket(&inner, request).await + }; Box::pin(fut) } } @@ -1334,8 +1715,14 @@ pub mod node_service_server { let method = MakeBucketSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1344,12 +1731,23 @@ pub mod node_service_server { "/node_service.NodeService/GetBucketInfo" => { #[allow(non_camel_case_types)] struct GetBucketInfoSvc(pub Arc); - impl tonic::server::UnaryService for GetBucketInfoSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for GetBucketInfoSvc { type Response = super::GetBucketInfoResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::get_bucket_info(&inner, request).await }; + let fut = async move { + ::get_bucket_info(&inner, request).await + }; Box::pin(fut) } } @@ -1362,8 +1760,14 @@ pub mod node_service_server { let method = GetBucketInfoSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1372,12 +1776,23 @@ pub mod node_service_server { "/node_service.NodeService/DeleteBucket" => { #[allow(non_camel_case_types)] struct DeleteBucketSvc(pub Arc); - impl tonic::server::UnaryService for DeleteBucketSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for DeleteBucketSvc { type Response = super::DeleteBucketResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::delete_bucket(&inner, request).await }; + let fut = async move { + ::delete_bucket(&inner, request).await + }; Box::pin(fut) } } @@ -1390,8 +1805,14 @@ pub mod node_service_server { let method = DeleteBucketSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1400,12 +1821,23 @@ pub mod node_service_server { "/node_service.NodeService/ReadAll" => { #[allow(non_camel_case_types)] struct ReadAllSvc(pub Arc); - impl tonic::server::UnaryService for ReadAllSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for ReadAllSvc { type Response = super::ReadAllResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::read_all(&inner, request).await }; + let fut = async move { + ::read_all(&inner, request).await + }; Box::pin(fut) } } @@ -1418,8 +1850,14 @@ pub mod node_service_server { let method = ReadAllSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1428,12 +1866,23 @@ pub mod node_service_server { "/node_service.NodeService/WriteAll" => { #[allow(non_camel_case_types)] struct WriteAllSvc(pub Arc); - impl tonic::server::UnaryService for WriteAllSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for WriteAllSvc { type Response = super::WriteAllResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::write_all(&inner, request).await }; + let fut = async move { + ::write_all(&inner, request).await + }; Box::pin(fut) } } @@ -1446,8 +1895,14 @@ pub mod node_service_server { let method = WriteAllSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1456,12 +1911,23 @@ pub mod node_service_server { "/node_service.NodeService/Delete" => { #[allow(non_camel_case_types)] struct DeleteSvc(pub Arc); - impl tonic::server::UnaryService for DeleteSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for DeleteSvc { type Response = super::DeleteResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::delete(&inner, request).await }; + let fut = async move { + ::delete(&inner, request).await + }; Box::pin(fut) } } @@ -1474,8 +1940,14 @@ pub mod node_service_server { let method = DeleteSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1484,12 +1956,23 @@ pub mod node_service_server { "/node_service.NodeService/RenamePart" => { #[allow(non_camel_case_types)] struct RenamePartSvc(pub Arc); - impl tonic::server::UnaryService for RenamePartSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for RenamePartSvc { type Response = super::RenamePartResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::rename_part(&inner, request).await }; + let fut = async move { + ::rename_part(&inner, request).await + }; Box::pin(fut) } } @@ -1502,8 +1985,14 @@ pub mod node_service_server { let method = RenamePartSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1512,12 +2001,23 @@ pub mod node_service_server { "/node_service.NodeService/RenameFile" => { #[allow(non_camel_case_types)] struct RenameFileSvc(pub Arc); - impl tonic::server::UnaryService for RenameFileSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for RenameFileSvc { type Response = super::RenameFileResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::rename_file(&inner, request).await }; + let fut = async move { + ::rename_file(&inner, request).await + }; Box::pin(fut) } } @@ -1530,8 +2030,14 @@ pub mod node_service_server { let method = RenameFileSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1540,12 +2046,21 @@ pub mod node_service_server { "/node_service.NodeService/Write" => { #[allow(non_camel_case_types)] struct WriteSvc(pub Arc); - impl tonic::server::UnaryService for WriteSvc { + impl tonic::server::UnaryService + for WriteSvc { type Response = super::WriteResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::write(&inner, request).await }; + let fut = async move { + ::write(&inner, request).await + }; Box::pin(fut) } } @@ -1558,8 +2073,14 @@ pub mod node_service_server { let method = WriteSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1568,12 +2089,23 @@ pub mod node_service_server { "/node_service.NodeService/ReadAt" => { #[allow(non_camel_case_types)] struct ReadAtSvc(pub Arc); - impl tonic::server::UnaryService for ReadAtSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for ReadAtSvc { type Response = super::ReadAtResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::read_at(&inner, request).await }; + let fut = async move { + ::read_at(&inner, request).await + }; Box::pin(fut) } } @@ -1586,8 +2118,14 @@ pub mod node_service_server { let method = ReadAtSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1596,12 +2134,23 @@ pub mod node_service_server { "/node_service.NodeService/ListDir" => { #[allow(non_camel_case_types)] struct ListDirSvc(pub Arc); - impl tonic::server::UnaryService for ListDirSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for ListDirSvc { type Response = super::ListDirResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::list_dir(&inner, request).await }; + let fut = async move { + ::list_dir(&inner, request).await + }; Box::pin(fut) } } @@ -1614,8 +2163,14 @@ pub mod node_service_server { let method = ListDirSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1624,12 +2179,23 @@ pub mod node_service_server { "/node_service.NodeService/WalkDir" => { #[allow(non_camel_case_types)] struct WalkDirSvc(pub Arc); - impl tonic::server::UnaryService for WalkDirSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for WalkDirSvc { type Response = super::WalkDirResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::walk_dir(&inner, request).await }; + let fut = async move { + ::walk_dir(&inner, request).await + }; Box::pin(fut) } } @@ -1642,8 +2208,14 @@ pub mod node_service_server { let method = WalkDirSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1652,12 +2224,23 @@ pub mod node_service_server { "/node_service.NodeService/RenameData" => { #[allow(non_camel_case_types)] struct RenameDataSvc(pub Arc); - impl tonic::server::UnaryService for RenameDataSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for RenameDataSvc { type Response = super::RenameDataResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::rename_data(&inner, request).await }; + let fut = async move { + ::rename_data(&inner, request).await + }; Box::pin(fut) } } @@ -1670,8 +2253,14 @@ pub mod node_service_server { let method = RenameDataSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1680,12 +2269,23 @@ pub mod node_service_server { "/node_service.NodeService/MakeVolumes" => { #[allow(non_camel_case_types)] struct MakeVolumesSvc(pub Arc); - impl tonic::server::UnaryService for MakeVolumesSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for MakeVolumesSvc { type Response = super::MakeVolumesResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::make_volumes(&inner, request).await }; + let fut = async move { + ::make_volumes(&inner, request).await + }; Box::pin(fut) } } @@ -1698,8 +2298,14 @@ pub mod node_service_server { let method = MakeVolumesSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1708,12 +2314,23 @@ pub mod node_service_server { "/node_service.NodeService/MakeVolume" => { #[allow(non_camel_case_types)] struct MakeVolumeSvc(pub Arc); - impl tonic::server::UnaryService for MakeVolumeSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for MakeVolumeSvc { type Response = super::MakeVolumeResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::make_volume(&inner, request).await }; + let fut = async move { + ::make_volume(&inner, request).await + }; Box::pin(fut) } } @@ -1726,8 +2343,14 @@ pub mod node_service_server { let method = MakeVolumeSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1736,12 +2359,23 @@ pub mod node_service_server { "/node_service.NodeService/ListVolumes" => { #[allow(non_camel_case_types)] struct ListVolumesSvc(pub Arc); - impl tonic::server::UnaryService for ListVolumesSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for ListVolumesSvc { type Response = super::ListVolumesResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::list_volumes(&inner, request).await }; + let fut = async move { + ::list_volumes(&inner, request).await + }; Box::pin(fut) } } @@ -1754,8 +2388,14 @@ pub mod node_service_server { let method = ListVolumesSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1764,12 +2404,23 @@ pub mod node_service_server { "/node_service.NodeService/StatVolume" => { #[allow(non_camel_case_types)] struct StatVolumeSvc(pub Arc); - impl tonic::server::UnaryService for StatVolumeSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for StatVolumeSvc { type Response = super::StatVolumeResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::stat_volume(&inner, request).await }; + let fut = async move { + ::stat_volume(&inner, request).await + }; Box::pin(fut) } } @@ -1782,8 +2433,14 @@ pub mod node_service_server { let method = StatVolumeSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1792,12 +2449,23 @@ pub mod node_service_server { "/node_service.NodeService/DeletePaths" => { #[allow(non_camel_case_types)] struct DeletePathsSvc(pub Arc); - impl tonic::server::UnaryService for DeletePathsSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for DeletePathsSvc { type Response = super::DeletePathsResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::delete_paths(&inner, request).await }; + let fut = async move { + ::delete_paths(&inner, request).await + }; Box::pin(fut) } } @@ -1810,8 +2478,14 @@ pub mod node_service_server { let method = DeletePathsSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1820,12 +2494,23 @@ pub mod node_service_server { "/node_service.NodeService/UpdateMetadata" => { #[allow(non_camel_case_types)] struct UpdateMetadataSvc(pub Arc); - impl tonic::server::UnaryService for UpdateMetadataSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for UpdateMetadataSvc { type Response = super::UpdateMetadataResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::update_metadata(&inner, request).await }; + let fut = async move { + ::update_metadata(&inner, request).await + }; Box::pin(fut) } } @@ -1838,8 +2523,14 @@ pub mod node_service_server { let method = UpdateMetadataSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1848,12 +2539,23 @@ pub mod node_service_server { "/node_service.NodeService/WriteMetadata" => { #[allow(non_camel_case_types)] struct WriteMetadataSvc(pub Arc); - impl tonic::server::UnaryService for WriteMetadataSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for WriteMetadataSvc { type Response = super::WriteMetadataResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::write_metadata(&inner, request).await }; + let fut = async move { + ::write_metadata(&inner, request).await + }; Box::pin(fut) } } @@ -1866,8 +2568,14 @@ pub mod node_service_server { let method = WriteMetadataSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1876,12 +2584,23 @@ pub mod node_service_server { "/node_service.NodeService/ReadVersion" => { #[allow(non_camel_case_types)] struct ReadVersionSvc(pub Arc); - impl tonic::server::UnaryService for ReadVersionSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for ReadVersionSvc { type Response = super::ReadVersionResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::read_version(&inner, request).await }; + let fut = async move { + ::read_version(&inner, request).await + }; Box::pin(fut) } } @@ -1894,8 +2613,14 @@ pub mod node_service_server { let method = ReadVersionSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1904,12 +2629,23 @@ pub mod node_service_server { "/node_service.NodeService/ReadXL" => { #[allow(non_camel_case_types)] struct ReadXLSvc(pub Arc); - impl tonic::server::UnaryService for ReadXLSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for ReadXLSvc { type Response = super::ReadXlResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::read_xl(&inner, request).await }; + let fut = async move { + ::read_xl(&inner, request).await + }; Box::pin(fut) } } @@ -1922,8 +2658,14 @@ pub mod node_service_server { let method = ReadXLSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1932,12 +2674,23 @@ pub mod node_service_server { "/node_service.NodeService/DeleteVersion" => { #[allow(non_camel_case_types)] struct DeleteVersionSvc(pub Arc); - impl tonic::server::UnaryService for DeleteVersionSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for DeleteVersionSvc { type Response = super::DeleteVersionResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::delete_version(&inner, request).await }; + let fut = async move { + ::delete_version(&inner, request).await + }; Box::pin(fut) } } @@ -1950,8 +2703,14 @@ pub mod node_service_server { let method = DeleteVersionSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1960,12 +2719,23 @@ pub mod node_service_server { "/node_service.NodeService/DeleteVersions" => { #[allow(non_camel_case_types)] struct DeleteVersionsSvc(pub Arc); - impl tonic::server::UnaryService for DeleteVersionsSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for DeleteVersionsSvc { type Response = super::DeleteVersionsResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::delete_versions(&inner, request).await }; + let fut = async move { + ::delete_versions(&inner, request).await + }; Box::pin(fut) } } @@ -1978,8 +2748,14 @@ pub mod node_service_server { let method = DeleteVersionsSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -1988,12 +2764,23 @@ pub mod node_service_server { "/node_service.NodeService/ReadMultiple" => { #[allow(non_camel_case_types)] struct ReadMultipleSvc(pub Arc); - impl tonic::server::UnaryService for ReadMultipleSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for ReadMultipleSvc { type Response = super::ReadMultipleResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::read_multiple(&inner, request).await }; + let fut = async move { + ::read_multiple(&inner, request).await + }; Box::pin(fut) } } @@ -2006,8 +2793,14 @@ pub mod node_service_server { let method = ReadMultipleSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; @@ -2016,12 +2809,23 @@ pub mod node_service_server { "/node_service.NodeService/DeleteVolume" => { #[allow(non_camel_case_types)] struct DeleteVolumeSvc(pub Arc); - impl tonic::server::UnaryService for DeleteVolumeSvc { + impl< + T: NodeService, + > tonic::server::UnaryService + for DeleteVolumeSvc { type Response = super::DeleteVolumeResponse; - type Future = BoxFuture, tonic::Status>; - fn call(&mut self, request: tonic::Request) -> Self::Future { + type Future = BoxFuture< + tonic::Response, + tonic::Status, + >; + fn call( + &mut self, + request: tonic::Request, + ) -> Self::Future { let inner = Arc::clone(&self.0); - let fut = async move { ::delete_volume(&inner, request).await }; + let fut = async move { + ::delete_volume(&inner, request).await + }; Box::pin(fut) } } @@ -2034,21 +2838,34 @@ pub mod node_service_server { let method = DeleteVolumeSvc(inner); let codec = tonic::codec::ProstCodec::default(); let mut grpc = tonic::server::Grpc::new(codec) - .apply_compression_config(accept_compression_encodings, send_compression_encodings) - .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size); + .apply_compression_config( + accept_compression_encodings, + send_compression_encodings, + ) + .apply_max_message_size_config( + max_decoding_message_size, + max_encoding_message_size, + ); let res = grpc.unary(method, req).await; Ok(res) }; Box::pin(fut) } - _ => Box::pin(async move { - Ok(http::Response::builder() - .status(200) - .header("grpc-status", tonic::Code::Unimplemented as i32) - .header(http::header::CONTENT_TYPE, tonic::metadata::GRPC_CONTENT_TYPE) - .body(empty_body()) - .unwrap()) - }), + _ => { + Box::pin(async move { + Ok( + http::Response::builder() + .status(200) + .header("grpc-status", tonic::Code::Unimplemented as i32) + .header( + http::header::CONTENT_TYPE, + tonic::metadata::GRPC_CONTENT_TYPE, + ) + .body(empty_body()) + .unwrap(), + ) + }) + } } } } diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 9eb470bae..ea6b5d0b5 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -10,7 +10,7 @@ use crate::disk::error::{ }; use crate::disk::os::check_path_length; use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE}; -use crate::utils::fs::{lstat, O_RDONLY}; +use crate::utils::fs::{lstat, O_APPEND, O_CREATE, O_RDONLY, O_WRONLY}; use crate::utils::path::{has_suffix, SLASH_SEPARATOR}; use crate::{ error::{Error, Result}, @@ -26,8 +26,8 @@ use std::{ use time::OffsetDateTime; use tokio::fs::{self, File}; use tokio::io::{AsyncReadExt, AsyncWriteExt, ErrorKind}; -use tokio::sync::Mutex; -use tracing::{debug, warn}; +use tokio::sync::RwLock; +use tracing::{error, warn}; use uuid::Uuid; #[derive(Debug)] @@ -52,7 +52,7 @@ impl FormatInfo { pub struct LocalDisk { pub root: PathBuf, pub format_path: PathBuf, - pub format_info: Mutex, + pub format_info: RwLock, pub endpoint: Endpoint, // pub id: Mutex>, // pub format_data: Mutex>, @@ -106,7 +106,7 @@ impl LocalDisk { root, endpoint: ep.clone(), format_path: format_path, - format_info: Mutex::new(format_info), + format_info: RwLock::new(format_info), // // format_legacy, // format_file_info: Mutex::new(format_meta), // format_data: Mutex::new(format_data), @@ -197,21 +197,6 @@ impl LocalDisk { // }) // } - pub async fn rename_all(&self, src_data_path: &PathBuf, dst_data_path: &PathBuf, skip: &PathBuf) -> Result<()> { - if !skip.starts_with(src_data_path) { - fs::create_dir_all(dst_data_path.parent().unwrap_or(Path::new("/"))).await?; - } - - // debug!( - // "rename_all from \n {:?} \n to \n {:?} \n skip:{:?}", - // &src_data_path, &dst_data_path, &skip - // ); - - fs::rename(&src_data_path, &dst_data_path).await?; - - Ok(()) - } - pub async fn move_to_trash(&self, delete_path: &PathBuf, _recursive: bool, _immediate_purge: bool) -> Result<()> { let trash_path = self.get_object_path(super::RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?; if let Some(parent) = trash_path.parent() { @@ -332,6 +317,12 @@ impl LocalDisk { } } + async fn read_metadata(&self, file_path: impl AsRef) -> Result> { + // TODO: suport timeout + let (data, _) = self.read_metadata_with_dmtime(file_path.as_ref()).await?; + Ok(data) + } + // FIXME: read_metadata only suport async fn read_metadata_with_dmtime(&self, file_path: impl AsRef) -> Result<(Vec, Option)> { check_path_length(&file_path.as_ref().to_string_lossy().to_string())?; @@ -499,7 +490,7 @@ impl LocalDisk { // write_all_public for trail async fn write_all_public(&self, volume: &str, path: &str, data: Vec) -> Result<()> { if volume == super::RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { - let mut format_info = self.format_info.lock().await; + let mut format_info = self.format_info.write().await; format_info.data = data.clone(); } @@ -624,15 +615,6 @@ pub async fn read_file_metadata(p: impl AsRef) -> Result { Ok(meta) } -pub async fn check_volume_exists(p: impl AsRef) -> Result<()> { - fs::metadata(&p).await.map_err(|e| match e.kind() { - ErrorKind::NotFound => Error::from(DiskError::VolumeNotFound), - ErrorKind::PermissionDenied => Error::from(DiskError::FileAccessDenied), - _ => Error::from(e), - })?; - Ok(()) -} - fn skip_access_checks(p: impl AsRef) -> bool { let vols = [ super::RUSTFS_META_TMP_DELETED_BUCKET, @@ -685,8 +667,7 @@ impl DiskAPI for LocalDisk { } async fn get_disk_id(&self) -> Result> { - // TODO: check format file - let mut format_info = self.format_info.lock().await; + let mut format_info = self.format_info.write().await; let id = format_info.id.clone(); @@ -737,18 +718,24 @@ impl DiskAPI for LocalDisk { format_info.last_check = Some(OffsetDateTime::now_utc()); Ok(Some(disk_id)) - // TODO: 判断源文件id,是否有效 } async fn set_disk_id(&self, id: Option) -> Result<()> { // 本地不需要设置 - let mut format_info = self.format_info.lock().await; + // TODO: add check_id_store + let mut format_info = self.format_info.write().await; format_info.id = id; Ok(()) } #[must_use] async fn read_all(&self, volume: &str, path: &str) -> Result> { + if volume == super::RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE { + let format_info = self.format_info.read().await; + if format_info.data.len() > 0 { + return Ok(format_info.data.clone()); + } + } // TOFIX: let p = self.get_object_path(volume, path)?; let (data, _) = read_file_all(&p).await?; @@ -761,29 +748,22 @@ impl DiskAPI for LocalDisk { } async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { - let vol_path = self.get_bucket_path(volume)?; + let volume_dir = self.get_bucket_path(volume)?; if !skip_access_checks(volume) { - check_volume_exists(&vol_path).await?; + if let Err(e) = utils::fs::access(&volume_dir).await { + return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); + } } - let fpath = self.get_object_path(volume, path)?; + let file_path = volume_dir.join(Path::new(&path)); + check_path_length(file_path.to_string_lossy().to_string().as_str())?; - self.delete_file(&vol_path, &fpath, opt.recursive, opt.immediate).await?; - - // if opt.recursive { - // let trash_path = self.get_object_path(RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?; - // fs::create_dir_all(&trash_path).await?; - // fs::rename(&fpath, &trash_path).await?; - - // // TODO: immediate - - // return Ok(()); - // } - - // fs::remove_file(fpath).await?; + self.delete_file(&volume_dir, &file_path, opt.recursive, opt.immediate) + .await?; Ok(()) } + async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()> { let src_volume_dir = self.get_bucket_path(src_volume)?; let dst_volume_dir = self.get_bucket_path(dst_volume)?; @@ -866,156 +846,220 @@ impl DiskAPI for LocalDisk { Ok(()) } async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> { - let src_volume_path = self.get_bucket_path(src_volume)?; - let dst_volume_path = self.get_bucket_path(dst_volume)?; + let src_volume_dir = self.get_bucket_path(src_volume)?; + let dst_volume_dir = self.get_bucket_path(dst_volume)?; if !skip_access_checks(src_volume) { - utils::fs::access(&src_volume_path).await.map_err(map_err_not_exists)?; + if let Err(e) = utils::fs::access(&src_volume_dir).await { + if os_is_not_exist(&e) { + return Err(Error::from(DiskError::VolumeNotFound)); + } else if is_sys_err_io(&e) { + return Err(Error::from(DiskError::FaultyDisk)); + } + + return Err(Error::new(e)); + } } if !skip_access_checks(dst_volume) { - utils::fs::access(&dst_volume_path).await.map_err(map_err_not_exists)?; + if let Err(e) = utils::fs::access(&dst_volume_dir).await { + if os_is_not_exist(&e) { + return Err(Error::from(DiskError::VolumeNotFound)); + } else if is_sys_err_io(&e) { + return Err(Error::from(DiskError::FaultyDisk)); + } + + return Err(Error::new(e)); + } } - let srcp = self.get_object_path(src_volume, src_path)?; - let dstp = self.get_object_path(dst_volume, dst_path)?; - - let src_is_dir = srcp.is_dir(); - let dst_is_dir = dstp.is_dir(); - if !src_is_dir && dst_is_dir || src_is_dir && !dst_is_dir { + let src_is_dir = has_suffix(&src_path, SLASH_SEPARATOR); + let dst_is_dir = has_suffix(&dst_path, SLASH_SEPARATOR); + if !(src_is_dir && dst_is_dir || !src_is_dir && !dst_is_dir) { return Err(Error::from(DiskError::FileAccessDenied)); } - // TODO: check path length + let src_file_path = src_volume_dir.join(Path::new(&src_path)); + check_path_length(src_file_path.to_string_lossy().to_string().as_str())?; + + let dst_file_path = dst_volume_dir.join(Path::new(&dst_path)); + check_path_length(dst_file_path.to_string_lossy().to_string().as_str())?; if src_is_dir { - // TODO: remove dst_dir - } + let meta_op = match lstat(&src_file_path).await { + Ok(meta) => Some(meta), + Err(e) => { + if is_sys_err_io(&e) { + return Err(Error::new(DiskError::FaultyDisk)); + } - fs::create_dir_all(dstp.parent().unwrap_or_else(|| Path::new("."))).await?; - - let mut idx = 0; - loop { - if let Err(e) = fs::rename(&srcp, &dstp).await { - if e.kind() == ErrorKind::NotFound && idx == 0 { - idx += 1; - continue; + if !os_is_not_exist(&e) { + return Err(Error::new(e)); + } + None } }; - break; + if let Some(meta) = meta_op { + if !meta.is_dir() { + return Err(Error::new(DiskError::FileAccessDenied)); + } + } + + if let Err(e) = utils::fs::remove(&dst_file_path).await { + if is_sys_err_not_empty(&e) || is_sys_err_not_dir(&e) { + return Err(Error::new(DiskError::FileAccessDenied)); + } else if is_sys_err_io(&e) { + return Err(Error::new(DiskError::FaultyDisk)); + } + + return Err(Error::new(e)); + } } - if let Some(dir_path) = srcp.parent() { - self.delete_file(&src_volume_path, &PathBuf::from(dir_path), false, false) - .await?; + if let Err(err) = os::rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await { + if let Some(e) = err.to_io_err() { + if is_sys_err_not_empty(&e) || is_sys_err_not_dir(&e) { + return Err(Error::new(DiskError::FileAccessDenied)); + } + + return Err(os_err_to_file_err(e)); + } + + return Err(err); + } + + if let Some(parent) = src_file_path.parent() { + let _ = self.delete_file(&src_volume_dir, &parent.to_path_buf(), false, false).await; } Ok(()) } - async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result { - let volpath = self.get_bucket_path(&volume)?; - // check exists - fs::metadata(&volpath).await.map_err(|e| match e.kind() { - ErrorKind::NotFound => Error::new(DiskError::VolumeNotFound), - _ => Error::new(e), - })?; - - let fpath = self.get_object_path(volume, path)?; - - debug!("CreateFile fpath: {:?}", fpath); - - if let Some(_dir_path) = fpath.parent() { - fs::create_dir_all(&_dir_path).await?; + // TODO: use io.reader + async fn create_file(&self, origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result { + if !origvolume.is_empty() { + let origvolume_dir = self.get_bucket_path(&origvolume)?; + if !skip_access_checks(origvolume) { + if let Err(e) = utils::fs::access(origvolume_dir).await { + return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); + } + } } - let file = File::create(&fpath).await?; + let volume_dir = self.get_bucket_path(&volume)?; + let file_path = volume_dir.join(Path::new(&path)); + check_path_length(file_path.to_string_lossy().to_string().as_str())?; - Ok(FileWriter::Local(LocalFileWriter::new(file))) - // Ok(FileWriter::new(file)) + // TODO: writeAllDirect io.copy + if let Some(parent) = file_path.parent() { + os::make_dir_all(parent, &volume_dir).await?; + } + let f = utils::fs::open_file(&file_path, O_CREATE | O_WRONLY) + .await + .map_err(os_err_to_file_err)?; - // let mut writer = BufWriter::new(file); - - // io::copy(&mut r, &mut writer).await?; + Ok(FileWriter::Local(LocalFileWriter::new(f))) // Ok(()) } // async fn append_file(&self, volume: &str, path: &str, mut r: DuplexStream) -> Result { async fn append_file(&self, volume: &str, path: &str) -> Result { - let p = self.get_object_path(volume, path)?; - - if let Some(dir_path) = p.parent() { - fs::create_dir_all(&dir_path).await?; - } - - // debug!("append_file open {} {:?}", self.id(), &p); - - let file = File::options() - .read(true) - .create(true) - .write(true) - .append(true) - .open(&p) - .await?; - - Ok(FileWriter::Local(LocalFileWriter::new(file))) - // Ok(FileWriter::new(file)) - - // let mut writer = BufWriter::new(file); - - // io::copy(&mut r, &mut writer).await?; - - // debug!("append_file end {} {}", self.id(), path); - // io::copy(&mut r, &mut file).await?; - - // Ok(()) - } - async fn read_file(&self, volume: &str, path: &str) -> Result { - let p = self.get_object_path(volume, path)?; - - // debug!("read_file {:?}", &p); - let file = File::options().read(true).open(&p).await?; - - Ok(FileReader::Local(LocalFileReader::new(file))) - - // file.seek(SeekFrom::Start(offset as u64)).await?; - - // let mut buffer = vec![0; length]; - - // let bytes_read = file.read(&mut buffer).await?; - - // buffer.truncate(bytes_read); - - // Ok((buffer, bytes_read)) - } - async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result> { - let p = self.get_bucket_path(volume)?; - - let mut entries = fs::read_dir(&p).await?; - - let mut volumes = Vec::new(); - - while let Some(entry) = entries.next_entry().await? { - if let Ok(_metadata) = entry.metadata().await { - // if !metadata.is_dir() { - // continue; - // } - - let name = entry.file_name().to_string_lossy().to_string(); - - // let created = match metadata.created() { - // Ok(md) => OffsetDateTime::from(md), - // Err(_) => return Err(Error::msg("Not supported created on this platform")), - // }; - - volumes.push(name); + let volume_dir = self.get_bucket_path(&volume)?; + if !skip_access_checks(volume) { + if let Err(e) = utils::fs::access(&volume_dir).await { + return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); } } - Ok(volumes) + let file_path = volume_dir.join(Path::new(&path)); + check_path_length(file_path.to_string_lossy().to_string().as_str())?; + + let f = self.open_file(file_path, O_CREATE | O_APPEND | O_WRONLY, volume_dir).await?; + + Ok(FileWriter::Local(LocalFileWriter::new(f))) } + // TODO: io verifier + async fn read_file(&self, volume: &str, path: &str) -> Result { + let volume_dir = self.get_bucket_path(&volume)?; + if !skip_access_checks(volume) { + if let Err(e) = utils::fs::access(&volume_dir).await { + return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); + } + } + + let file_path = volume_dir.join(Path::new(&path)); + check_path_length(file_path.to_string_lossy().to_string().as_str())?; + + let f = self.open_file(file_path, O_RDONLY, volume_dir).await.map_err(|err| { + if let Some(e) = err.to_io_err() { + if os_is_not_exist(&e) { + Error::new(DiskError::FileNotFound) + } else if os_is_permission(&e) { + Error::new(DiskError::FileAccessDenied) + } else if is_sys_err_not_dir(&e) { + Error::new(DiskError::FileAccessDenied) + } else if is_sys_err_io(&e) { + Error::new(DiskError::FaultyDisk) + } else if is_sys_err_too_many_files(&e) { + Error::new(DiskError::TooManyOpenFiles) + } else { + Error::new(e) + } + } else { + err + } + })?; + + Ok(FileReader::Local(LocalFileReader::new(f))) + } + + async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result> { + if !origvolume.is_empty() { + let origvolume_dir = self.get_bucket_path(&origvolume)?; + if !skip_access_checks(origvolume) { + if let Err(e) = utils::fs::access(origvolume_dir).await { + return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); + } + } + } + + let volume_dir = self.get_bucket_path(&volume)?; + let dir_path_abs = volume_dir.join(Path::new(&dir_path)); + + let entries = match os::read_dir(&dir_path_abs, count).await { + Ok(res) => res, + Err(e) => { + if DiskError::FileNotFound.is(&e) && !skip_access_checks(&volume) { + if let Err(e) = utils::fs::access(&volume_dir).await { + return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); + } + } + + return Err(e); + } + }; + Ok(entries) + } + + // TODO: io.writer async fn walk_dir(&self, opts: WalkDirOptions) -> Result> { - let mut entries = self.list_dir("", &opts.bucket, &opts.base_dir, -1).await?; + let mut entries = match self.list_dir("", &opts.bucket, &opts.base_dir, -1).await { + Ok(res) => res, + Err(e) => { + if !DiskError::VolumeNotFound.is(&e) && !DiskError::FileNotFound.is(&e) { + error!("list_dir err {:?}", &e); + } + + if opts.report_notfound && DiskError::FileNotFound.is(&e) { + return Err(Error::new(DiskError::FileNotFound)); + } + return Ok(Vec::new()); + } + }; + + if entries.is_empty() { + return Ok(Vec::new()); + } entries.sort(); @@ -1046,12 +1090,7 @@ impl DiskAPI for LocalDisk { let fpath = self.get_object_path(bucket, format!("{}/{}", &meta.name, STORAGE_FORMAT_FILE).as_str())?; - let (fdata, _) = match self.read_metadata_with_dmtime(&fpath).await { - Ok(res) => res, - Err(_) => (Vec::new(), None), - }; - - meta.metadata = fdata; + meta.metadata = self.read_metadata(&fpath).await.unwrap_or_default(); metas.push(meta); } @@ -1068,107 +1107,196 @@ impl DiskAPI for LocalDisk { dst_volume: &str, dst_path: &str, ) -> Result { - let src_volume_path = self.get_bucket_path(src_volume)?; + let src_volume_dir = self.get_bucket_path(src_volume)?; if !skip_access_checks(src_volume) { - check_volume_exists(&src_volume_path).await?; + if let Err(e) = utils::fs::access(&src_volume_dir).await { + return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); + } } - let dst_volume_path = self.get_bucket_path(dst_volume)?; + let dst_volume_dir = self.get_bucket_path(dst_volume)?; if !skip_access_checks(dst_volume) { - check_volume_exists(&dst_volume_path).await?; + if let Err(e) = utils::fs::access(&dst_volume_dir).await { + return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); + } } // xl.meta路径 - let src_file_path = self.get_object_path(src_volume, format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str())?; - let dst_file_path = self.get_object_path(dst_volume, format!("{}/{}", &dst_path, super::STORAGE_FORMAT_FILE).as_str())?; + let src_file_path = src_volume_dir.join(Path::new(format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str())); + let dst_file_path = dst_volume_dir.join(Path::new(format!("{}/{}", &dst_path, super::STORAGE_FORMAT_FILE).as_str())); // data_dir 路径 - let (src_data_path, dst_data_path) = { - let mut data_dir = String::new(); - if !fi.is_remote() { - data_dir = utils::path::retain_slash(fi.data_dir.unwrap_or(Uuid::nil()).to_string().as_str()); - } + let has_data_dir_path = { + let has_data_dir = { + if !fi.is_remote() { + if let Some(dir) = fi.data_dir { + Some(utils::path::retain_slash(dir.to_string().as_str())) + } else { + None + } + } else { + None + } + }; - if !data_dir.is_empty() { - let src_data_path = self.get_object_path( - src_volume, + if let Some(data_dir) = has_data_dir { + let src_data_path = src_volume_dir.join(Path::new( utils::path::retain_slash(format!("{}/{}", &src_path, data_dir).as_str()).as_str(), - )?; - let dst_data_path = self.get_object_path(dst_volume, format!("{}/{}", &dst_path, data_dir).as_str())?; + )); + let dst_data_path = dst_volume_dir.join(Path::new( + utils::path::retain_slash(format!("{}/{}", &dst_path, data_dir).as_str()).as_str(), + )); - (src_data_path, dst_data_path) + Some((src_data_path, dst_data_path)) } else { - (PathBuf::new(), PathBuf::new()) + None } }; - // 读旧xl.meta - let mut meta = FileMeta::new(); + check_path_length(&src_file_path.to_string_lossy().to_string().as_str())?; + check_path_length(&dst_file_path.to_string_lossy().to_string().as_str())?; - let (dst_buf, _) = read_file_exists(&dst_file_path).await?; - if !dst_buf.is_empty() { - // 有旧文件，加载 - match meta.unmarshal_msg(&dst_buf) { - Ok(_) => {} - Err(_) => meta = FileMeta::new(), + // 读旧xl.meta + + let has_dst_buf = match utils::fs::read_file(&dst_file_path).await { + Ok(res) => Some(res), + Err(e) => { + if is_sys_err_not_dir(&e) && !cfg!(target_os = "windows") { + return Err(Error::new(DiskError::FileAccessDenied)); + } + + if !os_is_not_exist(&e) { + return Err(os_err_to_file_err(e)); + } + + None + } + }; + + // let current_data_path = dst_volume_dir.join(Path::new(&dst_path)); + + let mut xlmeta = FileMeta::new(); + + if let Some(dst_buf) = has_dst_buf.as_ref() { + if FileMeta::is_xl_format(&dst_buf) { + if let Ok(nmeta) = FileMeta::load(&dst_buf) { + xlmeta = nmeta + } } } - let mut skip_parent = dst_volume_path.clone(); - if !dst_buf.is_empty() { - skip_parent = PathBuf::from(&dst_file_path.parent().unwrap_or(Path::new("/"))); + let mut skip_parent = dst_volume_dir.clone(); + if has_dst_buf.as_ref().is_some() { + if let Some(parent) = dst_file_path.parent() { + skip_parent = parent.to_path_buf(); + } } - // 查找版本是否已存在 - let old_data_dir = meta - .find_version(fi.version_id) - .map(|(_, version)| { - version - .get_data_dir() - .filter(|data_dir| meta.shard_data_dir_count(&fi.version_id, &Some(data_dir.clone())) == 0) - }) - .unwrap_or_default(); + // TODO: Healing - // 添加版本，写入xl.meta文件 - meta.add_version(fi.clone())?; + let has_old_data_dir = { + if let Ok((_, ver)) = xlmeta.find_version(fi.version_id) { + let has_data_dir = ver.get_data_dir(); + if let Some(data_dir) = has_data_dir { + if xlmeta.shard_data_dir_count(&fi.version_id, &Some(data_dir.clone())) == 0 { + // TODO: Healing + // remove inlinedata\ + Some(data_dir) + } else { + None + } + } else { + None + } + } else { + None + } + }; - let fm_data = meta.marshal_msg()?; + xlmeta.add_version(fi.clone())?; - // 写入xl.meta - self.write_all(src_volume, format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str(), fm_data) - .await?; - - let no_inline = src_data_path.has_root() && fi.data.is_none() && fi.size > 0; - if no_inline { - self.rename_all(&src_data_path, &dst_data_path, &skip_parent).await?; + if xlmeta.versions.len() <= 10 { + // TODO: Sign } - // warn!("old_data_dir {:?}", old_data_dir); - // 有旧目录，把old xl.meta存到旧目录里 - if old_data_dir.is_some() { - self.write_all( - &dst_volume, - format!("{}/{}/{}", &dst_path, &old_data_dir.unwrap().to_string(), super::STORAGE_FORMAT_FILE).as_str(), - dst_buf, - ) - .await?; + let new_dst_buf = xlmeta.marshal_msg()?; + + self.write_all(&src_volume, format!("{}/{}", &src_path, super::STORAGE_FORMAT_FILE).as_str(), new_dst_buf) + .await + .map_err(|err| { + if let Some(e) = err.to_io_err() { + os_err_to_file_err(e) + } else { + err + } + })?; + + if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() { + let no_inline = fi.data.is_none() && fi.size > 0; + if no_inline { + if let Err(err) = os::rename_all(&src_data_path, &dst_data_path, &skip_parent).await { + let _ = self.delete_file(&dst_volume_dir, &dst_data_path, false, false).await; + + return Err({ + if let Some(e) = err.to_io_err() { + os_err_to_file_err(e) + } else { + err + } + }); + } + } } - if let Err(e) = self.rename_all(&src_file_path, &dst_file_path, &skip_parent).await { - // 如果 失败删除目标目录 - let _ = self.delete_file(&dst_volume_path, &dst_data_path, false, false).await; - return Err(e); + if let Some(old_data_dir) = has_old_data_dir { + // preserve current xl.meta inside the oldDataDir. + if let Some(dst_buf) = has_dst_buf { + if let Err(err) = self + .write_all_private( + &dst_volume, + format!("{}/{}/{}", &dst_path, &old_data_dir.to_string(), super::STORAGE_FORMAT_FILE).as_str(), + &dst_buf, + true, + &skip_parent, + ) + .await + { + return Err({ + if let Some(e) = err.to_io_err() { + os_err_to_file_err(e) + } else { + err + } + }); + } + } } - if src_volume != super::RUSTFS_META_MULTIPART_BUCKET { - fs::remove_dir(&src_file_path.parent().unwrap()).await?; - } else { - self.delete_file(&src_volume_path, &PathBuf::from(src_file_path.parent().unwrap()), true, false) - .await?; + if let Err(err) = os::rename_all(&src_file_path, &dst_file_path, &skip_parent).await { + if let Some((_, dst_data_path)) = has_data_dir_path.as_ref() { + let _ = self.delete_file(&dst_volume_dir, &dst_data_path, false, false).await; + } + return Err({ + if let Some(e) = err.to_io_err() { + os_err_to_file_err(e) + } else { + err + } + }); + } + + if let Some(src_file_path_parent) = src_file_path.parent() { + if src_volume != super::RUSTFS_META_MULTIPART_BUCKET { + let _ = utils::fs::remove(src_file_path_parent).await; + } else { + let _ = self + .delete_file(&dst_volume_dir, &src_file_path_parent.to_path_buf(), true, false) + .await; + } } Ok(RenameDataResp { - old_data_dir: old_data_dir, + old_data_dir: has_old_data_dir, sign: None, // TODO: }) } @@ -1189,12 +1317,19 @@ impl DiskAPI for LocalDisk { return Err(Error::msg("Invalid arguments specified")); } - let p = self.get_bucket_path(volume)?; + let volume_dir = self.get_bucket_path(volume)?; - if let Err(e) = utils::fs::access(&p).await { + if let Err(e) = utils::fs::access(&volume_dir).await { if os_is_not_exist(&e) { - os::make_dir_all(&p, self.root.as_path()).await?; + os::make_dir_all(&volume_dir, self.root.as_path()).await?; } + if os_is_permission(&e) { + return Err(Error::new(DiskError::DiskAccessDenied)); + } else if is_sys_err_io(&e) { + return Err(Error::new(DiskError::FaultyDisk)); + } + + return Err(Error::new(e)); } Err(Error::from(DiskError::VolumeExists)) @@ -1202,7 +1337,7 @@ impl DiskAPI for LocalDisk { async fn list_volumes(&self) -> Result> { let mut volumes = Vec::new(); - let entries = os::read_dir(&self.root, 0).await.map_err(|e| { + let entries = os::read_dir(&self.root, -1).await.map_err(|e| { if DiskError::FileAccessDenied.is(&e) { Error::new(DiskError::DiskAccessDenied) } else if DiskError::FileNotFound.is(&e) { @@ -1226,17 +1361,27 @@ impl DiskAPI for LocalDisk { Ok(volumes) } async fn stat_volume(&self, volume: &str) -> Result { - let p = self.get_bucket_path(volume)?; - - let m = read_file_metadata(&p).await?; - let modtime = match m.modified() { - Ok(md) => Some(OffsetDateTime::from(md)), - Err(_) => { - warn!("Not supported modified on this platform"); - None + let volume_dir = self.get_bucket_path(volume)?; + let meta = match utils::fs::lstat(&volume_dir).await { + Ok(res) => res, + Err(e) => { + if os_is_not_exist(&e) { + return Err(Error::new(DiskError::VolumeNotFound)); + } else if os_is_permission(&e) { + return Err(Error::new(DiskError::DiskAccessDenied)); + } else if is_sys_err_io(&e) { + return Err(Error::new(DiskError::FaultyDisk)); + } else { + return Err(Error::new(e)); + } } }; + let modtime = match meta.modified() { + Ok(md) => Some(OffsetDateTime::from(md)), + Err(_) => None, + }; + Ok(VolumeInfo { name: volume.to_string(), created: modtime, diff --git a/ecstore/src/disk/os.rs b/ecstore/src/disk/os.rs index 78d0ada5c..cb118d6aa 100644 --- a/ecstore/src/disk/os.rs +++ b/ecstore/src/disk/os.rs @@ -68,18 +68,12 @@ pub async fn make_dir_all(path: impl AsRef, base_dir: impl AsRef) -> } // read_dir count read limit. when count == 0 unlimit. -pub async fn read_dir(path: impl AsRef, count: usize) -> Result> { +pub async fn read_dir(path: impl AsRef, count: i32) -> Result> { let mut entries = fs::read_dir(path.as_ref()).await?; let mut volumes = Vec::new(); - let mut count: i32 = { - if count == 0 { - -1 - } else { - count as i32 - } - }; + let mut count = count; while let Some(entry) = entries.next_entry().await? { let name = entry.file_name().to_string_lossy().to_string(); @@ -208,12 +202,3 @@ pub async fn os_mkdir_all(dir_path: impl AsRef, base_dir: impl AsRef Ok(()) } - -#[cfg(test)] -mod test { - - use super::*; - - #[tokio::test] - async fn test_make_dir() {} -} diff --git a/ecstore/src/error.rs b/ecstore/src/error.rs index 1fc8dfabf..db5b27561 100644 --- a/ecstore/src/error.rs +++ b/ecstore/src/error.rs @@ -98,7 +98,11 @@ impl Clone for Error { if let Some(e) = self.downcast_ref::() { clone_disk_err(e) } else if let Some(e) = self.downcast_ref::() { - Error::new(io::Error::new(e.kind(), e.to_string())) + if let Some(code) = e.raw_os_error() { + Error::new(io::Error::from_raw_os_error(code)) + } else { + Error::new(io::Error::new(e.kind(), e.to_string())) + } } else { // TODO: 优化其他类型 Error::msg(self.to_string()) diff --git a/ecstore/src/utils/fs.rs b/ecstore/src/utils/fs.rs index afc13c5a3..a3012ea28 100644 --- a/ecstore/src/utils/fs.rs +++ b/ecstore/src/utils/fs.rs @@ -130,3 +130,7 @@ pub async fn mkdir(path: impl AsRef) -> io::Result<()> { pub async fn rename(from: impl AsRef, to: impl AsRef) -> io::Result<()> { fs::rename(from, to).await } + +pub async fn read_file(path: impl AsRef) -> io::Result> { + fs::read(path.as_ref()).await +}