refactored

Cargo.lock

@@ -2,7 +2,7 @@
 # It is not intended for manual editing.
 [[package]]
 name = "aho-corasick"
-version = "0.7.9"
+version = "0.7.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "memchr 2.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -155,22 +155,22 @@ dependencies = [
 name = "hbb_common"
 version = "0.1.0"
 dependencies = [
+ "bytes 0.5.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "env_logger 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures 0.3.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "protobuf 2.10.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "protobuf-codegen-pure 2.10.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio 0.2.13 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-util 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
 name = "hbbs"
 version = "0.1.0"
 dependencies = [
- "bytes 0.5.4 (registry+https://github.com/rust-lang/crates.io-index)",
- "env_logger 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "futures 0.3.4 (registry+https://github.com/rust-lang/crates.io-index)",
  "hbb_common 0.1.0",
- "log 0.4.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "simple-error 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "tokio 0.2.13 (registry+https://github.com/rust-lang/crates.io-index)",
- "tokio-util 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -385,7 +385,7 @@ name = "regex"
 version = "1.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
- "aho-corasick 0.7.9 (registry+https://github.com/rust-lang/crates.io-index)",
+ "aho-corasick 0.7.10 (registry+https://github.com/rust-lang/crates.io-index)",
  "memchr 2.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "regex-syntax 0.6.16 (registry+https://github.com/rust-lang/crates.io-index)",
  "thread_local 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -551,7 +551,7 @@ dependencies = [
 ]
 
 [metadata]
-"checksum aho-corasick 0.7.9 (registry+https://github.com/rust-lang/crates.io-index)" = "d5e63fd144e18ba274ae7095c0197a870a7b9468abc801dd62f190d80817d2ec"
+"checksum aho-corasick 0.7.10 (registry+https://github.com/rust-lang/crates.io-index)" = "8716408b8bc624ed7f65d223ddb9ac2d044c0547b6fa4b0d554f3a9540496ada"
 "checksum arc-swap 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)" = "d7b8a9123b8027467bce0099fe556c628a53c8d83df0507084c31e9ba2e39aff"
 "checksum atty 0.2.14 (registry+https://github.com/rust-lang/crates.io-index)" = "d9b39be18770d11421cdb1b9947a45dd3f37e93092cbf377614828a319d5fee8"
 "checksum bitflags 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "cf1de2fe8c75bc145a2f577add951f8134889b4795d47466a54a5c846d691693"

Cargo.toml

@@ -7,12 +7,6 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-tokio = { version = "0.2", features = ["full"] }
-tokio-util = { version = "0.3", features = ["full"] }
-log = "0.4"
-env_logger = "0.7"
-futures = "0.3"
-bytes = "0.5"
 simple-error = "0.2"
 hbb_common = { path = "libs/hbb_common" }
 

libs/hbb_common

@@ -1 +1 @@
-Subproject commit 974815baaaa2efd886be73c3be8fb5931bcc29c4
+Subproject commit 7ba274fdaba5057fdefe6a3b47dc3255e78da3fe

src/main.rs

@@ -1,10 +1,11 @@
 // https://tools.ietf.org/rfc/rfc5128.txt
 // https://blog.csdn.net/bytxl/article/details/44344855
 
+use hbb_common::{env_logger, log, tokio, ResultType};
 use hbbs::*;
 
 #[tokio::main]
-async fn main() -> Result<(), Box<dyn std::error::Error>> {
+async fn main() -> ResultType<()> {
     env_logger::init();
     let addr = "0.0.0.0:21116";
     log::info!("Start Server {}", addr);

src/rendezvous_server.rs

@@ -1,13 +1,13 @@
-use bytes::{Bytes, BytesMut};
-use futures::SinkExt;
 use hbb_common::{
+    bytes::BytesMut,
+    log,
     message_proto::*,
-    protobuf::{parse_from_bytes, Message as _},
-    AddrMangle,
+    protobuf::parse_from_bytes,
+    tokio::{net::UdpSocket, stream::StreamExt},
+    udp::FramedSocket,
+    AddrMangle, ResultType,
 };
-use std::{collections::HashMap, error::Error, net::SocketAddr, time::Duration};
-use tokio::{net::UdpSocket, stream::StreamExt, time::delay_for};
-use tokio_util::{codec::BytesCodec, udp::UdpFramed};
+use std::{collections::HashMap, net::SocketAddr};
 
 pub struct Peer {
     socket_addr: SocketAddr,
@@ -19,13 +19,10 @@ pub struct RendezvousServer {
     peer_map: PeerMap,
 }
 
-type FramedSocket = UdpFramed<BytesCodec>;
-type ResultType = Result<(), Box<dyn Error>>;
-
 impl RendezvousServer {
-    pub async fn start(addr: &str) -> ResultType {
+    pub async fn start(addr: &str) -> ResultType<()> {
         let socket = UdpSocket::bind(addr).await?;
-        let mut socket = UdpFramed::new(socket, BytesCodec::new());
+        let mut socket = FramedSocket::new(socket);
         let mut rs = Self {
             peer_map: PeerMap::new(),
         };
@@ -40,7 +37,7 @@ impl RendezvousServer {
         bytes: &BytesMut,
         addr: SocketAddr,
         socket: &mut FramedSocket,
-    ) -> ResultType {
+    ) -> ResultType<()> {
         if let Ok(msg_in) = parse_from_bytes::<Message>(&bytes) {
             match msg_in.union {
                 Some(Message_oneof_union::register_peer(rp)) => {
@@ -64,7 +61,7 @@ impl RendezvousServer {
                             socket_addr: AddrMangle::encode(&addr),
                             ..Default::default()
                         });
-                        send_to(&msg_out, peer.socket_addr, socket).await?;
+                        socket.send(&msg_out, peer.socket_addr).await?;
                     }
                 }
                 Some(Message_oneof_union::punch_hole_sent(phs)) => {
@@ -76,7 +73,7 @@ impl RendezvousServer {
                         socket_addr: AddrMangle::encode(&addr),
                         ..Default::default()
                     });
-                    send_to(&msg_out, addr_a, socket).await?;
+                    socket.send(&msg_out, addr_a).await?;
                 }
                 _ => {}
             }
@@ -85,22 +82,11 @@ impl RendezvousServer {
     }
 }
 
-#[inline]
-pub async fn send_to(msg: &Message, addr: SocketAddr, socket: &mut FramedSocket) -> ResultType {
-    socket
-        .send((Bytes::from(msg.write_to_bytes().unwrap()), addr))
-        .await?;
-    Ok(())
-}
-
-#[inline]
-pub async fn sleep(sec: f32) {
-    delay_for(Duration::from_secs_f32(sec)).await;
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
+    use hbb_common::tokio;
+    use std::time::Duration;
 
     #[allow(unused_must_use)]
     #[tokio::main]
@@ -118,23 +104,23 @@ mod tests {
             // B register it to server
             let socket_b = UdpSocket::bind("127.0.0.1:0").await.unwrap();
             let local_addr_b = socket_b.local_addr().unwrap();
-            let mut socket_b = UdpFramed::new(socket_b, BytesCodec::new());
+            let mut socket_b = FramedSocket::new(socket_b);
             let mut msg_out = Message::new();
             msg_out.set_register_peer(RegisterPeer {
                 hbb_addr: "123".to_string(),
                 ..Default::default()
             });
-            send_to(&msg_out, addr_server, &mut socket_b).await;
+            socket_b.send(&msg_out, addr_server).await;
 
             // A send punch request to server
             let socket_a = UdpSocket::bind("127.0.0.1:0").await.unwrap();
             let local_addr_a = socket_a.local_addr().unwrap();
-            let mut socket_a = UdpFramed::new(socket_a, BytesCodec::new());
+            let mut socket_a = FramedSocket::new(socket_a);
             msg_out.set_punch_hole_request(PunchHoleRequest {
                 hbb_addr: "123".to_string(),
                 ..Default::default()
             });
-            send_to(&msg_out, addr_server, &mut socket_a).await;
+            socket_a.send(&msg_out, addr_server).await;
 
             println!(
                 "A {:?} request punch hole to B {:?} via server {:?}",
@@ -160,7 +146,7 @@ mod tests {
                     socket_addr: AddrMangle::encode(&remote_addr_a),
                     ..Default::default()
                 });
-                send_to(&msg_out, addr_server, &mut socket_b).await;
+                socket_b.send(&msg_out, addr_server).await;
             }
 
             // on A side