You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: negotiate_finished.rs
    path of file: ./repos/MirrorX/mirrorx_core/src/api/endpoint/handlers
    
    the code of the file until where you have to start completion: use crate::{
    api::endpoint::{client::EndPointClient, message::EndPointMessage},
    component::{
        audio::{encoder::AudioEncoder, recorder::new_record_stream_and_rx},
        desktop::{monitor::get_active_monitors, Duplicator},
        video_encoder::{config::*, encoder::VideoEncoder},
    },
    error::CoreError,
};
use cpal::traits::StreamTrait;
use scopeguard::defer;
use std::sync::Arc;

pub struct NegotiateFinishedRequest {
    pub active_device_id: i64,
    pub passive_device_id: i64,
    pub expect_frame_rate: u8,
    pub texture_id: i64,
}

pub fn handle_negotiate_finished_request(client: Arc<EndPointClient>) {
    spawn_desktop_capture_and_encode_process(client.clone());
    spawn_audio_capture_and_encode_process(client);
}

#[cfg(target_os = "macos")]
fn spawn_desktop_capture_and_encode_process(client: Arc<EndPointClient>) {
    let (capture_frame_tx, mut capture_frame_rx) = tokio::sync::mpsc::channel(180);

    tokio::task::spawn_blocking(move || {
        tracing::info_span!("desktop_capture_and_encode_process", client = ?client);

        defer! {
            tracing::info!("desktop capture process exit");
        }

        let monitors = match get_active_monitors(false) {
            Ok(params) => params,
            Err(err) => {
                tracing::error!(?err, "get_primary_monitor_params failed");
                return;
            }
        };

        let mut encoder = match VideoEncoder::new(libx264::Libx264Config::default(), client.clone())
        {
            Ok(encoder) => encoder,
            Err(err) => {
                tracing::error!(?err, "initialize encoder failed");
                return;
            }
        };

        let primary_monitor = monitors.iter().find(|monitor| monitor.is_primary);

        let (duplicator, monitor_id) = match Duplicator::new(
            primary_monitor.map(|monitor| monitor.id.to_owned()),
            capture_frame_tx,
        ) {
            Ok(duplicator) => duplicator,
            Err(err) => {
                tracing::error!(?err, "initialize encoder failed");
                return;
            }
        };

        let select_monitor = match monitors
            .into_i
    
fn spawn_desktop_capture_and_encode_process(client: Arc<EndPointClient>) {
    let monitors = match get_active_monitors(false) {
        Ok(params) => params,
        Err(err) => {
            tracing::error!(?err, "get_active_monitors failed");
            return;
        }
    };

    let (capture_frame_tx, mut capture_frame_rx) = tokio::sync::mpsc::channel(180);

    tokio::task::spawn_blocking(move || {
        defer! {
            tracing::info!( "desktop capture process exit");
        }

        let primary_monitor = monitors.iter().find(|monitor| monitor.is_primary);

        let (mut duplicator, _) =
            match Duplicator::new(primary_monitor.map(|monitor| monitor.id.to_owned())) {
                Ok(duplicator) => duplicator,
                Err(err) => {
                    tracing::error!(?err, "initialize encoder failed");
                    return;
                }
            };

        loop {
            match duplicator.capture() {
                Ok(capture_frame) => {
                    if capture_frame_tx.blocking_send(capture_frame).is_err() {
                        return;
                    }
                }
                Err(err) => {
                    tracing::error!(?err, "desktop duplicator capture failed");
                    break;
                }
            };
        }
    });

    tokio::task::spawn_blocking(move || {
        loop {
            // defer! {
            //     tracing::info!(?active_device_id, ?passive_device_id, "video encode process exit");
            // }

            let mut encoder =
                match VideoEncoder::new(libx264::Libx264Config::default(), client.clone()) {
                    Ok(encoder) => encoder,
                    Err(err) => {
                        tracing::error!(?err, "video encoder initialize failed");
                        return;
                    }
                };

            loop {
                match capture_frame_rx.blocking_recv() {
                    Some(capture_frame) => {
                        if let Err(err) = encoder.encode(capture_frame) {
                            if let CoreError::OutgoingMessageChannelDisconnect = err {
                                tracing::info!("desktop capture and encode process exit");
                                return;
                            } else {
                                tracing::error!(?err, "video encode failed");
                            }
                        }
                    }
                    None => {
                        tracing::error!("capture frame channel closed");
                        return;
                    }
                }
            }
        }
    });
}