feat: create new session if ice-restart request with not exists conn_id

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

@@ -78,7 +78,7 @@ impl EndpointLocalTrack {
     fn on_join_room(&mut self, _now: Instant, room: ClusterRoomHash) {
         assert_eq!(self.room, None);
         assert_eq!(self.bind, None);
-        log::info!("[EndpointLocalTrack] join room {room}");
+        log::info!("[EndpointLocalTrack] track {} join room {room}", self.kind);
         self.room = Some(room);
     }
 
@@ -148,9 +148,12 @@ impl EndpointLocalTrack {
                     let peer = source.peer;
                     let track = source.track;
                     let now_ms = self.timer.timestamp_ms(now);
-                    log::info!("[EndpointLocalTrack] view room {room} peer {peer} track {track}");
+                    log::info!("[EndpointLocalTrack] track {} view room {room} peer {peer} track {track}", self.kind);
                     if let Some((_peer, _track, _status)) = self.bind.take() {
-                        log::info!("[EndpointLocalTrack] view room {room} peer {peer} track {track} => unsubscribe current {_peer} {_track}");
+                        log::info!(
+                            "[EndpointLocalTrack]  track {} view room {room} peer {peer} track {track} => unsubscribe current {_peer} {_track}",
+                            self.kind
+                        );
                         self.queue.push_back(Output::Cluster(*room, ClusterLocalTrackControl::Unsubscribe));
                         self.queue.push_back(Output::Stopped(self.kind));
                     }
@@ -160,7 +163,7 @@ impl EndpointLocalTrack {
                     self.queue.push_back(Output::Cluster(*room, ClusterLocalTrackControl::Subscribe(peer, track)));
                     self.selector.reset();
                 } else {
-                    log::warn!("[EndpointLocalTrack] view but not in room");
+                    log::warn!("[EndpointLocalTrack] track {} view but not in room", self.kind);
                     self.queue
                         .push_back(Output::RpcRes(req_id, EndpointLocalTrackRes::Attach(Err(RpcError::new2(EndpointErrors::EndpointNotInRoom)))));
                 }

packages/media_runner/src/worker.rs

@@ -314,14 +314,14 @@ impl MediaServerWorker {
                     Err(e) => self.queue.push_back(Output::ExtRpc(req_id, RpcRes::Webrtc(webrtc::RpcRes::Connect(Err(e))))),
                 },
                 webrtc::RpcReq::RemoteIce(conn, ice) => {
-                    log::info!("on rcp request {req_id}, webrtc::RpcReq::RemoteIce");
+                    log::info!("on rpc request {req_id}, webrtc::RpcReq::RemoteIce");
                     self.media_webrtc.input(&mut self.switcher).on_event(
                         now,
                         GroupInput::Ext(conn.into(), transport_webrtc::ExtIn::RemoteIce(req_id, transport_webrtc::Variant::Webrtc, ice.candidates)),
                     );
                 }
                 webrtc::RpcReq::RestartIce(conn, ip, token, user_agent, req) => {
-                    log::info!("on rcp request {req_id}, webrtc::RpcReq::RestartIce");
+                    log::info!("on rpc request {req_id}, webrtc::RpcReq::RestartIce");
                     self.media_webrtc.input(&mut self.switcher).on_event(
                         now,
                         GroupInput::Ext(conn.into(), transport_webrtc::ExtIn::RestartIce(req_id, transport_webrtc::Variant::Webrtc, ip, token, user_agent, req)),

packages/transport_webrtc/src/lib.rs

@@ -15,4 +15,5 @@ pub enum WebrtcError {
     RpcTrackNameNotFound = 0x2003,
     RpcTrackNotAttached = 0x2004,
     RpcTrackAlreadyAttached = 0x2005,
+    RpcEndpointNotFound = 0x2006,
 }

packages/transport_webrtc/src/transport.rs

@@ -23,7 +23,7 @@ use sans_io_runtime::{
 use str0m::{
     bwe::Bitrate,
     change::{DtlsCert, SdpOffer},
-    channel::ChannelId,
+    channel::{ChannelConfig, ChannelId},
     format::CodecConfig,
     ice::IceCreds,
     media::{KeyframeRequestKind, Mid},
@@ -128,20 +128,30 @@ impl TransportWebrtc {
         let ice_ufrag = rtc_config.local_ice_credentials().as_ref().expect("should have ice credentials").ufrag.clone();
 
         let mut rtc = rtc_config.build();
-        rtc.direct_api().enable_twcc_feedback();
+        let mut internal: Box<dyn TransportWebrtcInternal> = match variant {
+            VariantParams::Whip(room, peer) => Box::new(whip::TransportWebrtcWhip::new(room, peer)),
+            VariantParams::Whep(room, peer) => Box::new(whep::TransportWebrtcWhep::new(room, peer)),
+            VariantParams::Webrtc(_ip, _token, _user_agent, req) => {
+                rtc.direct_api().create_data_channel(ChannelConfig {
+                    label: "data".to_string(),
+                    negotiated: Some(1000),
+                    ..Default::default()
+                });
+                //we need to start sctp as client side for handling restart-ice in new server
+                //if not, datachannel will not connect successful after reconnect to new server
+                rtc.direct_api().start_sctp(true);
+                Box::new(webrtc::TransportWebrtcSdk::new(req))
+            }
+        };
 
+        rtc.direct_api().enable_twcc_feedback();
         let mut ports = Small2dMap::default();
         for (local_addr, slot) in local_addrs {
             ports.insert(local_addr, slot);
             rtc.add_local_candidate(Candidate::host(local_addr, Protocol::Udp).expect("Should add local candidate"));
         }
         let answer = rtc.sdp_api().accept_offer(offer).map_err(|_e| RpcError::new2(WebrtcError::InternalServerError))?;
         let mut local_convert = LocalMediaConvert::default();
-        let mut internal: Box<dyn TransportWebrtcInternal> = match variant {
-            VariantParams::Whip(room, peer) => Box::new(whip::TransportWebrtcWhip::new(room, peer)),
-            VariantParams::Whep(room, peer) => Box::new(whep::TransportWebrtcWhep::new(room, peer)),
-            VariantParams::Webrtc(_ip, _token, _user_agent, req) => Box::new(webrtc::TransportWebrtcSdk::new(req)),
-        };
         internal.on_codec_config(rtc.codec_config());
         local_convert.set_config(rtc.codec_config());
 
@@ -207,7 +217,7 @@ impl TransportWebrtc {
                 }
             }
             InternalOutput::Str0mResetBwe(init_bitrate) => {
-                log::info!("Reset str0m bwe to init_bitrate {init_bitrate} bps");
+                log::info!("[TransportWebrtc] Reset str0m bwe to init_bitrate {init_bitrate} bps");
                 self.rtc.bwe().reset(init_bitrate.into());
             }
             InternalOutput::TransportOutput(out) => {

packages/transport_webrtc/src/transport/webrtc.rs

@@ -372,14 +372,18 @@ impl TransportWebrtcInternal for TransportWebrtcSdk {
                 ))));
             }
             Str0mEvent::StreamPaused(event) => {
+                // We need to map media ssrc here for avoiding unknown pkt
+                // without it, sometime we will failed to restore session from restart-ice
+                self.media_convert.get_mid(event.ssrc, Some(event.mid));
+
                 let track = return_if_none!(self.remote_track_by_mid(event.mid)).name().to_string();
                 let status = if event.paused {
                     ProtoSenderStatus::Inactive
                 } else {
                     ProtoSenderStatus::Active
                 };
 
-                log::info!("[TransportWebrtcSdk] track {track} set status {:?}", status);
+                log::info!("[TransportWebrtcSdk] track {track} mid {} ssrc {} set status {:?}", event.mid, event.ssrc, status);
                 self.send_event(ProtoServerEvent::Sender(ProtoSenderEventContainer {
                     name: track,
                     event: Some(ProtoSenderEvent::State(ProtoSenderState { status: status as i32 })),

packages/transport_webrtc/src/worker.rs

@@ -4,7 +4,7 @@ use media_server_core::{
     cluster::{ClusterEndpointControl, ClusterEndpointEvent, ClusterRoomHash},
     endpoint::{Endpoint, EndpointCfg, EndpointInput, EndpointOutput},
 };
-use media_server_protocol::transport::RpcResult;
+use media_server_protocol::transport::{RpcError, RpcResult};
 use sans_io_runtime::{
     backend::{BackendIncoming, BackendOutgoing},
     group_owner_type, group_task, return_if_none, return_if_some, TaskSwitcher, TaskSwitcherChild,
@@ -14,6 +14,7 @@ use str0m::change::DtlsCert;
 use crate::{
     shared_port::SharedUdpPort,
     transport::{ExtIn, ExtOut, TransportWebrtc, VariantParams},
+    WebrtcError,
 };
 
 group_task!(Endpoints, Endpoint<TransportWebrtc, ExtIn, ExtOut>, EndpointInput<ExtIn>, EndpointOutput<ExtOut>);
@@ -118,7 +119,24 @@ impl MediaWorkerWebrtc {
             }
             GroupInput::Ext(owner, ext) => {
                 log::info!("[MediaWorkerWebrtc] on ext to owner {:?}", owner);
-                self.endpoints.on_event(now, owner.index(), EndpointInput::Ext(ext));
+                if let Some(&Some(_)) = self.endpoints.tasks.get(owner.index()) {
+                    self.endpoints.on_event(now, owner.index(), EndpointInput::Ext(ext));
+                } else {
+                    match ext {
+                        ExtIn::RemoteIce(req_id, variant, ..) => {
+                            self.queue
+                                .push_back(GroupOutput::Ext(owner, ExtOut::RemoteIce(req_id, variant, Err(RpcError::new2(WebrtcError::RpcEndpointNotFound)))));
+                        }
+                        ExtIn::RestartIce(req_id, variant, remote, useragent, token, req) => {
+                            if let Ok((ice_lite, sdp, index)) = self.spawn(VariantParams::Webrtc(remote, useragent, token, req.clone()), &req.sdp) {
+                                self.queue.push_back(GroupOutput::Ext(index.into(), ExtOut::RestartIce(req_id, variant, Ok((ice_lite, sdp)))));
+                            } else {
+                                self.queue
+                                    .push_back(GroupOutput::Ext(owner, ExtOut::RestartIce(req_id, variant, Err(RpcError::new2(WebrtcError::RpcEndpointNotFound)))));
+                            }
+                        }
+                    }
+                }
             }
             GroupInput::Close(owner) => {
                 self.endpoints.on_event(now, owner.index(), EndpointInput::Close);