Can stop streaming

src/api_data/call_peer_interrupted_streaming.rs

@@ -0,0 +1,24 @@
+//! # Call peer interrupted streaming
+//!
+//! @author Pierre Hubert
+
+use serde::Serialize;
+
+use crate::data::conversation::ConvID;
+use crate::data::user::UserID;
+
+#[derive(Serialize)]
+#[allow(non_snake_case)]
+pub struct CallPeerInterruptedStreamingAPI {
+    callID: u64,
+    peerID: u64,
+}
+
+impl CallPeerInterruptedStreamingAPI {
+    pub fn new(call_id: &ConvID, peer_id: &UserID) -> Self {
+        Self {
+            callID: call_id.clone(),
+            peerID: peer_id.id(),
+        }
+    }
+}

src/api_data/mod.rs

@@ -62,4 +62,5 @@ pub mod joined_call_message;
 pub mod call_member_info;
 pub mod left_call_message;
 pub mod new_call_signal;
-pub mod call_peer_ready;
+pub mod call_peer_ready;
+pub mod call_peer_interrupted_streaming;

src/controllers/calls_controller.rs

@@ -7,6 +7,7 @@ use std::collections::HashMap;
 use webrtc_sdp::attribute_type::SdpAttribute;
 
 use crate::api_data::call_member_info::CallMemberInfo;
+use crate::api_data::call_peer_interrupted_streaming::CallPeerInterruptedStreamingAPI;
 use crate::api_data::call_peer_ready::CallPeerReadyAPI;
 use crate::api_data::joined_call_message::JoinedCallMessage;
 use crate::api_data::left_call_message::LeftCallMessage;
@@ -15,7 +16,7 @@ use crate::api_data::user_calls_config::UserCallsConfig;
 use crate::controllers::routes::RequestResult;
 use crate::controllers::user_ws_controller;
 use crate::data::base_request_handler::BaseRequestHandler;
-use crate::data::call_signal::{CallSignal, IceCandidate, NewUserCallSignal, SdpType, UserCallOfferRequest};
+use crate::data::call_signal::{CallSignal, CloseCallStream, IceCandidate, NewUserCallSignal, SdpType, UserCallOfferRequest};
 use crate::data::config::conf;
 use crate::data::conversation::ConvID;
 use crate::data::error::{ExecError, Res};
@@ -63,6 +64,20 @@ impl UserWsRequestHandler {
         Ok(peer_id)
     }
 
+    /// Get call information, if available
+    fn call_info(&mut self, call_id: &ConvID) -> Res<ActiveCall> {
+        let call = self.get_conn()
+            .active_call
+            .clone()
+            .ok_or(ExecError::new("No call found !"))?;
+
+        if &call.conv_id != call_id {
+            self.bad_request("The call active is not the one requested!".to_string())?;
+        }
+
+        Ok(call)
+    }
+
     /// Update call information
     fn update_call<F>(&mut self, update: F) -> Res where F: FnOnce(&mut ActiveCall) {
         self.update_conn(|conn| {
@@ -309,6 +324,32 @@ pub fn request_offer(r: &mut UserWsRequestHandler) -> Res {
     r.success("Request sent")
 }
 
+/// Notify the user stopped to stream
+pub fn stop_streaming(r: &mut UserWsRequestHandler) -> Res {
+    let call_id = r.post_call_id("callID")?;
+    let user_id = r.user_id()?;
+
+    // Propagate notification only if required
+    if r.call_info(&call_id)?.ready {
+        r.update_call(|c| c.ready = false)?;
+
+        // Notify all other users
+        user_ws_controller::send_message_to_specific_connections(
+            |c| c.is_having_call_with_conversation(&call_id) && c.user_id != &user_id,
+            |_| UserWsMessage::no_id_message("call_peer_interrupted_streaming", CallPeerInterruptedStreamingAPI::new(&call_id, &user_id)),
+            None::<fn(&_) -> _>,
+        )?;
+    }
+
+    // Notify proxy
+    events_helper::propagate_event(&Event::CloseCallStream(&CloseCallStream {
+        call_hash: gen_call_hash(&call_id, &user_id),
+        peer_id: None,
+    }))?;
+
+    r.success("ok")
+}
+
 
 /// Make the user leave the call
 pub fn make_user_leave_call(conv_id: &ConvID, connection: &UserWsConnection) -> Res {

src/controllers/rtc_relay_controller.rs

@@ -68,6 +68,8 @@ struct OfferRequest {
     peerId: String,
 }
 
+type CloseCallStream = OfferRequest;
+
 /// Current WebSocket connection
 static mut ACTIVE_RTC_CONNECTION: Option<Addr<RtcRelayActor>> = None;
 
@@ -293,6 +295,17 @@ pub fn handle_event(e: &events_helper::Event) -> Res {
             })?;
         }
 
+        Event::CloseCallStream(request) => {
+            send_message_to_relay(CloseCallStream {
+                title: "close_conn".to_string(),
+                callHash: request.call_hash.to_string(),
+                peerId: match &request.peer_id {
+                    None => "0".to_string(),
+                    Some(id) => id.id().to_string()
+                },
+            })?;
+        }
+
         _ => {}
     }
 

src/controllers/user_ws_routes.rs

@@ -45,6 +45,7 @@ pub fn get_user_ws_routes() -> Vec<UserWsRoute> {
         UserWsRoute::new("calls/signal", calls_controller::on_client_signal),
         UserWsRoute::new("calls/mark_ready", calls_controller::mark_user_ready),
         UserWsRoute::new("calls/request_offer", calls_controller::request_offer),
+        UserWsRoute::new("calls/stop_streaming", calls_controller::stop_streaming),
     ]
 }
 

src/data/call_signal.rs

@@ -48,6 +48,12 @@ pub struct UserCallOfferRequest {
     pub user_id: UserID,
 }
 
+/// Close call stream
+pub struct CloseCallStream {
+    pub call_hash: String,
+    pub peer_id: Option<UserID>,
+}
+
 impl SdpType {
     pub fn from_str(val: &str) -> Res<SdpType> {
         match val {

src/helpers/events_helper.rs

@@ -6,7 +6,7 @@
 
 use crate::controllers::{calls_controller, comments_controller, conversations_controller, notifications_controller, rtc_relay_controller, user_ws_controller};
 use crate::data::api_client::APIClient;
-use crate::data::call_signal::{NewRtcRelayMessage, NewUserCallSignal, UserCallOfferRequest};
+use crate::data::call_signal::{CloseCallStream, NewRtcRelayMessage, NewUserCallSignal, UserCallOfferRequest};
 use crate::data::comment::Comment;
 use crate::data::conversation::ConvID;
 use crate::data::conversation_message::ConversationMessage;
@@ -65,6 +65,9 @@ pub enum Event<'a> {
     /// User requested an offer for a call
     UserRequestedCallOffer(&'a UserCallOfferRequest),
 
+    /// Close call stream
+    CloseCallStream(&'a CloseCallStream),
+
     /// No event
     None,
 }