feat: graceful disconnect with whip, whep, sdk

packages/media_core/src/endpoint.rs

@@ -171,7 +171,6 @@ pub enum EndpointInput<Ext> {
     Net(BackendIncoming),
     Cluster(ClusterEndpointEvent),
     Ext(Ext),
-    Close,
 }
 
 pub enum EndpointOutput<Ext> {
@@ -238,14 +237,11 @@ where
             EndpointInput::Cluster(event) => {
                 self.internal.input(&mut self.switcher).on_cluster_event(now, event);
             }
-            EndpointInput::Close => {
-                self.transport.input(&mut self.switcher).on_input(now, TransportInput::Close);
-            }
         }
     }
 
     fn on_shutdown(&mut self, now: Instant) {
-        self.transport.input(&mut self.switcher).on_input(now, TransportInput::Close);
+        self.transport.input(&mut self.switcher).on_input(now, TransportInput::SystemClose);
     }
 }
 

packages/media_core/src/transport.rs

@@ -93,7 +93,7 @@ pub enum TransportInput<Ext> {
     Endpoint(EndpointEvent),
     RpcRes(EndpointReqId, EndpointRes),
     Ext(Ext),
-    Close,
+    SystemClose,
 }
 
 /// This is event from transport, in general is is result of transport protocol

packages/media_runner/src/worker.rs

@@ -441,6 +441,11 @@ impl<ES: 'static + MediaEdgeSecure> MediaServerWorker<ES> {
                         )
                     }))),
                 ),
+                transport_webrtc::ExtOut::Disconnect(req_id, variant, res) => match variant {
+                    transport_webrtc::Variant::Whip => Output::ExtRpc(req_id, RpcRes::Whip(whip::RpcRes::Delete(res.map(|_| WhipDeleteRes {})))),
+                    transport_webrtc::Variant::Whep => Output::ExtRpc(req_id, RpcRes::Whep(whep::RpcRes::Delete(res.map(|_| WhepDeleteRes {})))),
+                    transport_webrtc::Variant::Webrtc => Output::ExtRpc(req_id, RpcRes::Webrtc(webrtc::RpcRes::Delete(res))),
+                },
             },
             transport_webrtc::GroupOutput::Shutdown(_session) => Output::Continue,
             transport_webrtc::GroupOutput::Continue => Output::Continue,
@@ -454,6 +459,7 @@ impl<ES: 'static + MediaEdgeSecure> MediaServerWorker<ES> {
         match req {
             RpcReq::Whip(req) => match req {
                 whip::RpcReq::Connect(req) => {
+                    log::info!("on rpc request {req_id}, whip::RpcReq::Connect");
                     match self
                         .media_webrtc
                         .input(&mut self.switcher)
@@ -471,13 +477,15 @@ impl<ES: 'static + MediaEdgeSecure> MediaServerWorker<ES> {
                     );
                 }
                 whip::RpcReq::Delete(req) => {
-                    //TODO check error instead of auto response ok
-                    self.queue.push_back(Output::ExtRpc(req_id, RpcRes::Whip(whip::RpcRes::Delete(Ok(WhipDeleteRes {})))));
-                    self.media_webrtc.input(&mut self.switcher).on_event(now, GroupInput::Close(req.conn_id.into()));
+                    log::info!("on rpc request {req_id}, whip::RpcReq::Delete");
+                    self.media_webrtc
+                        .input(&mut self.switcher)
+                        .on_event(now, GroupInput::Ext(req.conn_id.into(), transport_webrtc::ExtIn::Disconnect(req_id, transport_webrtc::Variant::Whip)));
                 }
             },
             RpcReq::Whep(req) => match req {
                 whep::RpcReq::Connect(req) => {
+                    log::info!("on rpc request {req_id}, whep::RpcReq::Connect");
                     let peer_id = format!("whep-{}", random::<u64>());
                     match self
                         .media_webrtc
@@ -496,13 +504,15 @@ impl<ES: 'static + MediaEdgeSecure> MediaServerWorker<ES> {
                     );
                 }
                 whep::RpcReq::Delete(req) => {
-                    //TODO check error instead of auto response ok
-                    self.queue.push_back(Output::ExtRpc(req_id, RpcRes::Whep(whep::RpcRes::Delete(Ok(WhepDeleteRes {})))));
-                    self.media_webrtc.input(&mut self.switcher).on_event(now, GroupInput::Close(req.conn_id.into()));
+                    log::info!("on rpc request {req_id}, whep::RpcReq::Delete");
+                    self.media_webrtc
+                        .input(&mut self.switcher)
+                        .on_event(now, GroupInput::Ext(req.conn_id.into(), transport_webrtc::ExtIn::Disconnect(req_id, transport_webrtc::Variant::Whep)));
                 }
             },
             RpcReq::Webrtc(req) => match req {
                 webrtc::RpcReq::Connect(session_id, ip, user_agent, req, userdata, record) => {
+                    log::info!("on rpc request {req_id}, webrtc::RpcReq::Connect");
                     match self
                         .media_webrtc
                         .input(&mut self.switcher)
@@ -539,7 +549,12 @@ impl<ES: 'static + MediaEdgeSecure> MediaServerWorker<ES> {
                         ),
                     );
                 }
-                webrtc::RpcReq::Delete(_) => todo!(),
+                webrtc::RpcReq::Delete(conn) => {
+                    log::info!("on rpc request {req_id}, webrtc::RpcReq::Delete");
+                    self.media_webrtc
+                        .input(&mut self.switcher)
+                        .on_event(now, GroupInput::Ext(conn.into(), transport_webrtc::ExtIn::Disconnect(req_id, transport_webrtc::Variant::Webrtc)));
+                }
             },
         }
     }

packages/transport_webrtc/src/transport.rs

@@ -61,13 +61,15 @@ pub enum ExtIn {
     RemoteIce(u64, Variant, Vec<String>),
     /// Last option<string>, bool is userdata and record flag
     RestartIce(u64, Variant, IpAddr, String, ConnectRequest, Option<String>, bool),
+    Disconnect(u64, Variant),
 }
 
 #[derive(Debug, PartialEq, Eq)]
 pub enum ExtOut {
     RemoteIce(u64, Variant, RpcResult<u32>),
     /// response is (ice_lite, answer_sdp)
     RestartIce(u64, Variant, RpcResult<(bool, String)>),
+    Disconnect(u64, Variant, RpcResult<()>),
 }
 
 #[derive(Debug, PartialEq, Eq)]
@@ -323,9 +325,13 @@ impl<ES: 'static + MediaEdgeSecure> Transport<ExtIn, ExtOut> for TransportWebrtc
                             .push_back(TransportOutput::Ext(ExtOut::RestartIce(req_id, variant, Err(RpcError::new2(WebrtcError::InvalidSdp)))));
                     }
                 }
+                ExtIn::Disconnect(req_id, variant) => {
+                    self.internal.close(now);
+                    self.queue.push_back(TransportOutput::Ext(ExtOut::Disconnect(req_id, variant, Ok(()))));
+                }
             },
-            TransportInput::Close => {
-                log::info!("[TransportWebrtc] close request");
+            TransportInput::SystemClose => {
+                log::info!("[TransportWebrtc] system close request");
                 self.rtc.disconnect();
                 self.internal.close(now);
             }

packages/transport_webrtc/src/worker.rs

@@ -35,7 +35,6 @@ pub enum GroupInput {
     Net(BackendIncoming),
     Cluster(WebrtcSession, ClusterEndpointEvent),
     Ext(WebrtcSession, ExtIn),
-    Close(WebrtcSession),
 }
 
 #[derive(Debug)]
@@ -165,12 +164,13 @@ impl<ES: MediaEdgeSecure> MediaWorkerWebrtc<ES> {
                                     .push_back(GroupOutput::Ext(owner, ExtOut::RestartIce(req_id, variant, Err(RpcError::new2(WebrtcError::RpcEndpointNotFound)))));
                             }
                         }
+                        ExtIn::Disconnect(req_id, variant) => {
+                            self.queue
+                                .push_back(GroupOutput::Ext(owner, ExtOut::Disconnect(req_id, variant, Err(RpcError::new2(WebrtcError::RpcEndpointNotFound)))));
+                        }
                     }
                 }
             }
-            GroupInput::Close(owner) => {
-                self.endpoints.on_event(now, owner.index(), EndpointInput::Close);
-            }
         }
     }