fix-interceptor-poll_write-bug

Author: yngrtc

examples/sync_signal/mod.rs

@@ -117,7 +117,7 @@ pub fn sync_run(
             }
         };
 
-        write_socket_output(&socket, &pipeline);
+        write_socket_output(&socket, &pipeline)?;
 
         // Spawn new incoming signal message from the signaling server thread.
         if let Ok(signal_message) = rx.try_recv() {
@@ -161,12 +161,13 @@ pub fn sync_run(
 fn write_socket_output(
     socket: &UdpSocket,
     pipeline: &Rc<Pipeline<TaggedBytesMut, TaggedBytesMut>>,
-) {
+)  -> anyhow::Result<()>{
     while let Some(transmit) = pipeline.poll_transmit() {
         socket
-            .send_to(&transmit.message, transmit.transport.peer_addr)
-            .expect("sending UDP data");
+            .send_to(&transmit.message, transmit.transport.peer_addr)?;
     }
+
+    Ok(())
 }
 
 fn read_socket_input(socket: &UdpSocket, buf: &mut [u8]) -> Option<TaggedBytesMut> {

src/handler/gateway.rs

@@ -386,6 +386,8 @@ impl GatewayHandler {
         transport_context: TransportContext,
         rtp_packet: rtp::packet::Packet,
     ) -> Result<Vec<TaggedMessageEvent>> {
+        debug!("handle_rtp_message {}", transport_context.peer_addr);
+
         //TODO: Selective Forwarding RTP Packets
         let peers =
             GatewayHandler::get_other_media_transport_contexts(server_states, &transport_context)?;
@@ -408,6 +410,8 @@ impl GatewayHandler {
         transport_context: TransportContext,
         rtcp_packets: Vec<Box<dyn rtcp::packet::Packet>>,
     ) -> Result<Vec<TaggedMessageEvent>> {
+        debug!("handle_rtcp_message {}", transport_context.peer_addr);
+
         //TODO: Selective Forwarding RTCP Packets
         let peers =
             GatewayHandler::get_other_media_transport_contexts(server_states, &transport_context)?;

src/handler/interceptor.rs

@@ -2,7 +2,7 @@ use crate::interceptor::InterceptorEvent;
 use crate::messages::{MessageEvent, RTPMessageEvent, TaggedMessageEvent};
 use crate::types::FourTuple;
 use crate::ServerStates;
-use log::error;
+use log::{debug, error};
 use retty::channel::{Context, Handler};
 use shared::error::Result;
 use std::cell::RefCell;
@@ -56,6 +56,7 @@ impl Handler for InterceptorHandler {
                     for event in events {
                         match event {
                             InterceptorEvent::Inbound(inbound) => {
+                                debug!("interceptor forward Rtcp {:?}", msg.transport.peer_addr);
                                 ctx.fire_read(inbound);
                             }
                             InterceptorEvent::Outbound(outbound) => {
@@ -77,10 +78,12 @@ impl Handler for InterceptorHandler {
             if let MessageEvent::Rtp(RTPMessageEvent::Rtcp(_)) = &msg.message {
                 // RTCP message read must end here in SFU case. If any rtcp packet needs to be forwarded to other Endpoints,
                 // just add a new interceptor to forward it.
+                debug!("interceptor terminates Rtcp {:?}", msg.transport.peer_addr);
                 return;
             }
         }
 
+        debug!("interceptor read bypass {:?}", msg.transport.peer_addr);
         ctx.fire_read(msg);
     }
 
@@ -197,6 +200,9 @@ impl Handler for InterceptorHandler {
                     }
                 };
             }
+
+            debug!("interceptor write {:?}", msg.transport.peer_addr);
+            self.transmits.push_back(msg);
         }
 
         self.transmits.pop_front()

src/handler/stun.rs

@@ -58,7 +58,7 @@ impl Handler for StunHandler {
                 }
             }
         } else {
-            debug!("bypass StunInbound for {}", msg.transport.peer_addr);
+            debug!("bypass StunHandler read for {}", msg.transport.peer_addr);
             ctx.fire_read(msg);
         }
     }
@@ -81,7 +81,7 @@ impl Handler for StunHandler {
                     message: MessageEvent::Stun(STUNMessageEvent::Raw(message)),
                 })
             } else {
-                debug!("bypass StunOutbound for {}", msg.transport.peer_addr);
+                debug!("bypass StunHandler write for {}", msg.transport.peer_addr);
                 Some(msg)
             }
         } else {