Bring back RTP and RTCP notifications

versatica · Sep 28, 2024 · b247930 · b247930
1 parent 4e161f6
commit b247930
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 24 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/rust/src/router/consumer.rs b/rust/src/router/consumer.rs
@@ -593,8 +593,7 @@ enum Notification {
     ProducerClose,
     ProducerPause,
     ProducerResume,
-    // TODO.
-    // Rtp,
+    Rtp(Vec<u8>),
     Score(ConsumerScore),
     LayersChange(Option<ConsumerLayers>),
     Trace(ConsumerTraceEventData),
@@ -608,6 +607,17 @@ impl Notification {
             notification::Event::ConsumerProducerClose => Ok(Notification::ProducerClose),
             notification::Event::ConsumerProducerPause => Ok(Notification::ProducerPause),
             notification::Event::ConsumerProducerResume => Ok(Notification::ProducerResume),
+            notification::Event::ConsumerRtp => {
+                let Ok(Some(notification::BodyRef::ConsumerRtpNotification(body))) =
+                    notification.body()
+                else {
+                    panic!("Wrong message from worker: {notification:?}");
+                };
+
+                let rtp_notification_fbs = consumer::RtpNotification::try_from(body).unwrap();
+
+                Ok(Notification::Rtp(rtp_notification_fbs.data))
+            }
             notification::Event::ConsumerScore => {
                 let Ok(Some(notification::BodyRef::ConsumerScoreNotification(body))) =
                     notification.body()
@@ -841,14 +851,11 @@ impl Consumer {
                                 handlers.resume.call_simple();
                             }
                         }
-                        /*
-                         * TODO.
-                        Notification::Rtp => {
+                        Notification::Rtp(data) => {
                             handlers.rtp.call(|callback| {
-                                callback(notification);
+                                callback(&data);
                             });
                         }
-                        */
                         Notification::Score(consumer_score) => {
                             *score.lock() = consumer_score.clone();
                             handlers.score.call_simple(&consumer_score);

diff --git a/rust/src/router/direct_transport.rs b/rust/src/router/direct_transport.rs
@@ -231,8 +231,7 @@ struct Handlers {
 #[serde(tag = "event", rename_all = "lowercase", content = "data")]
 enum Notification {
     Trace(TransportTraceEventData),
-    // TODO.
-    // Rtcp,
+    Rtcp(Vec<u8>),
 }
 
 impl Notification {
@@ -252,17 +251,18 @@ impl Notification {
 
                 Ok(Notification::Trace(trace_notification))
             }
-            /*
-             * TODO.
             notification::Event::DirecttransportRtcp => {
-                let Ok(Some(notification::BodyRef::RtcpNotification(_body))) = notification.body()
+                let Ok(Some(notification::BodyRef::DirectTransportRtcpNotification(body))) =
+                    notification.body()
                 else {
                     panic!("Wrong message from worker: {notification:?}");
                 };
 
-                Ok(Notification::Rtcp)
+                let rtcp_notification_fbs =
+                    direct_transport::RtcpNotification::try_from(body).unwrap();
+
+                Ok(Notification::Rtcp(rtcp_notification_fbs.data))
             }
-            */
             _ => Err(NotificationParseError::InvalidEvent),
         }
     }
@@ -574,14 +574,12 @@ impl DirectTransport {
                     Ok(notification) => match notification {
                         Notification::Trace(trace_event_data) => {
                             handlers.trace.call_simple(&trace_event_data);
-                        } /*
-                           * TODO.
-                          Notification::Rtcp => {
-                              handlers.rtcp.call(|callback| {
-                                  callback(notification);
-                              });
-                          }
-                          */
+                        }
+                        Notification::Rtcp(data) => {
+                            handlers.rtcp.call(|callback| {
+                                callback(&data);
+                            });
+                        }
                     },
                     Err(error) => {
                         error!("Failed to parse notification: {}", error);