o_sfu_protocol/core/
request_flow.rs

1use super::{
2    Command, Commands, FlushMode, NegotiationKind, NegotiationRejection, PendingRequest,
3    PendingRequestKind, ProtocolCore, REQUEST_TIMEOUT_MS, close_for_protocol_error,
4};
5use crate::signaling::{
6    ClientEnvelope, ClientRequest, ClientResponse, RecordingOptions, RequestId, ServerRequest,
7    ServerResponse, SessionDescriptionPayload,
8};
9
10impl ProtocolCore {
11    /// Requests partial recording changes through the legacy `startrecording` wire.
12    ///
13    /// Existing servers refuse this request. Future backends must preserve its
14    /// partial-update semantics for released bundles or negotiate a replacement.
15    pub fn start_recording(&mut self, options: RecordingOptions) -> Vec<Command> {
16        begin_request(self, ClientRequest::StartRecording(options))
17    }
18
19    pub fn stop_recording(&mut self) -> Vec<Command> {
20        begin_request(self, ClientRequest::StopRecording)
21    }
22
23    /// Replies to the currently pending negotiation request.
24    ///
25    /// The host must echo the exact `request_id` and `kind` from
26    /// [`Command::ApplyNegotiation`]. Mismatches are ignored so a stale or
27    /// reordered SDP answer cannot accidentally resolve the wrong negotiation.
28    pub fn submit_negotiation_answer(
29        &mut self,
30        request_id: &RequestId,
31        kind: NegotiationKind,
32        sdp: impl Into<String>,
33    ) -> Vec<Command> {
34        if !self.phase.resolve_negotiation(request_id, kind) {
35            return Vec::new();
36        }
37        let payload = SessionDescriptionPayload {
38            sdp: sdp.into(),
39            upload_slots: Vec::new(),
40        };
41        let response = match kind {
42            NegotiationKind::Offer => ClientResponse::Offer(payload),
43            NegotiationKind::Renegotiate => ClientResponse::Renegotiate(payload),
44        };
45        let Some(envelope) = ClientEnvelope::Response {
46            response_to: request_id.clone(),
47            response,
48        }
49        .into_envelope()
50        .ok() else {
51            return Vec::new();
52        };
53        self.outbound_batch.enqueue(envelope, FlushMode::Immediate)
54    }
55}
56
57pub(super) fn handle_server_request(
58    core: &mut ProtocolCore,
59    request_id: RequestId,
60    request: ServerRequest,
61) -> Commands {
62    match request {
63        ServerRequest::Offer(payload) => {
64            handle_negotiation_request(core, request_id, NegotiationKind::Offer, payload)
65        }
66        ServerRequest::Renegotiate(payload) => {
67            handle_negotiation_request(core, request_id, NegotiationKind::Renegotiate, payload)
68        }
69    }
70}
71
72pub(super) fn handle_server_response(
73    core: &mut ProtocolCore,
74    response_to: &RequestId,
75    response: ServerResponse,
76) -> Commands {
77    match response {
78        ServerResponse::StartRecording(payload) => core.request_tracker.resolve_response(
79            response_to,
80            PendingRequestKind::StartRecording,
81            payload.ok,
82        ),
83        ServerResponse::StopRecording(payload) => core.request_tracker.resolve_response(
84            response_to,
85            PendingRequestKind::StopRecording,
86            payload.ok,
87        ),
88    }
89}
90
91fn handle_negotiation_request(
92    core: &mut ProtocolCore,
93    request_id: RequestId,
94    kind: NegotiationKind,
95    payload: SessionDescriptionPayload,
96) -> Commands {
97    match core.phase.accept_negotiation(&request_id, kind) {
98        Ok(()) => {}
99        Err(NegotiationRejection::Ignored) => return Vec::new(),
100        Err(NegotiationRejection::ProtocolError) => {
101            return close_for_protocol_error();
102        }
103    }
104    vec![Command::ApplyNegotiation {
105        request_id,
106        kind,
107        sdp: payload.sdp,
108        upload_slots: payload.upload_slots,
109    }]
110}
111
112fn begin_request(core: &mut ProtocolCore, request: ClientRequest) -> Commands {
113    if !core.phase.can_send_client_messages() {
114        return Vec::new();
115    }
116    let kind = match &request {
117        ClientRequest::StartRecording(_) => PendingRequestKind::StartRecording,
118        ClientRequest::StopRecording => PendingRequestKind::StopRecording,
119    };
120    let request_start = core.request_tracker.begin(kind);
121    let request_id = request_start.request_id;
122    let pending_request = PendingRequest {
123        request_id: request_id.clone(),
124        timeout_timer_id: request_start.timeout_timer_id.raw(),
125        timeout_ms: REQUEST_TIMEOUT_MS,
126    };
127    let Some(envelope) = ClientEnvelope::Request {
128        request_id,
129        request,
130    }
131    .into_envelope()
132    .ok() else {
133        return Vec::new();
134    };
135    let mut commands = vec![Command::BeginPendingRequest {
136        request: pending_request,
137    }];
138    commands.extend(core.outbound_batch.enqueue(envelope, FlushMode::Batched));
139    commands
140}
o_sfu_protocol/core/request_flow.rs

o_sfu_protocol/core/
request_flow.rs
