init

Cargo.toml

@@ -10,9 +10,16 @@ name = "server"
 path = "src/server.rs"
 
 [[bin]] # Bin to run the HelloWorld gRPC client
-name = "client"
-path = "src/client.rs"
+name = "client-push"
+path = "src/client-push.rs"
 
+# [[bin]] # Bin to run the HelloWorld gRPC client
+# name = "client-pull"
+# path = "src/client-pull.rs"
+
+[[bin]] # Bin to run the HelloWorld gRPC client
+name = "test"
+path = "src/test.rs"
 
 [dependencies]
 tonic = "0.7"
@@ -20,9 +27,13 @@ prost = "0.10"
 tokio = { version = "1.0", features = [ "rt-multi-thread", "time", "fs", "macros", "net",] }
 tokio-stream = { version = "0.1", features = ["net"] }
 
-futures = { version = "0.3", default-features = false, features = ["alloc"] }
+# futures = { version = "0.3", default-features = false, features = ["alloc"] }
+futures = { version = "0.3" }
+
 
 h2 = "0.3"
 
 [build-dependencies]
-tonic-build = "0.7"
\ No newline at end of file
+tonic-build = "0.7"
+futures = "0.3"
+futures-util = "0.3"
\ No newline at end of file

proto/MetadataOps.proto

@@ -18,29 +18,40 @@
      int32 sendtype = 1;        
      //Tenent及用户各使用16bits标识码，称为tcode;
      //在rocksdb里面每个key都会该信息，可以把这部分信息抽离出来，再在数据接收端进行key值的拼接，以减少网络层数据传输的压力
-     string t_code = 2;  
+     string tcode = 2;  
  }
  message PullCreateResp {
-     string t_code = 1;
-    //  repeated Metadata md = 2; 
+     string tcode = 1;
+     repeated Metadata md = 2; 
      // 0:成功;  -1: 失败; 其他：未定义行为
      int32 ret_code = 3;
  }
  
-//  //client向server推送数据请求创建文件或目录
-//  message PushCreateReq {
-//      //0: 客户端向服务端发送请求; 1: 服务端向客户端发送请求; 其他：未定义行为
-//      int32 type = 1;        
-//      //Tenent及用户各使用16bits标识码，称为tcode;
-//      string tcode = 2;  
-//      repeated Metadata metadata = 3; 
-//  }
-//  message PushCreateResp {
-//      string tcode = 1;
-//      //数据推送是否更新成功
-//      int32 retCode = 2;
-//  }
+ //client向server推送数据请求创建文件或目录
+ message PushCreateReq {
+     //0: 客户端向服务端发送请求; 1: 服务端向客户端发送请求; 其他：未定义行为
+     int32 sendtype = 1;        
+     //Tenent及用户各使用16bits标识码，称为tcode;
+     string tcode = 2;  
+     repeated Metadata metadata = 3; 
+ }
+ message PushCreateResp {
+     string tcode = 1;
+     //数据推送是否更新成功
+     int32 retCode = 2;
+ }
  
+message WatchUpdateReq{
+    int32 clientId = 1;
+    string tcode = 2;
+}
+
+message WatchUpdateResp{
+    int32 clientId = 1;
+    string notifyMsg = 2;
+}
+
+
 //  //文件路径,对应rocksdb的key值
 //  message AbsPath{
 //      string k = 1;
@@ -54,7 +65,7 @@
 //      string tcode = 1;
 //      int32 retCode = 2;
 //  }
-//  //更新
+ //更新
 //  message UpdateReq{
 //      string tcode = 1;
 //      repeated Metadata md = 2;
@@ -75,7 +86,8 @@
  
  service MetadataOps {
      rpc CreateByPull(stream PullCreateReq) returns (stream PullCreateResp) {}
-    //  rpc CreateByPush(stream PushCreateReq) returns (stream PushCreateResp) {}
+     rpc CreateByPush(stream PushCreateReq) returns (stream PushCreateResp) {}
+     rpc WatchUpdate(stream WatchUpdateReq) returns (stream WatchUpdateResp){}
     //  rpc Remove(stream RemoveReq) returns(stream RemoveResp){}
     //  rpc Update(stream UpdateReq) returns(stream UpdateResp){}
     //  rpc QueryDirdents(stream QueryDirdentsReq) returns(stream QueryDirdentsResp){}

tmp/test/src/client.rs → src/client-pull.rs

@@ -5,34 +5,66 @@ pub mod registry {
 use futures::stream::Stream;
 use std::time::Duration;
 use tokio_stream::StreamExt;
-use tonic::{server::ClientStreamingService, transport::Channel};
+use tonic::{codegen::http::response, server::ClientStreamingService, transport::Channel};
 
-use registry::{metadata_ops_client::MetadataOpsClient, PullCreateReq};
+use registry::{metadata_ops_client::MetadataOpsClient, Metadata, PullCreateReq};
 
 fn generate_request() -> impl Stream<Item = PullCreateReq> {
     tokio_stream::iter(1..3).map(|i| PullCreateReq {
-        t_code: format!("t_code:{:02}", i),
+        tcode: format!("tcode:{:02}", i),
         sendtype: i as i32,
     })
 }
 
-async fn create_by_pull(client: &mut MetadataOpsClient<Channel>, dur: Duration) {
+fn my_request() -> impl Stream<Item = PushCreateReq> {
+    let rdata = fs::read("./test.txt");
+    // String::from_utf8_lossy(&fs::read("address.txt").unwrap());
+    let mut Send_metadata = vec![];
+    match rdata {
+        Ok(data) => {
+            let c_data = std::str::from_utf8(&data).unwrap();
+            // println!("data = {}", c_data.clone());
+            let d = Metadata {
+                kv: c_data.to_string(),
+            };
+            Send_metadata.push(d);
+            return tokio_stream::iter(1..2).map(move |i| PushCreateReq {
+                tcode: "my_tcode".to_string(),
+                sendtype: 1,
+                metadata: Send_metadata.clone(),
+            });
+        }
+        Err(e) => {
+            panic!("dfdf");
+            // println!("d={:?}", e);
+        }
+    }
+}
+
+fn pull_request() -> impl Stream<Item = PullCreateReq> {
+    tokio_stream::iter(1..2).map(|i| PullCreateReq {
+        tcode: format!("tcode:{:02}", i),
+        sendtype: i as i32,
+    })
+}
+
+async fn create_by_pull(client: &mut MetadataOpsClient<Channel>) {
     // let in_stream = generate_request().throttle(dur);
-    let in_stream = generate_request();
+    let in_stream = pull_request();
     let respone = client.create_by_pull(in_stream).await.unwrap();
     let mut resp_stream = respone.into_inner();
     while let Some(received) = resp_stream.next().await {
         let received = received.unwrap();
-        println!("received :{}", received.t_code);
+        println!("received :{:?}", received.md[0].kv);
     }
 
-    let in_stream_2 = generate_request();
-    let respone = client.create_by_pull(in_stream_2).await.unwrap();
-    let mut resp_stream = respone.into_inner();
-    while let Some(received) = resp_stream.next().await {
-        let received = received.unwrap();
-        println!("received :{}", received.t_code);
-    }
+    // let in_stream_2 = generate_request();
+    // let respone = client.create_by_pull(in_stream_2).await.unwrap();
+    // let mut resp_stream = respone.into_inner();
+    // while let Some(received) = resp_stream.next().await {
+    //     let received = received.unwrap();
+    //     println!("received :{}", received.tcode);
+    // }
 }
 
 use std::fs;
@@ -40,29 +72,36 @@ use std::fs;
 use crate::registry::PushCreateReq;
 
 async fn create_by_push(client: &mut MetadataOpsClient<Channel>) {
-    let rdata = fs::read("./test.txt");
-    // String::from_utf8_lossy(&fs::read("address.txt").unwrap());
-    match rdata {
-        Ok(data) => {
-            println!("data = {}", std::str::from_utf8(&data).unwrap());
-            let req_data = PushCreateReq {
-                sendtype: 1,
-                tcode: "hello".to_string(),
-                metadata: Vec::new(),
-            };
-            // let messages = vec![req_data.clone(), req_data];
-            let b = tokio_stream::iter(1..3).map(|i| PullCreateReq {
-                t_code: format!("t_code:{:02}", i),
-                sendtype: i as i32,
-            });
-            // client.record_route(Request::new(stream::iter(messages.clone())));
-            let respone = client
-                .create_by_push(b as impl Stream<Item = PullCreateReq>)
-                .await
-                .unwrap();
-        }
-        Err(e) => println!("d={:?}", e),
+    // let rdata = fs::read("./test.txt");
+    // // String::from_utf8_lossy(&fs::read("address.txt").unwrap());
+    // match rdata {
+    //     Ok(data) => {
+    //         println!("data = {}", std::str::from_utf8(&data).unwrap());
+
+    let in_stream = my_request();
+    let respone = client.create_by_push(in_stream).await.unwrap();
+    let mut resp_stream = respone.into_inner();
+    while let Some(received) = resp_stream.next().await {
+        let received = received.unwrap();
+        println!("received: {}", received.tcode);
     }
+    // let req_data = PushCreateReq {
+    //     sendtype: 1,
+    //     tcode: "hello".to_string(),
+    //     metadata: Vec::new(),
+    // };
+    // let b = tokio_stream::iter(1..3).map(|i| PullCreateReq {
+    //     tcode: format!("tcode:{:02}", i),
+    //     sendtype: i as i32,
+    // });
+    // // client.record_route(Request::new(stream::iter(messages.clone())));
+    // let respone = client
+    //     .create_by_push(b as impl Stream<Item = PullCreateReq>)
+    //     .await
+    //     .unwrap();
+    //     }
+    //     Err(e) => println!("d={:?}", e),
+    // }
     // let respone = client.create_by_push()
 }
 
@@ -73,7 +112,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .await
         .unwrap();
     println!("start bidirectional stream echo ");
-    // create_by_pull(&mut client, Duration::from_micros(1)).await;
-    create_by_push(&mut client).await;
+    create_by_pull(&mut client).await;
+    // create_by_push(&mut client).await;
     Ok(())
 }

src/client-push.rs

+pub mod registry {
+    tonic::include_proto!("bky.registry.api");
+}
+
+use futures::{executor::block_on, stream::Stream};
+use registry::{metadata_ops_client::MetadataOpsClient, Metadata, PullCreateReq, WatchUpdateReq};
+use std::time::Duration;
+use tokio::runtime::Builder;
+use tokio::{self, runtime::Runtime, task, time};
+use tokio_stream::StreamExt;
+use tonic::{codegen::http::response, server::ClientStreamingService, transport::Channel};
+fn generate_request() -> impl Stream<Item = PullCreateReq> {
+    tokio_stream::iter(1..3).map(|i| PullCreateReq {
+        tcode: format!("tcode:{:02}", i),
+        sendtype: i as i32,
+    })
+}
+
+fn my_request() -> impl Stream<Item = PushCreateReq> {
+    let rdata = fs::read("./test.txt");
+    // String::from_utf8_lossy(&fs::read("address.txt").unwrap());
+    let mut Send_metadata = vec![];
+    match rdata {
+        Ok(data) => {
+            let c_data = std::str::from_utf8(&data).unwrap();
+            // println!("data = {}", c_data.clone());
+            let d = Metadata {
+                kv: c_data.to_string(),
+            };
+            Send_metadata.push(d);
+            return tokio_stream::iter(1..2).map(move |i| PushCreateReq {
+                tcode: "my_tcode".to_string(),
+                sendtype: 1,
+                metadata: Send_metadata.clone(),
+            });
+        }
+        Err(e) => {
+            println!("d={:?}", e);
+            panic!("dfdf");
+        }
+    }
+}
+
+fn watch_update_request() -> impl Stream<Item = WatchUpdateReq> {
+    tokio_stream::iter(1..2).map(move |i| WatchUpdateReq {
+        tcode: "my_tcode".to_string(),
+        client_id: 2,
+    })
+}
+
+async fn watch_update(client: &mut MetadataOpsClient<Channel>) {
+    let in_stream = watch_update_request();
+    let respone = client.watch_update(in_stream).await.unwrap();
+    let mut resp_stream = respone.into_inner();
+    while let Some(received) = resp_stream.next().await {
+        let received = received.unwrap();
+        println!("received:{:?}", received);
+    }
+}
+
+async fn create_by_pull(client: &mut MetadataOpsClient<Channel>) {
+    // let in_stream = generate_request().throttle(dur);
+    let in_stream = generate_request();
+    let respone = client.create_by_pull(in_stream).await.unwrap();
+    let mut resp_stream = respone.into_inner();
+    while let Some(received) = resp_stream.next().await {
+        let received = received.unwrap();
+        println!("received :{}", received.tcode);
+    }
+
+    let in_stream_2 = generate_request();
+    let respone = client.create_by_pull(in_stream_2).await.unwrap();
+    let mut resp_stream = respone.into_inner();
+    while let Some(received) = resp_stream.next().await {
+        let received = received.unwrap();
+        println!("received :{}", received.tcode);
+    }
+}
+
+use std::fs;
+
+use crate::registry::PushCreateReq;
+
+async fn create_by_push(client: &mut MetadataOpsClient<Channel>) {
+    // let rdata = fs::read("./test.txt");
+    // // String::from_utf8_lossy(&fs::read("address.txt").unwrap());
+    // match rdata {
+    //     Ok(data) => {
+    //         println!("data = {}", std::str::from_utf8(&data).unwrap());
+
+    let in_stream = my_request();
+    let respone = client.create_by_push(in_stream).await.unwrap();
+    let mut resp_stream = respone.into_inner();
+    while let Some(received) = resp_stream.next().await {
+        let received = received.unwrap();
+        println!("received: {:?}", received);
+    }
+    // let req_data = PushCreateReq {
+    //     sendtype: 1,
+    //     tcode: "hello".to_string(),
+    //     metadata: Vec::new(),
+    // };
+    // let b = tokio_stream::iter(1..3).map(|i| PullCreateReq {
+    //     tcode: format!("tcode:{:02}", i),
+    //     sendtype: i as i32,
+    // });
+    // // client.record_route(Request::new(stream::iter(messages.clone())));
+    // let respone = client
+    //     .create_by_push(b as impl Stream<Item = PullCreateReq>)
+    //     .await
+    //     .unwrap();
+    //     }
+    //     Err(e) => println!("d={:?}", e),
+    // }
+    // let respone = client.create_by_push()
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let rt = Runtime::new().unwrap();
+    let guard1 = rt.enter();
+
+    task::spawn(async {
+        let mut client2 = MetadataOpsClient::connect("http://192.168.110.183:50050")
+            .await
+            .unwrap();
+        create_by_push(&mut client2).await;
+        println!("create_by_push done");
+    });
+    drop(guard1);
+
+    loop {
+        let guard2 = rt.enter();
+        task::spawn(async {
+            let mut client = MetadataOpsClient::connect("http://192.168.110.183:50050")
+                .await
+                .unwrap();
+            // watch_update(&mut client).await;
+            let in_stream = watch_update_request();
+            let respone = client.watch_update(in_stream).await.unwrap();
+            let mut resp_stream = respone.into_inner();
+            while let Some(received) = resp_stream.next().await {
+                let received = received.unwrap();
+                println!("received:{:?}", received);
+            }
+            println!("watch_update done");
+        });
+        drop(guard2);
+    }
+    // Ok(())
+}

src/client.rs

-pub mod registry {
-    tonic::include_proto!("bky.registry.api");
-}
-
-use futures::stream::Stream;
-use std::time::Duration;
-use tokio_stream::StreamExt;
-use tonic::{server::ClientStreamingService, transport::Channel};
-
-use registry::{metadata_ops_client::MetadataOpsClient, PullCreateReq};
-
-fn generate_request() -> impl Stream<Item = PullCreateReq> {
-    tokio_stream::iter(1..u8::MAX).map(|i| PullCreateReq {
-        t_code: format!("t_code:{:02}", i),
-        sendtype: i as i32,
-    })
-}
-
-async fn create_by_pull(client: &mut MetadataOpsClient<Channel>, dur: Duration) {
-    // let in_stream = generate_request().throttle(dur);
-    let in_stream = generate_request();
-    let respone = client.create_by_pull(in_stream).await.unwrap();
-    let mut resp_stream = respone.into_inner();
-    while let Some(received) = resp_stream.next().await {
-        let received = received.unwrap();
-        println!("received :{}", received.t_code);
-    }
-
-    let in_stream_2 = generate_request();
-    let respone = client.create_by_pull(in_stream_2).await.unwrap();
-    let mut resp_stream = respone.into_inner();
-    while let Some(received) = resp_stream.next().await {
-        let received = received.unwrap();
-        println!("received :{}", received.t_code);
-    }
-}
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let mut client = MetadataOpsClient::connect("http://192.168.110.183:50050")
-        // let mut client = MetadataOpsClient::connect("http://192.168.110.183:50050")
-        .await
-        .unwrap();
-    println!("start bidirectional stream echo ");
-    create_by_pull(&mut client, Duration::from_micros(1)).await;
-    Ok(())
-}

src/server.rs

@@ -3,8 +3,14 @@ pub mod registry {
 }
 
 use futures::Stream;
-use registry::{PullCreateReq, PullCreateResp};
-use std::{error::Error, io::ErrorKind, net::ToSocketAddrs, pin::Pin, result, time::Duration};
+use registry::{
+    Metadata, PullCreateReq, PullCreateResp, PushCreateReq, PushCreateResp, WatchUpdateReq,
+    WatchUpdateResp,
+};
+use std::{
+    error::Error, io::ErrorKind, net::ToSocketAddrs, pin::Pin, result, thread::sleep,
+    time::Duration,
+};
 use tokio::sync::mpsc;
 use tokio_stream::{wrappers::ReceiverStream, StreamExt};
 use tonic::{transport::Server, Request, Response, Status, Streaming};
@@ -15,6 +21,7 @@ pub struct RegistryServer {}
 
 type RegistryResult<T> = Result<Response<T>, Status>;
 type ResponseStream = Pin<Box<dyn Stream<Item = Result<PullCreateResp, Status>> + Send>>;
+
 fn match_for_io_error(err_status: &Status) -> Option<&std::io::Error> {
     let mut err: &(dyn Error + 'static) = err_status;
 
@@ -37,6 +44,8 @@ fn match_for_io_error(err_status: &Status) -> Option<&std::io::Error> {
         };
     }
 }
+use std::fs;
+use std::{thread, time};
 #[tonic::async_trait]
 impl registry::metadata_ops_server::MetadataOps for RegistryServer {
     type CreateByPullStream = ResponseStream;
@@ -54,13 +63,135 @@ impl registry::metadata_ops_server::MetadataOps for RegistryServer {
             while let Some(result) = in_stream.next().await {
                 println!("result = {:?}", result);
                 match result {
-                    Ok(v) => tx
-                        .send(Ok(PullCreateResp {
-                            t_code: v.t_code,
+                    Ok(v) => {
+                        let rdata = fs::read("./test_rcv.txt");
+                        let mut Send_metadata = vec![];
+                        match rdata {
+                            Ok(data) => {
+                                let c_data = std::str::from_utf8(&data).unwrap();
+                                // println!("data = {}", c_data.clone());
+                                let d = Metadata {
+                                    kv: c_data.to_string(),
+                                };
+                                Send_metadata.push(d);
+                            }
+                            Err(e) => {
+                                panic!("dfdf");
+                                // println!("d={:?}", e);
+                            }
+                        }
+
+                        tx.send(Ok(PullCreateResp {
+                            tcode: v.tcode,
+                            ret_code: 0,
+                            md: Send_metadata,
+                        }))
+                        .await
+                        .expect("working rx");
+                    }
+                    Err(err) => {
+                        if let Some(io_err) = match_for_io_error(&err) {
+                            if io_err.kind() == ErrorKind::BrokenPipe {
+                                eprintln!("\tclient disconnected: broken pipe");
+                                break;
+                            }
+                        }
+                        match tx.send(Err(err)).await {
+                            Ok(_) => (),
+                            Err(_err) => break,
+                        }
+                    }
+                }
+                println!("\tstream ended");
+            }
+        });
+        let out_stream = ReceiverStream::new(rx);
+        Ok(Response::new(Box::pin(out_stream)))
+    }
+
+    //======================================
+    type CreateByPushStream = Pin<Box<dyn Stream<Item = Result<PushCreateResp, Status>> + Send>>;
+    async fn create_by_push(
+        &self,
+        req: Request<Streaming<PushCreateReq>>,
+    ) -> RegistryResult<Self::CreateByPushStream> {
+        println!("create_by_push ");
+        println!("req={:?}", req);
+        let mut in_stream = req.into_inner();
+        println!("in_stream={:?}", in_stream);
+        let (tx, rx) = mpsc::channel(128);
+        tokio::spawn(async move {
+            println!("tokio spawn..");
+            while let Some(result) = in_stream.next().await {
+                // println!("result = {:?}", result);
+                match result {
+                    Ok(v) => {
+                        let mut r_metadata = &v.clone().metadata[0].kv;
+                        // println!("tx={:?}", r_metadata);
+                        let wdata_ret = fs::write("./test_rcv.txt", r_metadata);
+                        println!("wdata_ret = {:?}", wdata_ret);
+                        tx.send(Ok(PushCreateResp {
+                            tcode: v.tcode,
                             ret_code: 0,
                         }))
                         .await
-                        .expect("working rx"),
+                        .expect("working rx");
+                    }
+                    Err(err) => {
+                        if let Some(io_err) = match_for_io_error(&err) {
+                            if io_err.kind() == ErrorKind::BrokenPipe {
+                                eprintln!("\tclient disconnected: broken pipe");
+                                break;
+                            }
+                        }
+                        match tx.send(Err(err)).await {
+                            Ok(_) => (),
+                            Err(_err) => break,
+                        }
+                    }
+                }
+                println!("\tstream ended");
+            }
+        });
+        let out_stream = ReceiverStream::new(rx);
+        Ok(Response::new(Box::pin(out_stream)))
+    }
+    //======================================
+    type WatchUpdateStream = Pin<Box<dyn Stream<Item = Result<WatchUpdateResp, Status>> + Send>>;
+    async fn watch_update(
+        &self,
+        req: Request<Streaming<WatchUpdateReq>>,
+    ) -> RegistryResult<Self::WatchUpdateStream> {
+        println!("watch_update ");
+        println!("req={:?}", req);
+        let mut in_stream = req.into_inner();
+        println!("in_stream={:?}", in_stream);
+        let (tx, rx) = mpsc::channel(128);
+        tokio::spawn(async move {
+            println!("tokio spawn..");
+            while let Some(result) = in_stream.next().await {
+                // println!("result = {:?}", result);
+                match result {
+                    Ok(v) => {
+                        println!("watch_update sleep");
+                        // let ten_millis = time::Duration::from_millis(3000);
+                        // thread::sleep(ten_millis);
+                        // let sleep_time = ;
+                        let sleep_op = tokio::time::sleep(tokio::time::Duration::from_secs(4));
+                        println!("watch_update() ...");
+                        loop {
+                            sleep_op.await;
+                            if true {
+                                tx.send(Ok(WatchUpdateResp {
+                                    client_id: v.client_id,
+                                    notify_msg: "hello".to_string(),
+                                }))
+                                .await
+                                .expect("working rx");
+                            }
+                            break;
+                        }
+                    }
                     Err(err) => {
                         if let Some(io_err) = match_for_io_error(&err) {
                             if io_err.kind() == ErrorKind::BrokenPipe {

test_rcv.txt

+"\nfe\nfe\n":"33206|0|1000|1000|1653640181|1653640181|1653640181|795538473|795538473|795538473|1|0"|"/good""16895|0|1000|1000|1653640372|1653640372|1653640372|154554408|154554408|154554408|2|0"    
\ No newline at end of file

tmp/bi/Cargo.toml

-[package]
-name = "bi"
-version = "0.1.0"
-edition = "2021"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[[bin]] # Bin to run the HelloWorld gRPC server
-name = "helloworld-server"
-path = "src/server.rs"
-
-[[bin]] # Bin to run the HelloWorld gRPC client
-name = "helloworld-client"
-path = "src/client.rs"
-
-
-[dependencies]
-tonic = "0.7"
-prost = "0.10"
-tokio = { version = "1.0", features = [ "rt-multi-thread", "time", "fs", "macros", "net",] }
-tokio-stream = { version = "0.1", features = ["net"] }
-
-futures = { version = "0.3", default-features = false, features = ["alloc"] }
-
-h2 = "0.3"
-
-[build-dependencies]
-tonic-build = "0.7"
\ No newline at end of file

tmp/bi/build.rs

-fn main() -> Result<(), Box<dyn std::error::Error>> {
-    tonic_build::compile_protos("proto/echo.proto")?;
-    Ok(())
-}

tmp/bi/proto/echo.proto

-/*
- *
- * Copyright 2018 gRPC authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
- syntax = "proto3";
-
- package grpc.examples.echo;
-
- // EchoRequest is the request for echo.
- message EchoRequest {
-   string message = 1;
-   string req = 2;
- }
-
- // EchoResponse is the response for echo.
- message EchoResponse {
-   string message = 1;
-   string resp = 2;
- }
-
- // Echo is the echo service.
- service Echo {
-  //  // UnaryEcho is unary echo.
-  //  rpc UnaryEcho(EchoRequest) returns (EchoResponse) {}
-  //  // ServerStreamingEcho is server side streaming.
-  //  rpc ServerStreamingEcho(EchoRequest) returns (stream EchoResponse) {}
-  //  // ClientStreamingEcho is client side streaming.
-  //  rpc ClientStreamingEcho(stream EchoRequest) returns (EchoResponse) {}
-  //  // BidirectionalStreamingEcho is bidi streaming.
-   rpc BidirectionalStreamingEcho(stream EchoRequest) returns (stream EchoResponse) {}
- }
\ No newline at end of file

tmp/bi/src/client.rs

-pub mod pb {
-    tonic::include_proto!("grpc.examples.echo");
-}
-
-use futures::stream::Stream;
-use std::time::Duration;
-use tokio_stream::StreamExt;
-use tonic::transport::Channel;
-
-use pb::{echo_client::EchoClient, EchoRequest};
-
-fn echo_requests_iter() -> impl Stream<Item = EchoRequest> {
-    tokio_stream::iter(1..usize::MAX).map(|i| EchoRequest {
-        message: format!("msg {:02}", i),
-        req: "good".to_string(),
-    })
-}
-
-// async fn streaming_echo(client: &mut EchoClient<Channel>, num: usize) {
-//     let stream = client
-//         .server_streaming_echo(EchoRequest {
-//             message: "foo".into(),
-//             req: "world".to_string(),
-//         })
-//         .await
-//         .unwrap()
-//         .into_inner();
-
-//     // stream is infinite - take just 5 elements and then disconnect
-//     let mut stream = stream.take(num);
-//     while let Some(item) = stream.next().await {
-//         println!("item = {:?}", item);
-//         println!("\treceived: {}", item.unwrap().message);
-//     }
-//     // stream is droped here and the disconnect info is send to server
-// }
-
-// async fn bidirectional_streaming_echo(client: &mut EchoClient<Channel>, num: usize) {
-//     let in_stream = echo_requests_iter().take(num);
-
-//     let response = client
-//         .bidirectional_streaming_echo(in_stream)
-//         .await
-//         .unwrap();
-
-//     let mut resp_stream = response.into_inner();
-
-//     while let Some(received) = resp_stream.next().await {
-//         let received = received.unwrap();
-//         println!("\treceived message: `{}`", received.message);
-//     }
-// }
-
-async fn bidirectional_streaming_echo_throttle(client: &mut EchoClient<Channel>, dur: Duration) {
-    println!("bidirectional_streaming_echo_throttle come");
-    let in_stream = echo_requests_iter().throttle(dur);
-
-    let response = client
-        .bidirectional_streaming_echo(in_stream)
-        .await
-        .unwrap();
-
-    let mut resp_stream = response.into_inner();
-
-    while let Some(received) = resp_stream.next().await {
-        let received = received.unwrap();
-        println!("\treceived message: `{}`", received.message);
-    }
-}
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    // let mut client = EchoClient::connect("http://[::1]:50051").await.unwrap();
-    let mut client = EchoClient::connect("http://192.168.110.183:50051")
-        .await
-        .unwrap();
-
-    // println!("Streaming echo:");
-    // streaming_echo(&mut client, 5).await;
-    // tokio::time::sleep(Duration::from_secs(1)).await; //do not mess server println functions
-    //                                                   // tokio::time::sleep(Duration::from_micros(1)).await;
-
-    // // Echo stream that sends 17 requests then graceful end that connection
-    // println!("\r\nBidirectional stream echo:");
-    // bidirectional_streaming_echo(&mut client, 17).await;
-
-    // Echo stream that sends up to `usize::MAX` requets. One request each 2s.
-    // Exiting client with CTRL+C demonstrate how to distinguish broken pipe from
-    //graceful client disconnection (above example) on the server side.
-    println!("\r\nBidirectional stream echo (kill client with CTLR+C):");
-    println!("client ={:?}", client);
-    // bidirectional_streaming_echo_throttle(&mut client, Duration::from_secs(2)).await;
-    // bidirectional_streaming_echo_throttle(&mut client, Duration::from_micros(1)).await;
-
-    Ok(())
-}

tmp/bi/src/server.rs

-pub mod pb {
-    tonic::include_proto!("grpc.examples.echo");
-}
-
-use futures::Stream;
-use std::{error::Error, io::ErrorKind, net::ToSocketAddrs, pin::Pin, time::Duration};
-use tokio::sync::mpsc;
-use tokio_stream::{wrappers::ReceiverStream, StreamExt};
-use tonic::{transport::Server, Request, Response, Status, Streaming};
-
-use pb::{EchoRequest, EchoResponse};
-
-type EchoResult<T> = Result<Response<T>, Status>;
-type ResponseStream = Pin<Box<dyn Stream<Item = Result<EchoResponse, Status>> + Send>>;
-
-fn match_for_io_error(err_status: &Status) -> Option<&std::io::Error> {
-    let mut err: &(dyn Error + 'static) = err_status;
-
-    loop {
-        if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
-            return Some(io_err);
-        }
-
-        // h2::Error do not expose std::io::Error with `source()`
-        // https://github.com/hyperium/h2/pull/462
-        if let Some(h2_err) = err.downcast_ref::<h2::Error>() {
-            if let Some(io_err) = h2_err.get_io() {
-                return Some(io_err);
-            }
-        }
-
-        err = match err.source() {
-            Some(err) => err,
-            None => return None,
-        };
-    }
-}
-
-#[derive(Debug)]
-pub struct EchoServer {}
-
-#[tonic::async_trait]
-impl pb::echo_server::Echo for EchoServer {
-    // async fn unary_echo(&self, _: Request<EchoRequest>) -> EchoResult<EchoResponse> {
-    //     Err(Status::unimplemented("not implemented"))
-    // }
-
-    // type ServerStreamingEchoStream = ResponseStream;
-
-    // async fn server_streaming_echo(
-    //     &self,
-    //     req: Request<EchoRequest>,
-    // ) -> EchoResult<Self::ServerStreamingEchoStream> {
-    //     println!("EchoServer::server_streaming_echo");
-    //     println!("\tclient connected from: {:?}", req.remote_addr());
-
-    //     // creating infinite stream with requested message
-    //     let repeat = std::iter::repeat(EchoResponse {
-    //         message: req.into_inner().message,
-    //         resp: String::from("resp morning"),
-    //     });
-    //     let mut stream = Box::pin(tokio_stream::iter(repeat).throttle(Duration::from_millis(200)));
-
-    //     // spawn and channel are required if you want handle "disconnect" functionality
-    //     // the `out_stream` will not be polled after client disconnect
-    //     let (tx, rx) = mpsc::channel(128);
-    //     tokio::spawn(async move {
-    //         while let Some(item) = stream.next().await {
-    //             match tx.send(Result::<_, Status>::Ok(item)).await {
-    //                 Ok(_) => {
-    //                     println!("server loop item send");
-    //                     // item (server response) was queued to be send to client
-    //                 }
-    //                 Err(_item) => {
-    //                     // output_stream was build from rx and both are dropped
-    //                     break;
-    //                 }
-    //             }
-    //         }
-    //         println!("\tclient disconnected");
-    //     });
-
-    //     let output_stream = ReceiverStream::new(rx);
-    //     Ok(Response::new(
-    //         Box::pin(output_stream) as Self::ServerStreamingEchoStream
-    //     ))
-    // }
-
-    // async fn client_streaming_echo(
-    //     &self,
-    //     _: Request<Streaming<EchoRequest>>,
-    // ) -> EchoResult<EchoResponse> {
-    //     Err(Status::unimplemented("not implemented"))
-    // }
-
-    type BidirectionalStreamingEchoStream = ResponseStream;
-
-    async fn bidirectional_streaming_echo(
-        &self,
-        req: Request<Streaming<EchoRequest>>,
-    ) -> EchoResult<Self::BidirectionalStreamingEchoStream> {
-        println!("EchoServer::bidirectional_streaming_echo");
-        println!("req={:?}", req);
-        let mut in_stream = req.into_inner();
-        println!("in_stream={:?}", in_stream);
-        let (tx, rx) = mpsc::channel(128);
-
-        // this spawn here is required if you want to handle connection error.
-        // If we just map `in_stream` and write it back as `out_stream` the `out_stream`
-        // will be drooped when connection error occurs and error will never be propagated
-        // to mapped version of `in_stream`.
-        tokio::spawn(async move {
-            while let Some(result) = in_stream.next().await {
-                println!("result = {:?}", result);
-                match result {
-                    Ok(v) => tx
-                        .send(Ok(EchoResponse {
-                            message: v.message,
-                            resp: v.req,
-                        }))
-                        .await
-                        .expect("working rx"),
-                    Err(err) => {
-                        if let Some(io_err) = match_for_io_error(&err) {
-                            if io_err.kind() == ErrorKind::BrokenPipe {
-                                // here you can handle special case when client
-                                // disconnected in unexpected way
-                                eprintln!("\tclient disconnected: broken pipe");
-                                break;
-                            }
-                        }
-
-                        match tx.send(Err(err)).await {
-                            Ok(_) => (),
-                            Err(_err) => break, // response was droped
-                        }
-                    }
-                }
-            }
-            println!("\tstream ended");
-        });
-
-        // echo just write the same data that was received
-        let out_stream = ReceiverStream::new(rx);
-
-        Ok(Response::new(
-            Box::pin(out_stream) as Self::BidirectionalStreamingEchoStream
-        ))
-    }
-}
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    let server = EchoServer {};
-    Server::builder()
-        .add_service(pb::echo_server::EchoServer::new(server))
-        // .serve("[::1]:50051".to_socket_addrs().unwrap().next().unwrap())
-        .serve(
-            "192.168.110.183:50051"
-                .to_socket_addrs()
-                .unwrap()
-                .next()
-                .unwrap(),
-        )
-        .await
-        .unwrap();
-
-    Ok(())
-}

tmp/test/Cargo.toml

-[package]
-name = "registry"
-version = "0.1.0"
-edition = "2021"
-
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
-[[bin]] # Bin to run the HelloWorld gRPC server
-name = "server"
-path = "src/server.rs"
-
-[[bin]] # Bin to run the HelloWorld gRPC client
-name = "client"
-path = "src/client.rs"
-
-[[bin]] # Bin to run the HelloWorld gRPC client
-name = "test"
-path = "src/test.rs"
-
-[dependencies]
-tonic = "0.7"
-prost = "0.10"
-tokio = { version = "1.0", features = [ "rt-multi-thread", "time", "fs", "macros", "net",] }
-tokio-stream = { version = "0.1", features = ["net"] }
-
-futures = { version = "0.3", default-features = false, features = ["alloc"] }
-
-h2 = "0.3"
-
-[build-dependencies]
-tonic-build = "0.7"
-futures = "0.3"
-futures-util = "0.3"
\ No newline at end of file

tmp/test/README.md

-# registryFs
-

tmp/test/build.rs

-fn main() -> Result<(), Box<dyn std::error::Error>> {
-    tonic_build::compile_protos("proto/MetadataOps.proto")?;
-    Ok(())
-}

tmp/test/proto/MetadataOps.proto

-/*
- *
- * Copyright beikunyun filesystem team.
- *
- *
- */
-
- syntax = "proto3";
- package bky.registry.api;
- 
- 
- message Metadata{
-     string kv = 1;
- }
- //client向server拉取数据请求创建文件或目录
- message PullCreateReq {
-     //0: 客户端向服务端发送请求; 1: 服务端向客户端发送请求; 其他：未定义行为
-     int32 sendtype = 1;        
-     //Tenent及用户各使用16bits标识码，称为tcode;
-     //在rocksdb里面每个key都会该信息，可以把这部分信息抽离出来，再在数据接收端进行key值的拼接，以减少网络层数据传输的压力
-     string t_code = 2;  
- }
- message PullCreateResp {
-     string t_code = 1;
-    //  repeated Metadata md = 2; 
-     // 0:成功;  -1: 失败; 其他：未定义行为
-     int32 ret_code = 3;
- }
- 
- //client向server推送数据请求创建文件或目录
- message PushCreateReq {
-     //0: 客户端向服务端发送请求; 1: 服务端向客户端发送请求; 其他：未定义行为
-     int32 sendtype = 1;        
-     //Tenent及用户各使用16bits标识码，称为tcode;
-     string tcode = 2;  
-     repeated Metadata metadata = 3; 
- }
- message PushCreateResp {
-     string tcode = 1;
-     //数据推送是否更新成功
-     int32 retCode = 2;
- }
- 
-//  //文件路径,对应rocksdb的key值
-//  message AbsPath{
-//      string k = 1;
-//  }
-//  //删除
-//  message RemoveReq{
-//      string tcode = 1;
-//      repeated AbsPath path = 2;
-//  }
-//  message RemoveResp{
-//      string tcode = 1;
-//      int32 retCode = 2;
-//  }
-//  //更新
-//  message UpdateReq{
-//      string tcode = 1;
-//      repeated Metadata md = 2;
-//  }
-//  message UpdateResp{
-//      string tcode = 1;
-//      int32 retCode = 2;
-//  }
-//  //目录查询
-//  message QueryDirdentsReq{
-//      string tcode = 1;
-//      repeated AbsPath path= 2;
-//  }
-//  message QueryDirdentsResp{
-//      string tcode = 1;
-//      repeated Metadata md = 2;
-//  }
- 
- service MetadataOps {
-     rpc CreateByPull(stream PullCreateReq) returns (stream PullCreateResp) {}
-     rpc CreateByPush(stream PushCreateReq) returns (stream PushCreateResp) {}
-    //  rpc Remove(stream RemoveReq) returns(stream RemoveResp){}
-    //  rpc Update(stream UpdateReq) returns(stream UpdateResp){}
-    //  rpc QueryDirdents(stream QueryDirdentsReq) returns(stream QueryDirdentsResp){}
- }
\ No newline at end of file

tmp/test/src/server.rs

-pub mod registry {
-    tonic::include_proto!("bky.registry.api");
-}
-
-use futures::Stream;
-use registry::{PullCreateReq, PullCreateResp, PushCreateReq, PushCreateResp};
-use std::{error::Error, io::ErrorKind, net::ToSocketAddrs, pin::Pin, result, time::Duration};
-use tokio::sync::mpsc;
-use tokio_stream::{wrappers::ReceiverStream, StreamExt};
-use tonic::{transport::Server, Request, Response, Status, Streaming};
-
-type PushCreateResult<T> = Result<Response<T>, Status>;
-#[derive(Debug)]
-pub struct RegistryServer {}
-
-type RegistryResult<T> = Result<Response<T>, Status>;
-type ResponseStream = Pin<Box<dyn Stream<Item = Result<PullCreateResp, Status>> + Send>>;
-
-fn match_for_io_error(err_status: &Status) -> Option<&std::io::Error> {
-    let mut err: &(dyn Error + 'static) = err_status;
-
-    loop {
-        if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
-            return Some(io_err);
-        }
-
-        // h2::Error do not expose std::io::Error with `source()`
-        // https://github.com/hyperium/h2/pull/462
-        if let Some(h2_err) = err.downcast_ref::<h2::Error>() {
-            if let Some(io_err) = h2_err.get_io() {
-                return Some(io_err);
-            }
-        }
-
-        err = match err.source() {
-            Some(err) => err,
-            None => return None,
-        };
-    }
-}
-#[tonic::async_trait]
-impl registry::metadata_ops_server::MetadataOps for RegistryServer {
-    type CreateByPullStream = ResponseStream;
-    async fn create_by_pull(
-        &self,
-        req: Request<Streaming<PullCreateReq>>,
-    ) -> RegistryResult<Self::CreateByPullStream> {
-        println!("create_by_pull ");
-        println!("req={:?}", req);
-        let mut in_stream = req.into_inner();
-        println!("in_stream={:?}", in_stream);
-        let (tx, rx) = mpsc::channel(128);
-        tokio::spawn(async move {
-            println!("tokio spawn..");
-            while let Some(result) = in_stream.next().await {
-                println!("result = {:?}", result);
-                match result {
-                    Ok(v) => tx
-                        .send(Ok(PullCreateResp {
-                            t_code: v.t_code,
-                            ret_code: 0,
-                        }))
-                        .await
-                        .expect("working rx"),
-                    Err(err) => {
-                        if let Some(io_err) = match_for_io_error(&err) {
-                            if io_err.kind() == ErrorKind::BrokenPipe {
-                                eprintln!("\tclient disconnected: broken pipe");
-                                break;
-                            }
-                        }
-                        match tx.send(Err(err)).await {
-                            Ok(_) => (),
-                            Err(_err) => break,
-                        }
-                    }
-                }
-                println!("\tstream ended");
-            }
-        });
-        let out_stream = ReceiverStream::new(rx);
-        Ok(Response::new(Box::pin(out_stream)))
-    }
-
-    //======================================
-    type CreateByPushStream = Pin<Box<dyn Stream<Item = Result<PushCreateResp, Status>> + Send>>;
-    async fn create_by_push(
-        &self,
-        req: Request<Streaming<PushCreateReq>>,
-    ) -> RegistryResult<Self::CreateByPushStream> {
-        println!("create_by_pull ");
-        println!("req={:?}", req);
-        let mut in_stream = req.into_inner();
-        println!("in_stream={:?}", in_stream);
-        let (tx, rx) = mpsc::channel(128);
-        tokio::spawn(async move {
-            println!("tokio spawn..");
-            while let Some(result) = in_stream.next().await {
-                println!("result = {:?}", result);
-                match result {
-                    Ok(v) => tx
-                        .send(Ok(PushCreateResp {
-                            tcode: v.tcode,
-                            ret_code: 0,
-                        }))
-                        .await
-                        .expect("working rx"),
-                    Err(err) => {
-                        if let Some(io_err) = match_for_io_error(&err) {
-                            if io_err.kind() == ErrorKind::BrokenPipe {
-                                eprintln!("\tclient disconnected: broken pipe");
-                                break;
-                            }
-                        }
-                        match tx.send(Err(err)).await {
-                            Ok(_) => (),
-                            Err(_err) => break,
-                        }
-                    }
-                }
-                println!("\tstream ended");
-            }
-        });
-        let out_stream = ReceiverStream::new(rx);
-        Ok(Response::new(Box::pin(out_stream)))
-    }
-}
-
-#[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
-    println!("main...");
-    let server = RegistryServer {};
-    Server::builder()
-        .add_service(registry::metadata_ops_server::MetadataOpsServer::new(
-            server,
-        ))
-        .serve(
-            "192.168.110.183:50050"
-                .to_socket_addrs()
-                .unwrap()
-                .next()
-                .unwrap(),
-        )
-        .await
-        .unwrap();
-    Ok(())
-}