wip-remote

Signed-off-by: junxiang Mu <1948535941@qq.com>

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,8 +12,8 @@ use self::flatbuffers::{EndianScalar, Follow};
 #[allow(unused_imports, dead_code)]
 pub mod models {
 
-    use core::cmp::Ordering;
   use core::mem;
+  use core::cmp::Ordering;
 
   extern crate flatbuffers;
   use self::flatbuffers::{EndianScalar, Follow};
@@ -28,9 +29,7 @@ pub mod models {
   type Inner = PingBody<'a>;
   #[inline]
   unsafe fn follow(buf: &'a [u8], loc: usize) -> Self::Inner {
-            Self {
+    Self { _tab: flatbuffers::Table::new(buf, loc) }
-                _tab: flatbuffers::Table::new(buf, loc),
-            }
   }
 }
 
@@ -48,30 +47,28 @@ pub mod models {
   #[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>,
+    args: &'args PingBodyArgs<'args>
   ) -> flatbuffers::WIPOffset<PingBody<'bldr>> {
     let mut builder = PingBodyBuilder::new(_fbb);
-            if let Some(x) = args.payload {
+    if let Some(x) = args.payload { builder.add_payload(x); }
-                builder.add_payload(x);
-            }
     builder.finish()
   }
 
+
   #[inline]
   pub fn payload(&self) -> Option<flatbuffers::Vector<'a, u8>> {
     // Safety:
     // Created from valid Table for this object
     // which contains a valid value in this slot
-            unsafe {
+    unsafe { self._tab.get::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'a, u8>>>(PingBody::VT_PAYLOAD, None)}
-                self._tab
-                    .get::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'a, u8>>>(PingBody::VT_PAYLOAD, None)
-            }
   }
 }
 
 impl flatbuffers::Verifiable for PingBody<'_> {
   #[inline]
-        fn run_verifier(v: &mut flatbuffers::Verifier, pos: usize) -> Result<(), flatbuffers::InvalidFlatbuffer> {
+  fn run_verifier(
+    v: &mut flatbuffers::Verifier, pos: usize
+  ) -> Result<(), flatbuffers::InvalidFlatbuffer> {
     use self::flatbuffers::Verifiable;
     v.visit_table(pos)?
      .visit_field::<flatbuffers::ForwardsUOffset<flatbuffers::Vector<'_, u8>>>("payload", Self::VT_PAYLOAD, false)?
@@ -85,7 +82,9 @@ pub mod models {
 impl<'a> Default for PingBodyArgs<'a> {
   #[inline]
   fn default() -> Self {
-            PingBodyArgs { payload: None }
+    PingBodyArgs {
+      payload: None,
+    }
   }
 }
 
@@ -96,8 +95,7 @@ pub mod models {
 impl<'a: 'b, 'b, A: flatbuffers::Allocator + 'a> PingBodyBuilder<'a, 'b, A> {
   #[inline]
   pub fn add_payload(&mut self, payload: flatbuffers::WIPOffset<flatbuffers::Vector<'b , u8>>) {
-            self.fbb_
+    self.fbb_.push_slot_always::<flatbuffers::WIPOffset<_>>(PingBody::VT_PAYLOAD, payload);
-                .push_slot_always::<flatbuffers::WIPOffset<_>>(PingBody::VT_PAYLOAD, payload);
   }
   #[inline]
   pub fn new(_fbb: &'b mut flatbuffers::FlatBufferBuilder<'a, A>) -> PingBodyBuilder<'a, 'b, A> {
@@ -122,3 +120,4 @@ pub mod models {
   }
 }
 }  // pub mod models
+

ecstore/src/disk/remote.rs

@@ -1,4 +1,4 @@
-use std::{path::PathBuf, time::Duration};
+use std::{path::PathBuf, sync::RwLock, time::Duration};
 
 use bytes::Bytes;
 use protos::{
@@ -16,6 +16,7 @@ use tonic::{
     Request,
 };
 use tower::timeout::Timeout;
+use tracing::info;
 use uuid::Uuid;
 
 use crate::{
@@ -30,29 +31,51 @@ use super::{
 
 #[derive(Debug)]
 pub struct RemoteDisk {
-    channel: Channel,
+    channel: RwLock<Option<Channel>>,
+    url: url::Url,
     pub root: PathBuf,
 }
 
 impl RemoteDisk {
     pub async fn new(ep: &Endpoint, _opt: &DiskOption) -> Result<Self> {
-        let connector = tonic_Endpoint::from_shared(format!(
-            "{}://{}{}",
-            ep.url.scheme(),
-            ep.url.host_str().unwrap(),
-            ep.url.port().unwrap()
-        ))
-        .unwrap();
-        let channel = tokio::runtime::Runtime::new().unwrap().block_on(connector.connect()).unwrap();
-
         let root = fs::canonicalize(ep.url.path()).await?;
 
-        Ok(Self { channel, root })
+        Ok(Self { 
+            channel: RwLock::new(None),
+            url: ep.url.clone(),
+            root
+        })
     }
 
     fn get_client(&self) -> NodeServiceClient<Timeout<Channel>> {
+        let channel = {
+            let read_lock = self.channel.read().unwrap();
+            
+            if let Some(ref channel) = *read_lock {
+                channel.clone()
+            } else {
+                let addr = format!(
+                    "{}://{}:{}",
+                    self.url.scheme(),
+                    self.url.host_str().unwrap(),
+                    self.url.port().unwrap()
+                );
+                info!("disk url: {:?}", addr);
+                let connector = tonic_Endpoint::from_shared(addr).unwrap();
+                
+                let new_channel = tokio::runtime::Runtime::new()
+                    .unwrap()
+                    .block_on(connector.connect())
+                    .unwrap();
+                
+                *self.channel.write().unwrap() = Some(new_channel.clone());
+                
+                new_channel
+            }
+        };
+
         node_service_time_out_client(
-            self.channel.clone(),
+            channel,
             Duration::new(30, 0), // TODO: use config setting
             DEFAULT_GRPC_SERVER_MESSAGE_LEN,
             // grpc_enable_gzip,