fix: loop break

binbat · Jan 21, 2024 · 3585c15 · 3585c15
1 parent 7901382
commit 3585c15
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 9 deletions.
diff --git a/src/forward/forward_internal.rs b/src/forward/forward_internal.rs
@@ -126,12 +126,8 @@ impl PeerForwardInternal {
                     }
                 };
                 let r = Arc::clone(&raw);
-                tokio::spawn(async move {
-                    let _ = Self::data_channel_read_loop(r, sender).await;
-                });
-                tokio::spawn(async move {
-                    let _ = Self::data_channel_write_loop(raw, receiver).await;
-                });
+                tokio::spawn(Self::data_channel_read_loop(r, sender));
+                tokio::spawn(Self::data_channel_write_loop(raw, receiver));
             });
 
             Box::pin(async {})
@@ -148,6 +144,9 @@ impl PeerForwardInternal {
                     return;
                 }
             };
+            if n == 0 {
+                break;
+            }
             if let Err(err) = sender.send(buffer[..n].to_vec()) {
                 info!("send data channel err: {}", err);
                 return;
@@ -162,6 +161,7 @@ impl PeerForwardInternal {
         while let Ok(msg) = receiver.recv().await {
             if let Err(err) = d.write(&msg.into()).await {
                 info!("write data channel err: {}", err);
+                return;
             };
         }
     }

diff --git a/src/forward/subscribe.rs b/src/forward/subscribe.rs
@@ -123,15 +123,14 @@ impl SubscribeRTCPeerConnection {
             error!("[{}] [{}] subscribe video tracks is empty", path, id);
             return;
         }
-        info!("[{}] [{}] video up", path, id);
         let subscribe_video_track = subscribe_video_tracks.get(0).unwrap();
         let mut recv = match (subscribe_video_track.rtp_recv)() {
             Ok(recv) => recv,
             Err(_err) => {
                 return;
             }
         };
-
+        info!("[{}] [{}] video up", path, id);
         let mut track_binding_publish_rid_one = track_binding_publish_rid.write().await;
         track_binding_publish_rid_one.insert(
             track.stream_id().to_owned(),
@@ -175,7 +174,7 @@ impl SubscribeRTCPeerConnection {
                                     recv= match (subscribe_track.rtp_recv)() {
                                         Ok(recv) => recv,
                                         Err(_err) => {
-                                           return ;
+                                            return ;
                                         }
                                     };
                                     publish_rtcp_sender.send((RtcpMessage::PictureLossIndication, subscribe_track.track.ssrc())).await.unwrap();