Function attach_control_data_channel

packages/server/src/transport/webrtc.rs:494–566

(
    channel: Arc<RTCDataChannel>,
    session: crate::simulators::session::SimulatorSession,
    state: AppState,
    udid: String,
    stream_control_tx: mpsc::UnboundedSender<WebRtcStreamCommand>,

}

fn attach_control_data_channel(
    channel: Arc<RTCDataChannel>,
    session: crate::simulators::session::SimulatorSession,
    state: AppState,
    udid: String,
    stream_control_tx: mpsc::UnboundedSender<WebRtcStreamCommand>,
) {
    let (control_tx, control_rx) = mpsc::unbounded_channel::<ControlMessage>();
    task::spawn(run_webrtc_control_queue(
        session.clone(),
        state.clone(),
        udid.clone(),
        stream_control_tx.clone(),
        control_rx,
    ));
    channel.on_message(Box::new(move |message: DataChannelMessage| {
        let session = session.clone();
        let state = state.clone();
        let udid = udid.clone();
        let stream_control_tx = stream_control_tx.clone();
        let control_tx = control_tx.clone();
        Box::pin(async move {
            let Ok(text) = std::str::from_utf8(&message.data) else {
                warn!("Invalid WebRTC control message bytes for {udid}");
                return;
            };
            if let Ok(message) = serde_json::from_str::<WebRtcDataChannelMessage>(text) {
                match message {
                    WebRtcDataChannelMessage::ClientStats { stats } => {
                        if !stats.client_id.trim().is_empty() && !stats.kind.trim().is_empty() {
                            apply_stream_client_foreground_from_stats(&state, &stats);
                            state.metrics.record_client_stream_stats(*stats);
                        }
                    }
                    WebRtcDataChannelMessage::StreamControl {
                        client_id,
                        force_keyframe,
                        foreground,
                        snapshot,
                    } => {
                        apply_stream_client_foreground(&state, &session, &client_id, foreground);
                        let command = WebRtcStreamCommand {
                            force_keyframe: force_keyframe.unwrap_or(false),
                            snapshot: snapshot.unwrap_or(false),
                        };
                        if command.force_keyframe || command.snapshot {
                            session.request_keyframe();
                        }
                        let _ = stream_control_tx.send(command);
                    }
                    WebRtcDataChannelMessage::StreamQuality { config } => {
                        if let Err(error) = apply_stream_quality_payload(&state, &config) {
                            warn!("WebRTC stream quality update failed for {udid}: {error}");
                        } else {
                            session.request_keyframe();
                        }
                    }
                }