Merge pull request #48 from Tim-Zhang/port-from-dragonball

Port from dragonball

Author: lifupan

Cargo.toml

@@ -17,6 +17,7 @@ libc = { version = "0.2.59", features = [ "extra_traits" ] }
 nix = "0.16.1"
 log = "0.4"
 byteorder = "1.3.2"
+thiserror = "1.0"
 
 async-trait = { version = "0.1.31", optional = true }
 tokio = { version = "0.2", features = ["rt-threaded", "sync", "uds", "stream", "macros", "io-util"], optional = true }

src/common.rs

@@ -96,6 +96,7 @@ pub fn do_bind(host: &str) -> Result<(RawFd, Domain)> {
         }
     };
 
+    setsockopt(fd, sockopt::ReusePort, &true).ok();
     bind(fd, &sockaddr).map_err(err_to_Others!(e, ""))?;
 
     Ok((fd, domain))

src/error.rs

@@ -16,12 +16,18 @@
 
 use crate::ttrpc::{Code, Status};
 use std::result;
+use thiserror::Error;
 
 /// The error type for ttrpc.
-#[derive(Debug)]
+#[derive(Error, Debug)]
 pub enum Error {
+    #[error("socket err: {0}")]
     Socket(String),
+
+    #[error("rpc status: {0:?}")]
     RpcStatus(Status),
+
+    #[error("ttrpc err: {0}")]
     Others(String),
 }
 

src/sync/client.rs

@@ -29,6 +29,7 @@ use crate::error::{Error, Result};
 use crate::sync::channel::{read_message, write_message};
 use crate::ttrpc::{Code, Request, Response};
 use crate::MessageHeader;
+use std::time::Duration;
 
 type Sender = mpsc::Sender<(Vec<u8>, mpsc::SyncSender<Result<Vec<u8>>>)>;
 type Receiver = mpsc::Receiver<(Vec<u8>, mpsc::SyncSender<Result<Vec<u8>>>)>;
@@ -176,9 +177,17 @@ impl Client {
         self.sender_tx
             .send((buf, tx))
             .map_err(err_to_Others!(e, "Send packet to sender error "))?;
-        let result = rx
-            .recv()
-            .map_err(err_to_Others!(e, "Recive packet from recver error "))?;
+
+        let result: Result<Vec<u8>>;
+        if req.timeout_nano == 0 {
+            result = rx
+                .recv()
+                .map_err(err_to_Others!(e, "Receive packet from recver error: "))?;
+        } else {
+            result = rx
+                .recv_timeout(Duration::from_nanos(req.timeout_nano as u64))
+                .map_err(err_to_Others!(e, "Receive packet from recver timeout: "))?;
+        }
 
         let buf = result?;
         let mut s = CodedInputStream::from_bytes(&buf);

src/sync/server.rs

@@ -21,7 +21,7 @@ use nix::unistd::close;
 use nix::unistd::pipe2;
 use protobuf::{CodedInputStream, Message};
 use std::collections::HashMap;
-use std::os::unix::io::RawFd;
+use std::os::unix::io::{AsRawFd, FromRawFd, RawFd};
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::mpsc::{channel, sync_channel, Receiver, Sender, SyncSender};
 use std::sync::{Arc, Mutex};
@@ -279,8 +279,9 @@ impl Server {
         }
 
         let (fd, _) = common::do_bind(host)?;
-        self.listeners.push(fd);
+        common::do_listen(fd)?;
 
+        self.listeners.push(fd);
         Ok(self)
     }
 
@@ -341,8 +342,6 @@ impl Server {
         let service_quit = self.quit.clone();
         let monitor_fd = self.monitor_fd.0;
 
-        common::do_listen(listener)?;
-
         let handler = thread::Builder::new()
             .name("listener_loop".into())
             .spawn(move || {
@@ -508,3 +507,15 @@ impl Server {
         }
     }
 }
+
+impl FromRawFd for Server {
+    unsafe fn from_raw_fd(fd: RawFd) -> Self {
+        Self::default().add_listener(fd).unwrap()
+    }
+}
+
+impl AsRawFd for Server {
+    fn as_raw_fd(&self) -> RawFd {
+        self.listeners[0]
+    }
+}