From 76eaead0b92146824733fc8b2cf720d740aac799 Mon Sep 17 00:00:00 2001
From: lubeilin <1791778603@qq.com>
Date: Sun, 31 Dec 2023 13:42:32 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E5=AF=B9=E7=A7=B0=E7=BD=91?=
 =?UTF-8?q?=E7=BB=9C=E5=A4=84=E7=90=86=E9=80=BB=E8=BE=91?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 vnt/src/channel/channel.rs | 111 ++++++++++---------------------------
 1 file changed, 29 insertions(+), 82 deletions(-)

diff --git a/vnt/src/channel/channel.rs b/vnt/src/channel/channel.rs
index d2f090d..a00ab47 100644
--- a/vnt/src/channel/channel.rs
+++ b/vnt/src/channel/channel.rs
@@ -523,7 +523,6 @@ impl Channel {
             let context = context.clone();
             let main_channel = main_channel.try_clone().unwrap();
             let handler = handler.clone();
-            let buf_sender = buf_sender.clone();
             thread::Builder::new()
                 .name("channel_udp".into())
                 .spawn(move || {
@@ -570,7 +569,7 @@ impl Channel {
                                             Ok(udp) => {
                                                 let udp = Arc::new(udp);
                                                 let context = context.clone();
-                                                tokio::spawn(Self::start_(worker.worker("symmetric_channel"),context, udp,handler.clone(),buf_sender.clone(), head_reserve, false));
+                                                tokio::spawn(Self::start_(worker.worker("symmetric_channel"),context, udp,handler.clone(), head_reserve));
                                             }
                                             Err(e) => {
                                                 log::error!("{}",e);
@@ -653,9 +652,7 @@ impl Channel {
         context: Context,
         udp: Arc<UdpSocket>,
         handler: ChannelDataHandler,
-        buf_sender: Option<BufSenderGroup>,
         head_reserve: usize,
-        is_core: bool,
     ) {
         let mut status_receiver = context.inner.status_receiver.clone();
         #[cfg(target_os = "windows")]
@@ -668,92 +665,42 @@ impl Channel {
         let id = 3 + udp.as_raw_fd() as usize;
 
         context.insert_udp(id, udp.clone());
-        match buf_sender {
-            None => {
-                let mut buf = [0; 4096];
-                loop {
-                    tokio::select! {
-                        rs=udp.recv_from(&mut buf[head_reserve..])=>{
-                              match rs {
-                                Ok((len, addr)) => {
-                                    handler.handle(&mut buf, head_reserve, head_reserve + len, RouteKey::new(id, addr), &context);
-                                }
-                                Err(e) => {
-                                    log::error!("{:?}",e)
-                                }
-                            }
+        let mut buf = [0; 4096];
+        loop {
+            tokio::select! {
+                rs=udp.recv_from(&mut buf[head_reserve..])=>{
+                      match rs {
+                        Ok((len, addr)) => {
+                            handler.handle(&mut buf, head_reserve, head_reserve + len, RouteKey::new(id, addr), &context);
                         }
-                        changed=status_receiver.changed()=>{
-                                match changed {
-                                    Ok(_) => {
-                                        match *status_receiver.borrow() {
-                                            Status::Cone => {
-                                                if !is_core{
-                                                    break;
-                                                }
-                                            }
-                                            Status::Close=>{
-                                                break;
-                                            }
-                                            Status::Symmetric => {}
-                                        }
-                                    }
-                                    Err(_) => {
+                        Err(e) => {
+                            log::error!("{:?}",e)
+                        }
+                    }
+                }
+                changed=status_receiver.changed()=>{
+                        match changed {
+                            Ok(_) => {
+                                match *status_receiver.borrow() {
+                                    Status::Cone => {
                                         break;
                                     }
+                                    Status::Close=>{
+                                        break;
+                                    }
+                                    Status::Symmetric => {}
                                 }
+                            }
+                            Err(_) => {
+                                break;
+                            }
                         }
-                        _=worker.stop_wait()=>{
-                            break;
-                        }
-                    }
+                }
+                _=worker.stop_wait()=>{
+                    break;
                 }
             }
-            Some(mut buf_sender) => loop {
-                let mut buf = vec![0; 4096];
-                tokio::select! {
-                    rs=udp.recv_from(&mut buf[head_reserve..])=>{
-                         match rs {
-                            Ok((len, addr)) => {
-                                if !buf_sender.send((buf,head_reserve,head_reserve+len,RouteKey::new(id, addr))){
-                                     log::error!("udp buf_sender发送数据失败");
-                                     break;
-                                }
-                            }
-                            Err(e) => {
-                                log::error!("{:?}",e)
-                            }
-                        }
-                    }
-                    changed=status_receiver.changed()=>{
-                            match changed {
-                                Ok(_) => {
-                                    match *status_receiver.borrow() {
-                                        Status::Cone => {
-                                            if !is_core{
-                                                break;
-                                            }
-                                        }
-                                        Status::Close=>{
-                                            break;
-                                        }
-                                        Status::Symmetric => {}
-                                    }
-                                }
-                                Err(_) => {
-                                    break;
-                                }
-                            }
-                    }
-                    _=worker.stop_wait()=>{
-                        break;
-                    }
-                }
-            },
         }
         context.remove_udp(id);
-        if is_core {
-            worker.stop_all();
-        }
     }
 }