@@ -11,11 +11,14 @@ pub mod whip;
1111
1212mod mux;
1313
14+ use std:: collections:: HashMap ;
1415use std:: net:: SocketAddr ;
15- use std:: sync:: Arc ;
16+ use std:: sync:: { Arc , Mutex } ;
1617
1718use axum:: Router ;
18- use tokio:: sync:: OnceCell ;
19+ use axum:: extract:: { Path , State } ;
20+ use axum:: http:: StatusCode ;
21+ use tokio:: sync:: { OnceCell , oneshot} ;
1922
2023use crate :: Result ;
2124use mux:: Mux ;
@@ -76,6 +79,10 @@ struct Inner {
7679 /// The shared media socket + demux, bound lazily on the first accept so
7780/// `Server::new` can stay synchronous (and an idle server binds no port).
7881mux : OnceCell < Mux > ,
82+ /// Live sessions keyed by resource id, each holding a cancel sender the
83+ /// session task selects on. Lets [`Server::terminate`] (and the bundled
84+ /// `DELETE` route) end a session by its `Location` id.
85+ sessions : Mutex < HashMap < String , oneshot:: Sender < ( ) > > > ,
7986}
8087
8188impl Server {
@@ -88,6 +95,7 @@ impl Server {
8895 publisher,
8996 subscriber,
9097 mux : OnceCell :: new ( ) ,
98+ sessions : Mutex :: new ( HashMap :: new ( ) ) ,
9199 } ) ,
92100 }
93101 }
@@ -129,4 +137,77 @@ impl Server {
129137 pub ( crate ) fn subscriber ( & self ) -> & moq_net:: OriginConsumer {
130138 & self . inner . subscriber
131139 }
140+
141+ /// Register a session under its resource id, returning the cancel receiver
142+ /// the session task selects on. Called by [`whip::accept`] / [`whep::accept`]
143+ /// before spawning the session.
144+ pub ( crate ) fn register_session ( & self , resource_id : String ) -> oneshot:: Receiver < ( ) > {
145+ let ( tx, rx) = oneshot:: channel ( ) ;
146+ self . inner . sessions . lock ( ) . unwrap ( ) . insert ( resource_id, tx) ;
147+ rx
148+ }
149+
150+ /// Drop a session's registry entry once it has ended on its own.
151+ pub ( crate ) fn unregister_session ( & self , resource_id : & str ) {
152+ self . inner . sessions . lock ( ) . unwrap ( ) . remove ( resource_id) ;
153+ }
154+
155+ /// Terminate a negotiated session by its resource id (the `Location` path
156+ /// component from the WHIP/WHEP response). Returns `true` if a live session
157+ /// was found and signalled to stop; the session task then releases its
158+ /// broadcast announcement and mux registration. Embedders that own their own
159+ /// HTTP routing call this to honor a WHIP/WHEP `DELETE`; the bundled routers
160+ /// already wire it to the `DELETE` method.
161+ pub fn terminate ( & self , resource_id : & str ) -> bool {
162+ if let Some ( cancel) = self . inner . sessions . lock ( ) . unwrap ( ) . remove ( resource_id) {
163+ let _ = cancel. send ( ( ) ) ;
164+ true
165+ } else {
166+ false
167+ }
168+ }
169+ }
170+
171+ /// Shared `DELETE` handler for both bundled routers: parse the resource id from
172+ /// the trailing path segment and terminate the matching session.
173+ pub ( crate ) async fn delete ( State ( server) : State < Server > , Path ( path) : Path < String > ) -> StatusCode {
174+ match crate :: sdp:: parse_resource_id ( & path) {
175+ Ok ( id) if server. terminate ( & id. to_string ( ) ) => StatusCode :: OK ,
176+ Ok ( _) => StatusCode :: NOT_FOUND ,
177+ Err ( _) => StatusCode :: BAD_REQUEST ,
178+ }
179+ }
180+
181+ #[ cfg( test) ]
182+ mod tests {
183+ use super :: * ;
184+
185+ fn server ( ) -> Server {
186+ let publisher = moq_net:: Origin :: random ( ) . produce ( ) ;
187+ let subscriber = moq_net:: Origin :: random ( ) . produce ( ) . consume ( ) ;
188+ Server :: new ( Config :: default ( ) , publisher, subscriber)
189+ }
190+
191+ #[ test]
192+ fn terminate_unknown_session_is_false ( ) {
193+ assert ! ( !server( ) . terminate( "00000000-0000-0000-0000-000000000000" ) ) ;
194+ }
195+
196+ #[ test]
197+ fn terminate_registered_session_once ( ) {
198+ let server = server ( ) ;
199+ let id = "11111111-1111-1111-1111-111111111111" ;
200+ let _cancel = server. register_session ( id. to_string ( ) ) ;
201+ assert ! ( server. terminate( id) , "first terminate finds the session" ) ;
202+ assert ! ( !server. terminate( id) , "second terminate is a no-op" ) ;
203+ }
204+
205+ #[ test]
206+ fn unregister_drops_the_entry ( ) {
207+ let server = server ( ) ;
208+ let id = "22222222-2222-2222-2222-222222222222" ;
209+ let _cancel = server. register_session ( id. to_string ( ) ) ;
210+ server. unregister_session ( id) ;
211+ assert ! ( !server. terminate( id) , "unregistered session can't be terminated" ) ;
212+ }
132213}
0 commit comments