o_sfu/application/user_session/
client_input.rs

1use o_sfu_protocol::wire::{
2    ClientBroadcastPayload, ClientEnvelope, ClientMessage, ClientRequest, JsonPayload,
3    RecordingActionResult, RecordingOptions, RequestId, ServerEnvelope, ServerResponse, UserInfo,
4};
5
6use super::{User, UserError, UserOutput};
7
8impl User {
9    pub async fn apply_client_envelope(
10        &mut self,
11        envelope: ClientEnvelope,
12    ) -> Result<UserOutput, UserError> {
13        if let ClientEnvelope::Message(ClientMessage::Auth(_)) = &envelope {
14            return Err(UserError::ProtocolViolation);
15        }
16        self.reject_stale_connection().await?;
17        match envelope {
18            ClientEnvelope::Message(ClientMessage::Info(info)) => self.update_info(info).await,
19            ClientEnvelope::Message(ClientMessage::Broadcast(ClientBroadcastPayload {
20                message,
21            })) => self.broadcast(message).await,
22            ClientEnvelope::Message(ClientMessage::Subscribe(payload)) => {
23                self.subscribe(payload.user_id, payload.states).await
24            }
25            ClientEnvelope::Message(ClientMessage::Publish(payload)) => {
26                self.set_publication_active(payload.stream_type, true).await
27            }
28            ClientEnvelope::Message(ClientMessage::Unpublish(payload)) => {
29                self.set_publication_active(payload.stream_type, false)
30                    .await
31            }
32            ClientEnvelope::Response {
33                response_to,
34                response,
35            } => self.complete_negotiation(response_to, response).await,
36            ClientEnvelope::Request {
37                request_id,
38                request: ClientRequest::StartRecording(payload),
39            } => Ok(self.start_recording(request_id, payload).await),
40            ClientEnvelope::Request {
41                request_id,
42                request: ClientRequest::StopRecording,
43            } => Ok(self.stop_recording(request_id).await),
44            ClientEnvelope::Message(ClientMessage::Auth(_)) => Err(UserError::ProtocolViolation),
45        }
46    }
47
48    async fn update_info(&self, info: UserInfo) -> Result<UserOutput, UserError> {
49        self.media.session().update_info(info).await;
50        Ok(UserOutput::new())
51    }
52
53    async fn broadcast(&self, message: JsonPayload) -> Result<UserOutput, UserError> {
54        let result = self.media.session().broadcast(message).await;
55        result.map_err(|_error| UserError::ProtocolViolation)?;
56        Ok(UserOutput::new())
57    }
58
59    async fn start_recording(
60        &self,
61        request_id: RequestId,
62        options: RecordingOptions,
63    ) -> UserOutput {
64        let ok = self.media.session().start_recording(options).await;
65        vec![ServerEnvelope::Response {
66            response_to: request_id,
67            response: ServerResponse::StartRecording(RecordingActionResult { ok }),
68        }]
69    }
70
71    async fn stop_recording(&self, request_id: RequestId) -> UserOutput {
72        let ok = self.media.session().stop_recording().await;
73        vec![ServerEnvelope::Response {
74            response_to: request_id,
75            response: ServerResponse::StopRecording(RecordingActionResult { ok }),
76        }]
77    }
78}
o_sfu/application/user_session/client_input.rs

o_sfu/application/user_session/
client_input.rs
