feat: local track emit event

packages/media_core/src/endpoint.rs

@@ -131,6 +131,7 @@ pub enum EndpointRes {
 #[derive(Debug, PartialEq, Eq)]
 pub enum EndpointLocalTrackEvent {
     Media(MediaPacket),
+    Status(protobuf::shared::receiver::Status),
 }
 
 /// This is used for controlling the remote track, which is sent from endpoint

packages/media_core/src/endpoint/internal/local_track.rs

@@ -8,6 +8,7 @@ use atm0s_sdn::TimePivot;
 use media_server_protocol::{
     endpoint::{PeerId, TrackName, TrackPriority},
     media::MediaKind,
+    protobuf::shared::receiver::Status as ProtoStatus,
     transport::RpcError,
 };
 use sans_io_runtime::{return_if_none, Task, TaskSwitcherChild};
@@ -25,6 +26,8 @@ use super::bitrate_allocator::EgressAction;
 
 mod packet_selector;
 
+const MEDIA_TIMEOUT_MS: u64 = 2_000; //after 2s not receive media, the track will become inactive
+
 pub enum Input {
     JoinRoom(ClusterRoomHash),
     LeaveRoom,
@@ -43,10 +46,17 @@ pub enum Output {
     Stopped(MediaKind),
 }
 
+#[derive(Debug, PartialEq, Eq)]
+enum Status {
+    Waiting,
+    Active { last_media_ts: u64 },
+    Inactive,
+}
+
 pub struct EndpointLocalTrack {
     kind: MediaKind,
     room: Option<ClusterRoomHash>,
-    bind: Option<(PeerId, TrackName)>,
+    bind: Option<(PeerId, TrackName, Status)>,
     queue: VecDeque<Output>,
     selector: PacketSelector,
     timer: TimePivot,
@@ -76,7 +86,7 @@ impl EndpointLocalTrack {
         assert_ne!(self.room, None);
         let room = return_if_none!(self.room.take());
         log::info!("[EndpointLocalTrack] leave room {room}");
-        let (peer, track) = return_if_none!(self.bind.take());
+        let (peer, track, _) = return_if_none!(self.bind.take());
         log::info!("[EndpointLocalTrack] leave room {room} => auto Unsubscribe {peer} {track}");
         self.queue.push_back(Output::Cluster(room, ClusterLocalTrackControl::Unsubscribe));
     }
@@ -94,6 +104,19 @@ impl EndpointLocalTrack {
                 let now_ms = self.timer.timestamp_ms(now);
                 if self.selector.select(self.timer.timestamp_ms(now), channel, &mut pkt).is_some() {
                     self.pop_selector(now_ms);
+
+                    if let Some((_, _, status)) = &mut self.bind {
+                        match status {
+                            Status::Waiting | Status::Inactive => {
+                                *status = Status::Active { last_media_ts: now_ms };
+                                self.queue.push_back(Output::Event(EndpointLocalTrackEvent::Status(ProtoStatus::Active)));
+                            }
+                            Status::Active { last_media_ts } => {
+                                *last_media_ts = now_ms;
+                            }
+                        }
+                    }
+
                     self.queue.push_back(Output::Event(EndpointLocalTrackEvent::Media(pkt)));
                 }
             }
@@ -126,12 +149,12 @@ impl EndpointLocalTrack {
                     let track = source.track;
                     let now_ms = self.timer.timestamp_ms(now);
                     log::info!("[EndpointLocalTrack] view room {room} peer {peer} track {track}");
-                    if let Some((_peer, _track)) = self.bind.take() {
+                    if let Some((_peer, _track, _status)) = self.bind.take() {
                         log::info!("[EndpointLocalTrack] view room {room} peer {peer} track {track} => unsubscribe current {_peer} {_track}");
                         self.queue.push_back(Output::Cluster(*room, ClusterLocalTrackControl::Unsubscribe));
                         self.queue.push_back(Output::Stopped(self.kind));
                     }
-                    self.bind = Some((peer.clone(), track.clone()));
+                    self.bind = Some((peer.clone(), track.clone(), Status::Waiting));
                     self.selector.set_limit_layer(now_ms, config.max_spatial, config.max_temporal);
                     self.queue.push_back(Output::Started(self.kind, config.priority));
                     self.queue.push_back(Output::Cluster(*room, ClusterLocalTrackControl::Subscribe(peer, track)));
@@ -145,7 +168,7 @@ impl EndpointLocalTrack {
             EndpointLocalTrackReq::Detach() => {
                 //TODO process config here
                 if let Some(room) = self.room.as_ref() {
-                    if let Some((peer, track)) = self.bind.take() {
+                    if let Some((peer, track, _)) = self.bind.take() {
                         self.queue.push_back(Output::RpcRes(req_id, EndpointLocalTrackRes::Detach(Ok(()))));
                         self.queue.push_back(Output::Stopped(self.kind));
                         self.queue.push_back(Output::Cluster(*room, ClusterLocalTrackControl::Unsubscribe));
@@ -205,6 +228,15 @@ impl Task<Input, Output> for EndpointLocalTrack {
         let now_ms = self.timer.timestamp_ms(now);
         self.selector.on_tick(now_ms);
         self.pop_selector(now_ms);
+
+        if let Some((_, _, status)) = &mut self.bind {
+            if let Status::Active { last_media_ts } = status {
+                if now_ms >= *last_media_ts + MEDIA_TIMEOUT_MS {
+                    *status = Status::Inactive;
+                    self.queue.push_back(Output::Event(EndpointLocalTrackEvent::Status(ProtoStatus::Inactive)));
+                }
+            }
+        }
     }
 
     fn on_event(&mut self, now: Instant, input: Input) {

packages/protocol/proto/shared.proto

@@ -9,11 +9,9 @@ enum Kind {
 
 message Receiver {
     enum Status {
-        NO_SOURCE = 0;
-        WAITING = 1;
-        LIVE = 2;
-        KEY_ONLY = 3;
-        INACTIVE = 4;
+        WAITING = 0;
+        ACTIVE = 1;
+        INACTIVE = 2;
     }
 
     message Source {

packages/protocol/src/protobuf/shared.rs

@@ -54,11 +54,9 @@ pub mod receiver {
     )]
     #[repr(i32)]
     pub enum Status {
-        NoSource = 0,
-        Waiting = 1,
-        Live = 2,
-        KeyOnly = 3,
-        Inactive = 4,
+        Waiting = 0,
+        Active = 1,
+        Inactive = 2,
     }
     impl Status {
         /// String value of the enum field names used in the ProtoBuf definition.
@@ -67,20 +65,16 @@ pub mod receiver {
         /// (if the ProtoBuf definition does not change) and safe for programmatic use.
         pub fn as_str_name(&self) -> &'static str {
             match self {
-                Status::NoSource => "NO_SOURCE",
                 Status::Waiting => "WAITING",
-                Status::Live => "LIVE",
-                Status::KeyOnly => "KEY_ONLY",
+                Status::Active => "ACTIVE",
                 Status::Inactive => "INACTIVE",
             }
         }
         /// Creates an enum from field names used in the ProtoBuf definition.
         pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
             match value {
-                "NO_SOURCE" => Some(Self::NoSource),
                 "WAITING" => Some(Self::Waiting),
-                "LIVE" => Some(Self::Live),
-                "KEY_ONLY" => Some(Self::KeyOnly),
+                "ACTIVE" => Some(Self::Active),
                 "INACTIVE" => Some(Self::Inactive),
                 _ => None,
             }

packages/transport_webrtc/src/transport/webrtc.rs

@@ -10,14 +10,15 @@ use media_server_protocol::{
         self,
         conn::{
             server_event::{
+                receiver::{Event as ProtoReceiverEvent, State as ProtoReceiverState},
                 room::{Event as ProtoRoomEvent2, PeerJoined, PeerLeaved, TrackStarted, TrackStopped},
                 sender::{Event as ProtoSenderEvent, State as ProtoSenderState},
                 Event as ProtoServerEvent, Receiver as ProtoReceiverEventContainer, Room as ProtoRoomEvent, Sender as ProtoSenderEventContainer,
             },
             ClientEvent,
         },
         gateway::ConnectRequest,
-        shared::{receiver::Status as ProtoReceiverStatus, sender::Status as ProtoSenderStatus, Kind},
+        shared::{sender::Status as ProtoSenderStatus, Kind},
     },
     transport::{RpcError, RpcResult},
 };
@@ -248,6 +249,14 @@ impl TransportWebrtcInternal for TransportWebrtcSdk {
                     log::trace!("[TransportWebrtcSdk] send {:?} size {}", pkt.meta, pkt.data.len());
                     self.queue.push_back(InternalOutput::Str0mSendMedia(mid, pkt))
                 }
+                EndpointLocalTrackEvent::Status(status) => {
+                    let track = return_if_none!(self.local_track(track_id)).name().to_string();
+                    log::info!("[TransportWebrtcSdk] track {track} set status {:?}", status);
+                    self.send_event(ProtoServerEvent::Receiver(ProtoReceiverEventContainer {
+                        name: track,
+                        event: Some(ProtoReceiverEvent::State(ProtoReceiverState { status: status as i32 })),
+                    }));
+                }
             },
             EndpointEvent::BweConfig { current, desired } => {
                 let (current, desired) = self.bwe_state.filter_bwe_config(current, desired);

packages/transport_webrtc/src/transport/whep.rs

@@ -148,6 +148,7 @@ impl TransportWebrtcInternal for TransportWebrtcWhep {
                     };
                     self.queue.push_back(InternalOutput::Str0mSendMedia(mid, pkt));
                 }
+                EndpointLocalTrackEvent::Status(_) => {}
             },
             EndpointEvent::RemoteMediaTrack(_track, _event) => {}
             EndpointEvent::BweConfig { current, desired } => {