init delete endpoint

Author: lherman-cs

demo/src/main.ts

@@ -7,6 +7,7 @@ const statusEl = document.getElementById("status") as HTMLSpanElement;
 
 let pc: RTCPeerConnection | null = null;
 let localStream: MediaStream | null = null;
+let sessionUrl: string | null = null;
 
 form.onsubmit = async (e) => {
   e.preventDefault();
@@ -47,7 +48,9 @@ async function start(endpoint: string) {
   pc.addTransceiver("audio", { direction: "recvonly" });
   const remoteStream = new MediaStream();
   remoteVideo.srcObject = remoteStream;
-  pc.ontrack = (e) => remoteStream.addTrack(e.track);
+  pc.ontrack = (e) => {
+    remoteStream.addTrack(e.track);
+  };
 
   pc.onconnectionstatechange = () => {
     statusEl.textContent = pc?.connectionState ?? "Disconnected";
@@ -61,10 +64,16 @@ async function start(endpoint: string) {
   });
   if (!response.ok) throw new Error(`request failed: ${response.status}`);
   const answerSdp = await response.text();
+  sessionUrl = response.headers.get("location");
   await pc.setRemoteDescription({ type: "answer", sdp: answerSdp });
 }
 
-function stop() {
+async function stop() {
+  if (sessionUrl) {
+    await fetch(sessionUrl, { method: "DELETE" });
+    sessionUrl = null;
+  }
+
   pc?.close();
   localStream?.getTracks().forEach(track => track.stop());
   localVideo.srcObject = null;

pulsebeam/src/controller.rs

@@ -36,6 +36,7 @@ pub enum ControllerMessage {
         String,
         oneshot::Sender<Result<String, ControllerError>>,
     ),
+    RemoveParticipant(Arc<RoomId>, Arc<ParticipantId>),
 }
 
 pub struct ControllerActor {
@@ -80,6 +81,15 @@ impl actor::Actor for ControllerActor {
             ControllerMessage::Allocate(room_id, participant_id, offer, resp) => {
                 let _ = resp.send(self.allocate(ctx, room_id, participant_id, offer).await);
             }
+
+            ControllerMessage::RemoveParticipant(room_id, participant_id) => {
+                if let Some(room_handle) = self.rooms.get(&room_id) {
+                    // if the room has exited, the participants have already cleaned up too.
+                    let _ = room_handle
+                        .send_high(room::RoomMessage::RemoveParticipant(participant_id))
+                        .await;
+                }
+            }
         }
     }
 }

pulsebeam/src/entity.rs

@@ -193,7 +193,8 @@ impl AsRef<str> for ExternalRoomId {
     }
 }
 
-#[derive(serde::Serialize, serde::Deserialize)]
+#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash, serde::Serialize, serde::Deserialize)]
+#[serde(try_from = "String")]
 pub struct ParticipantId {
     pub internal: EntityId,
 }
@@ -211,24 +212,6 @@ impl Default for ParticipantId {
     }
 }
 
-impl hash::Hash for ParticipantId {
-    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
-        self.internal.hash(state);
-    }
-}
-
-impl PartialOrd for ParticipantId {
-    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
-        Some(self.cmp(other))
-    }
-}
-
-impl Ord for ParticipantId {
-    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
-        self.internal.cmp(&other.internal)
-    }
-}
-
 impl TryFrom<String> for ParticipantId {
     type Error = IdValidationError;
 
@@ -251,14 +234,6 @@ impl TryFrom<String> for ParticipantId {
     }
 }
 
-impl Eq for ParticipantId {}
-
-impl PartialEq for ParticipantId {
-    fn eq(&self, other: &Self) -> bool {
-        self.internal.eq(&other.internal)
-    }
-}
-
 impl fmt::Display for ParticipantId {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         fmt::Display::fmt(&self.internal, f)

pulsebeam/src/node.rs

@@ -13,6 +13,7 @@ pub async fn run(
     // Configure CORS
     let cors = CorsLayer::very_permissive()
         .allow_origin(AllowOrigin::mirror_request())
+        .expose_headers([hyper::header::LOCATION])
         .max_age(Duration::from_secs(86400));
 
     // Spawn system and controller actors

pulsebeam/src/room.rs

@@ -19,6 +19,7 @@ const EMPTY_ROOM_TIMEOUT: Duration = Duration::from_secs(30);
 pub enum RoomMessage {
     PublishTrack(Arc<TrackMeta>),
     AddParticipant(Arc<ParticipantId>, Box<Rtc>),
+    RemoveParticipant(Arc<ParticipantId>),
 }
 
 #[derive(Clone, Debug)]
@@ -103,6 +104,12 @@ impl actor::Actor for RoomActor {
                 self.handle_participant_joined(ctx, participant_id, rtc)
                     .await
             }
+            RoomMessage::RemoveParticipant(participant_id) => {
+                if let Some(participant_handle) = self.state.participants.get(&participant_id) {
+                    // if it's closed, then the participant has exited
+                    let _ = participant_handle.handle.terminate().await;
+                }
+            }
             RoomMessage::PublishTrack(track_meta) => {
                 self.handle_track_published(track_meta).await;
             }

pulsebeam/src/signaling.rs

@@ -51,8 +51,6 @@ impl IntoResponse for SignalingError {
     }
 }
 
-/// Join a room / create participant
-/// POST /api/v1/rooms/{roomId}
 #[axum::debug_handler]
 async fn join_room(
     Path(room_id): Path<ExternalRoomId>,
@@ -78,23 +76,33 @@ async fn join_room(
         .await
         .map_err(|_| controller::ControllerError::ServiceUnavailable)??;
 
+    // TODO: remove hardcoded URI
     let location_url = format!(
-        "/api/v1/rooms/{}/participants/{}",
-        &room_id, &participant_id
+        "http://localhost:3000/api/v1/rooms/{}/participants/{}",
+        &room_id.external, &participant_id
     );
     let mut response_headers = HeaderMap::new();
     response_headers.insert(LOCATION, location_url.parse().unwrap());
     Ok((StatusCode::CREATED, response_headers, answer_sdp))
 }
 
-/// Leave a room / delete participant
-/// DELETE /api/v1/rooms/{roomId}/participants/{participantId}
 #[axum::debug_handler]
 async fn leave_room(
     Path((room_id, participant_id)): Path<(ExternalRoomId, ParticipantId)>,
-    State(_controller): State<controller::ControllerHandle>,
+    State(con): State<controller::ControllerHandle>,
 ) -> Result<impl IntoResponse, SignalingError> {
-    // TODO: terminate participant session
+    let room_id = RoomId::new(room_id);
+    let room_id = Arc::new(room_id);
+    let participant_id = Arc::new(participant_id);
+
+    // If controller has exited, there's no dangling participants and rooms
+    let _ = con
+        .send_high(controller::ControllerMessage::RemoveParticipant(
+            room_id,
+            participant_id,
+        ))
+        .await;
+
     Ok(StatusCode::NO_CONTENT)
 }