fn run_laser_stream<M, F>(
    dac: &DetectedDac,
    state: &Arc<Mutex<State>>,
    model: &Arc<Mutex<Option<M>>>,
    render: F,
    state_update_rx: &mpsc::Receiver<StateUpdate>,
    model_update_rx: &mpsc::Receiver<ModelUpdate<M>>,
    is_closed: &AtomicBool,
) -> Result<(), RawStreamError>
where
    F: RenderFn<M>,
{
    // Currently only ether dream is supported, so retrieve the broadcast and addr.
    let (broadcast, src_addr) = match dac {
        DetectedDac::EtherDream { broadcast, source_addr } => {
            (broadcast, source_addr)
        }
    };

    // A buffer for collecting model updates.
    let mut pending_model_updates: Vec<ModelUpdate<M>> = Vec::new();

    // Establish the TCP connection.
    let ip = src_addr.ip().clone();

    let mut stream = ether_dream::dac::stream::connect(&broadcast, ip)
        .map_err(|err| EtherDreamStreamError::FailedToConnectStream { err })?;

    // Prepare the DAC's playback engine and await the repsonse.
    stream
        .queue_commands()
        .prepare_stream()
        .submit()
        .map_err(|err| EtherDreamStreamError::FailedToPrepareStream { err })?;

    let dac_max_point_hz = dac.max_point_hz();

    // Get the initial point hz by clamping via the DAC's maximum point rate.
    let init_point_hz = {
        let hz = state.lock().expect("failed to acquire raw state lock").point_hz;
        std::cmp::min(hz, dac_max_point_hz)
    };

    // Queue the initial frame and tell the DAC to begin producing output.
    let low_water_mark = 0;
    let n_points = dac_remaining_buffer_capacity(stream.dac());
    stream
        .queue_commands()
        .data((0..n_points).map(|_| centered_blank()))
        .begin(low_water_mark, init_point_hz)
        .submit()
        .map_err(|err| EtherDreamStreamError::FailedToBeginStream { err })?;

    // For collecting the ether-dream points.
    let mut ether_dream_points = vec![];

    while !is_closed.load(atomic::Ordering::Relaxed) {
        // Collect any pending updates.
        pending_model_updates.extend(model_update_rx.try_iter());
        // If there are some updates available, take the lock and apply them.
        if !pending_model_updates.is_empty() {
            if let Ok(mut guard) = model.lock() {
                let mut model = guard.take().unwrap();
                for mut update in pending_model_updates.drain(..) {
                    update(&mut model);
                }
                *guard = Some(model);
            }
        }

        // Check for updates and retrieve a copy of the state.
        let (state, prev_point_hz) = {
            let mut state = state.lock().expect("failed to acquare raw state lock");

            // Keep track of whether or not the `point_hz` as changed.
            let prev_point_hz = std::cmp::min(state.point_hz, dac.max_point_hz());

            // Apply updates.
            for mut state_update in state_update_rx.try_iter() {
                (*state_update)(&mut state);
            }

            (state.clone(), prev_point_hz)
        };

        // Clamp the point hz by the DAC's maximum point rate.
        let point_hz = std::cmp::min(state.point_hz, dac.max_point_hz());

        // If the point rate changed, we need to tell the DAC and set the control value on point 0.
        let point_rate_changed = point_hz != prev_point_hz;
        if point_rate_changed {
            stream
                .queue_commands()
                .point_rate(point_hz)
                .submit()
                .map_err(|err| EtherDreamStreamError::FailedToSubmitPointRate { err })?;
        }

        // Clamp the latency by the DAC's buffer capacity.
        let latency_points = std::cmp::min(state.latency_points, dac.buffer_capacity());
        // Determine how many points the DAC can currently receive.
        let n_points = points_to_generate(stream.dac(), latency_points as u16) as usize;

        // The buffer that the user will write to. TODO: Re-use this points buffer.
        let mut buffer = Buffer {
            point_hz,
            latency_points: latency_points as _,
            points: vec![RawPoint::centered_blank(); n_points].into_boxed_slice(),
        };

        // Request the points from the user.
        if let Ok(mut guard) = model.lock() {
            let mut m = guard.take().unwrap();
            render(&mut m, &mut buffer);
            *guard = Some(m);
        }

        // Retrieve the points.
        ether_dream_points.extend(buffer.iter().cloned().map(point_to_ether_dream_point));

        // If the point rate changed, set the control value on the first point to trigger it.
        if point_rate_changed && !ether_dream_points.is_empty() {
            ether_dream_points[0].control = ether_dream::dac::PointControl::CHANGE_RATE.bits();
        }

        // Submit the points.
        stream
            .queue_commands()
            .data(ether_dream_points.drain(..))
            .submit()
            .map_err(|err| EtherDreamStreamError::FailedToSubmitData { err })?;
    }

    stream
        .queue_commands()
        .stop()
        .submit()
        .map_err(|err| EtherDreamStreamError::FailedToStopStream { err })?;

    Ok(())
}
// The function to run on the laser stream thread.
The function to run on the laser stream thread.
The function to run on the laser stream thread .
fn run_laser_stream<M, F>(
    dac: &DetectedDac,
    state: &Arc<Mutex<State>>,
    model: &Arc<Mutex<Option<M>>>,
    render: F,
    state_update_rx: &mpsc::Receiver<StateUpdate>,
    model_update_rx: &mpsc::Receiver<ModelUpdate<M>>,
    is_closed: &AtomicBool,
) -> Result<(), RawStreamError>
where
    F: RenderFn<M>,
{
    // Currently only ether dream is supported, so retrieve the broadcast and addr.
    let (broadcast, src_addr) = match dac {
        DetectedDac::EtherDream { broadcast, source_addr } => {
            (broadcast, source_addr)
        }
    };
    // A buffer for collecting model updates.
    let mut pending_model_updates: Vec<ModelUpdate<M>> = Vec::new();
    // Establish the TCP connection.
    let ip = src_addr.ip().clone();
    let mut stream = ether_dream::dac::stream::connect(&broadcast, ip)
        .map_err(|err| EtherDreamStreamError::FailedToConnectStream { err })?;
    // Prepare the DAC's playback engine and await the repsonse.
    stream
        .queue_commands()
        .prepare_stream()
        .submit()
        .map_err(|err| EtherDreamStreamError::FailedToPrepareStream { err })?;
    let dac_max_point_hz = dac.max_point_hz();
    // Get the initial point hz by clamping via the DAC's maximum point rate.
    let init_point_hz = {
        let hz = state.lock().expect("failed to acquire raw state lock").point_hz;
        std::cmp::min(hz, dac_max_point_hz)
    };
    // Queue the initial frame and tell the DAC to begin producing output.
    let low_water_mark = 0;
    let n_points = dac_remaining_buffer_capacity(stream.dac());
    stream
        .queue_commands()
        .data((0..n_points).map(|_| centered_blank()))
        .begin(low_water_mark, init_point_hz)
        .submit()
        .map_err(|err| EtherDreamStreamError::FailedToBeginStream { err })?;
    // For collecting the ether-dream points.
    let mut ether_dream_points = vec![];
    while !is_closed.load(atomic::Ordering::Relaxed) {
        // Collect any pending updates.
        pending_model_updates.extend(model_update_rx.try_iter());
        // If there are some updates available, take the lock and apply them.
        if !pending_model_updates.is_empty() {
            if let Ok(mut guard) = model.lock() {
                let mut model = guard.take().unwrap();
                for mut update in pending_model_updates.drain(..) {
                    update(&mut model);
                }
                *guard = Some(model);
            }
        }
        // Check for updates and retrieve a copy of the state.
        let (state, prev_point_hz) = {
            let mut state = state.lock().expect("failed to acquare raw state lock");
            // Keep track of whether or not the `point_hz` as changed.
            let prev_point_hz = std::cmp::min(state.point_hz, dac.max_point_hz());
            // Apply updates.
            for mut state_update in state_update_rx.try_iter() {
                (*state_update)(&mut state);
            }
            (state.clone(), prev_point_hz)
        };
        // Clamp the point hz by the DAC's maximum point rate.
        let point_hz = std::cmp::min(state.point_hz, dac.max_point_hz());
        // If the point rate changed, we need to tell the DAC and set the control value on point 0.
        let point_rate_changed = point_hz != prev_point_hz;
        if point_rate_changed {
            stream
                .queue_commands()
                .point_rate(point_hz)
                .submit()
                .map_err(|err| EtherDreamStreamError::FailedToSubmitPointRate { err })?;
        }
        // Clamp the latency by the DAC's buffer capacity.
        let latency_points = std::cmp::min(state.latency_points, dac.buffer_capacity());
        // Determine how many points the DAC can currently receive.
        let n_points = points_to_generate(stream.dac(), latency_points as u16) as usize;
        // The buffer that the user will write to. TODO: Re-use this points buffer.
        let mut buffer = Buffer {
            point_hz,
            latency_points: latency_points as _,
            points: vec![RawPoint::centered_blank(); n_points].into_boxed_slice(),
        };
        // Request the points from the user.
        if let Ok(mut guard) = model.lock() {
            let mut m = guard.take().unwrap();
            render(&mut m, &mut buffer);
            *guard = Some(m);
        }
        // Retrieve the points.
        ether_dream_points.extend(buffer.iter().cloned().map(point_to_ether_dream_point));
        // If the point rate changed, set the control value on the first point to trigger it.
        if point_rate_changed && !ether_dream_points.is_empty() {
            ether_dream_points[0].control = ether_dream::dac::PointControl::CHANGE_RATE.bits();
        }
        // Submit the points.
        stream
            .queue_commands()
            .data(ether_dream_points.drain(..))
            .submit()
            .map_err(|err| EtherDreamStreamError::FailedToSubmitData { err })?;
    }
    stream
        .queue_commands()
        .stop()
        .submit()
        .map_err(|err| EtherDreamStreamError::FailedToStopStream { err })?;
    Ok(())
}
fn run_laser_stream < M , F > ( dac : & DetectedDac , state : & Arc < Mutex < State > > , model : & Arc < Mutex < Option < M > > > , render : F , state_update_rx : & mpsc :: Receiver < StateUpdate > , model_update_rx : & mpsc :: Receiver < ModelUpdate < M > > , is_closed : & AtomicBool , ) -> Result < ( ) , RawStreamError > where F : RenderFn < M > , { // Currently only ether dream is supported, so retrieve the broadcast and addr. let ( broadcast , src_addr ) = match dac { DetectedDac :: EtherDream { broadcast , source_addr } => { ( broadcast , source_addr ) } } ; // A buffer for collecting model updates. let mut pending_model_updates : Vec < ModelUpdate < M > > = Vec :: new ( ) ; // Establish the TCP connection. let ip = src_addr . ip ( ) . clone ( ) ; let mut stream = ether_dream :: dac :: stream :: connect ( & broadcast , ip ) . map_err ( | err | EtherDreamStreamError :: FailedToConnectStream { err } ) ? ; // Prepare the DAC's playback engine and await the repsonse. stream . queue_commands ( ) . prepare_stream ( ) . submit ( ) . map_err ( | err | EtherDreamStreamError :: FailedToPrepareStream { err } ) ? ; let dac_max_point_hz = dac . max_point_hz ( ) ; // Get the initial point hz by clamping via the DAC's maximum point rate. let init_point_hz = { let hz = state . lock ( ) . expect ( " " ) . point_hz ; std :: cmp :: min ( hz , dac_max_point_hz ) } ; // Queue the initial frame and tell the DAC to begin producing output. let low_water_mark = 0 ; let n_points = dac_remaining_buffer_capacity ( stream . dac ( ) ) ; stream . queue_commands ( ) . data ( ( 0 .. n_points ) . map ( | _ | centered_blank ( ) ) ) . begin ( low_water_mark , init_point_hz ) . submit ( ) . map_err ( | err | EtherDreamStreamError :: FailedToBeginStream { err } ) ? ; // For collecting the ether-dream points. let mut ether_dream_points = vec ! [ ] ; while ! is_closed . load ( atomic :: Ordering :: Relaxed ) { // Collect any pending updates. pending_model_updates . extend ( model_update_rx . try_iter ( ) ) ; // If there are some updates available, take the lock and apply them. if ! pending_model_updates . is_empty ( ) { if let Ok ( mut guard ) = model . lock ( ) { let mut model = guard . take ( ) . unwrap ( ) ; for mut update in pending_model_updates . drain ( .. ) { update ( & mut model ) ; } * guard = Some ( model ) ; } } // Check for updates and retrieve a copy of the state. let ( state , prev_point_hz ) = { let mut state = state . lock ( ) . expect ( " " ) ; // Keep track of whether or not the `point_hz` as changed. let prev_point_hz = std :: cmp :: min ( state . point_hz , dac . max_point_hz ( ) ) ; // Apply updates. for mut state_update in state_update_rx . try_iter ( ) { ( * state_update ) ( & mut state ) ; } ( state . clone ( ) , prev_point_hz ) } ; // Clamp the point hz by the DAC's maximum point rate. let point_hz = std :: cmp :: min ( state . point_hz , dac . max_point_hz ( ) ) ; // If the point rate changed, we need to tell the DAC and set the control value on point 0. let point_rate_changed = point_hz != prev_point_hz ; if point_rate_changed { stream . queue_commands ( ) . point_rate ( point_hz ) . submit ( ) . map_err ( | err | EtherDreamStreamError :: FailedToSubmitPointRate { err } ) ? ; } // Clamp the latency by the DAC's buffer capacity. let latency_points = std :: cmp :: min ( state . latency_points , dac . buffer_capacity ( ) ) ; // Determine how many points the DAC can currently receive. let n_points = points_to_generate ( stream . dac ( ) , latency_points as u16 ) as usize ; // The buffer that the user will write to. TODO: Re-use this points buffer. let mut buffer = Buffer { point_hz , latency_points : latency_points as _ , points : vec ! [ RawPoint centered_blank ( ) n_points ] . into_boxed_slice ( ) , } ; // Request the points from the user. if let Ok ( mut guard ) = model . lock ( ) { let mut m = guard . take ( ) . unwrap ( ) ; render ( & mut m , & mut buffer ) ; * guard = Some ( m ) ; } // Retrieve the points. ether_dream_points . extend ( buffer . iter ( ) . cloned ( ) . map ( point_to_ether_dream_point ) ) ; // If the point rate changed, set the control value on the first point to trigger it. if point_rate_changed && ! ether_dream_points . is_empty ( ) { ether_dream_points [ 0 ] . control = ether_dream :: dac :: PointControl :: CHANGE_RATE . bits ( ) ; } // Submit the points. stream . queue_commands ( ) . data ( ether_dream_points . drain ( .. ) ) . submit ( ) . map_err ( | err | EtherDreamStreamError :: FailedToSubmitData { err } ) ? ; } stream . queue_commands ( ) . stop ( ) . submit ( ) . map_err ( | err | EtherDreamStreamError :: FailedToStopStream { err } ) ? ; Ok ( ( ) ) }
The function to run on the laser stream thread.
The function to run on the laser stream thread .
dac
&DetectedDac
state
&Arc<Mutex<State>>
model
&Arc<Mutex<Option<M>>>
render
F
state_update_rx
&mpsc::Receiver<StateUpdate>
model_update_rx
&mpsc::Receiver<ModelUpdate<M>>
is_closed
&AtomicBool
dac
&DetectedDac
state
&Arc<Mutex<State>>
model
&Arc<Mutex<Option<M>>>
render
F
state_update_rx
&mpsc::Receiver<StateUpdate>
model_update_rx
&mpsc::Receiver<ModelUpdate<M>>
is_closed
&AtomicBool