udp处理初步完成

tickbh · Oct 28, 2023 · 28cc229 · 28cc229
1 parent 554aac6
commit 28cc229
Show file tree

Hide file tree

Showing 2 changed files with 21 additions and 7 deletions.
diff --git a/config/reverse.toml b/config/reverse.toml
@@ -1,28 +1,36 @@
+# 正向代理相关，http/https/socks5等代理配置
 [proxy]
 bind_addr = "0.0.0.0:8090"
 flag = 7
 
+# 反向代理相关，七层协议为http及https
 [http]
 aa = "b"
 
+# 反向代理中的负载均衡地址列表，按名字匹配
 [[http.upstream]]
 name = "server"
 server = [
   {addr="127.0.0.1:8080", fail_timeout=30},
   {addr="127.0.0.1:8081"}
 ]
 
+# 反向代理中的具体服务，可配置多个多组
 [[http.server]]
 bind_addr = "0.0.0.0:82"
 server_name = "soft.wm-proxy.com"
 root=""
+# 若有匹配密钥则表示为SSL连接，反之则为http连接
 #cert="key/soft.wm-proxy.com.pem"
 #key="key/soft.wm-proxy.com.key"
+
+# 请求头返回头相应的处理，如有proxy则为请求头处理，+表示添加，-表示删除，其它表示设置
 headers = [
   ["proxy", "x-forward-for", "$client_ip"],
   ["+", "last-modified", "from proxy"]
 ]
 
+# 按请求路径进行rule匹配，可匹配method，看具体的处理的内容如文件服务或者负载均衡
 [[http.server.location]]
 rule = "/root"
 file_server = { browse = true }
@@ -36,28 +44,31 @@ headers = [
   ["+", "aaa", "bbb"]
 ]
 
-
+# IP的四层协议处理
 [stream]
 
+# 四层协议的负载均衡
 [[stream.upstream]]
 name = "server"
 server = [
   {addr="127.0.0.1:8080", fail_timeout=30},
-  {addr="127.0.0.1:8081"}
+#  {addr="127.0.0.1:8081"}
 ]
 
-
+# 负载均衡，bind为udp则表示udp连接
 [[stream.upstream]]
 name = "udp"
 bind = "udp"
 server = [
   {addr="127.0.0.1:1080"}
 ]
 
+# 四层服务器，server_name映射upstream的name
 [[stream.server]]
 bind_addr = "0.0.0.0:83"
 server_name = "server"
 
+# 四层服务器，udp转发
 [[stream.server]]
 bind_addr = "0.0.0.0:84"
 bind_mode = "udp"

diff --git a/src/reverse/stream.rs b/src/reverse/stream.rs
@@ -139,7 +139,7 @@ impl StreamUdp {
                 return Ok(());
             }
         };
-        udp.connect(remote_addr).await?;
+        // udp.connect(remote_addr).await?;
         let mut cache = vec![0u8; 9096];
         let mut send_cache = LinkedList::<Vec<u8>>::new();
         send_cache.push_back(data);
@@ -156,12 +156,13 @@ impl StreamUdp {
                             Ok((s, _)) => {
                                 sender.send((cache[..s].to_vec(), origin_addr)).await.map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "sender close"))?;
                             },
+                            Err(e) if e.kind() == io::ErrorKind::WouldBlock => {},
                             Err(e) => return Err(e),
                         }
                     }
                     if r.is_writable() {
                         let value = send_cache.pop_front().unwrap();
-                        match udp.send(&value).await {
+                        match udp.send_to(&value, remote_addr).await {
                             Ok(_) => {},
                             Err(e) => {
                                 return Err(e)
@@ -192,7 +193,7 @@ impl StreamUdp {
             let mut remote_addr = None;;
             for up in &self.server.upstream {
                 if up.name == self.server.server_name {
-                    remote_addr = Some(self.server.upstream[0].get_server_addr()?);
+                    remote_addr = Some(up.get_server_addr()?);
                 }
             }
             if remote_addr.is_none() {
@@ -204,7 +205,9 @@ impl StreamUdp {
             self.remote_sockets.insert(addr, PollSender::new(sender));
             let mut sender_clone = self.sender.clone();
             tokio::spawn(async move {
-                let _ = Self::deal_udp_bind(&mut sender_clone, receiver, data, addr, remote_addr).await;
+                if let Err(e) = Self::deal_udp_bind(&mut sender_clone, receiver, data, addr, remote_addr).await {
+                    log::info!("处理UDP信息发生错误，退出:{:?}", e);
+                }
                 let _ = sender_clone.send((vec![], addr)).await;
             });
         }