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
10pub(super) fn start_recording(core: &mut ProtocolCore, options: RecordingOptions) -> Commands {
11    begin_request(
12        core,
13        ClientRequest::StartRecording(options),
14        PendingRequestKind::StartRecording,
15    )
16}
17
18pub(super) fn stop_recording(core: &mut ProtocolCore) -> Commands {
19    begin_request(
20        core,
21        ClientRequest::StopRecording,
22        PendingRequestKind::StopRecording,
23    )
24}
25
26pub(super) fn submit_negotiation_answer(
27    core: &mut ProtocolCore,
28    request_id: &RequestId,
29    kind: NegotiationKind,
30    sdp: impl Into<String>,
31) -> Commands {
32    if !core.can_send_client_messages() || !core.phase.resolve_negotiation(request_id, kind) {
33        return Vec::new();
34    }
35    let sdp = sdp.into();
36    let response = match kind {
37        NegotiationKind::Offer => ClientResponse::Offer(SessionDescriptionPayload {
38            sdp,
39            upload_slots: Vec::new(),
40        }),
41        NegotiationKind::Renegotiate => ClientResponse::Renegotiate(SessionDescriptionPayload {
42            sdp,
43            upload_slots: Vec::new(),
44        }),
45    };
46    let Some(envelope) = ClientEnvelope::Response {
47        response_to: request_id.clone(),
48        response,
49    }
50    .into_envelope()
51    .ok() else {
52        return Vec::new();
53    };
54    core.enqueue_envelope(envelope, FlushMode::Immediate)
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) => resolve_request(
79            core,
80            response_to,
81            PendingRequestKind::StartRecording,
82            payload.ok,
83        ),
84        ServerResponse::StopRecording(payload) => resolve_request(
85            core,
86            response_to,
87            PendingRequestKind::StopRecording,
88            payload.ok,
89        ),
90    }
91}
92
93fn handle_negotiation_request(
94    core: &mut ProtocolCore,
95    request_id: RequestId,
96    kind: NegotiationKind,
97    payload: SessionDescriptionPayload,
98) -> Commands {
99    match core.phase.accept_negotiation(&request_id, kind) {
100        Ok(()) => {}
101        Err(NegotiationRejection::Ignored) => return Vec::new(),
102        Err(NegotiationRejection::ProtocolError) => {
103            return close_for_protocol_error();
104        }
105    }
106    vec![Command::ApplyNegotiation {
107        request_id,
108        kind,
109        sdp: payload.sdp,
110        upload_slots: payload.upload_slots,
111    }]
112}
113
114fn begin_request(
115    core: &mut ProtocolCore,
116    request: ClientRequest,
117    kind: PendingRequestKind,
118) -> Commands {
119    if !core.can_send_client_messages() {
120        return Vec::new();
121    }
122    let Some(request_start) = core.request_tracker.try_begin(kind) else {
123        return Vec::new();
124    };
125    let request_id = request_start.request_id;
126    let pending_request = PendingRequest {
127        request_id: request_id.clone(),
128        timeout_timer_id: request_start.timeout_timer_id.raw(),
129        timeout_ms: REQUEST_TIMEOUT_MS,
130    };
131    let Some(envelope) = ClientEnvelope::Request {
132        request_id,
133        request,
134    }
135    .into_envelope()
136    .ok() else {
137        return Vec::new();
138    };
139
140    let mut commands = vec![Command::BeginPendingRequest {
141        request: pending_request,
142    }];
143    commands.extend(core.enqueue_envelope(envelope, FlushMode::Batched));
144    commands
145}
146
147fn resolve_request(
148    core: &mut ProtocolCore,
149    response_to: &RequestId,
150    expected_kind: PendingRequestKind,
151    ok: bool,
152) -> Commands {
153    core.request_tracker
154        .resolve_response(response_to, expected_kind, ok)
155}
o_sfu_protocol/core/request_flow.rs

o_sfu_protocol/core/
request_flow.rs
