/// Build the stream with the specified parameters.
///
/// **Note:** If no `dac` was specified, this will method will block until a DAC is detected.
/// The first detected DAC is the DAC with which a stream will be established.
Build the stream with the specified parameters.
Note:** If no `dac` was specified, this will method will block until a DAC is detected.
The first detected DAC is the DAC with which a stream will be established.
Build the stream with the specified parameters . Note : ** If no ` dac ` was specified this will method will block until a DAC is detected . The first detected DAC is the DAC with which a stream will be established .
pub fn build(self) -> io::Result<Stream<M>>
    where
        M: 'static + Send,
        F: 'static + RenderFn<M> + Send,
    {
        let Builder { api_inner, builder, model, render } = self;
        // Prepare the model for sharing between the laser thread and stream handle.
        let model = Arc::new(Mutex::new(Some(model)));
        let model_2 = model.clone();
        // The channels used for sending updates to the model via the stream handle.
        let (model_update_tx, m_rx) = mpsc::channel();
        let (state_update_tx, s_rx) = mpsc::channel();
        // Retrieve the specified point rate or use a default.
        let point_hz = builder.point_hz.unwrap_or(super::DEFAULT_POINT_HZ);
        // Retrieve the latency as a number of points.
        let latency_points = builder.latency_points
            .unwrap_or_else(|| default_latency_points(point_hz));
        // The raw laser stream state to live on the laser thread.
        let state = Arc::new(Mutex::new(State { point_hz, latency_points }));
        // Retrieve whether or not the user specified a detected DAC.
        let mut maybe_dac = builder.dac;
        // A flag for tracking whether or not the stream has been closed.
        let is_closed = Arc::new(AtomicBool::new(false));
        let is_closed2 = is_closed.clone();
        // Spawn the thread for communicating with the DAC.
        std::thread::Builder::new()
            .name("raw_laser_stream_thread".into())
            .spawn(move || {
                let mut connect_attempts = 3;
                while !is_closed2.load(atomic::Ordering::Relaxed) {
                    // If there are no more remaining connection attempts, try to redetect the DAC
                    // if a specific DAC was specified by the user.
                    if connect_attempts == 0 {
                        connect_attempts = 3;
                        if let Some(ref mut dac) = maybe_dac {
                            let dac_id = dac.id();
                            eprintln!("re-attempting to detect DAC with id: {:?}", dac_id);
                            *dac = match api_inner.detect_dac(dac_id) {
                                Ok(dac) => dac,
                                Err(err) => {
                                    let err = EtherDreamStreamError::FailedToDetectDacs { err };
                                    return Err(RawStreamError::EtherDreamStream { err });
                                }
                            };
                        }
                    }
                    // Retrieve the DAC or find one.
                    let dac = match maybe_dac {
                        Some(ref dac) => dac.clone(),
                        None => api_inner.detect_dacs()
                            .map_err(|err| EtherDreamStreamError::FailedToDetectDacs { err })?
                            .next()
                            .expect("ether dream DAC detection iterator should never return `None`")
                            .map_err(|err| EtherDreamStreamError::FailedToDetectDacs { err })?,
                    };
                    // Connect and run the laser stream.
                    match run_laser_stream(&dac, &state, &model_2, &render, &s_rx, &m_rx, &is_closed2) {
                        Ok(()) => return Ok(()),
                        Err(RawStreamError::EtherDreamStream { err }) => match err {
                            // If we failed to connect to the DAC, keep track of attempts.
                            EtherDreamStreamError::FailedToConnectStream { err } => {
                                eprintln!("failed to connect to stream: {}", err);
                                connect_attempts -= 1;
                                eprintln!(
                                    "connection attempts remaining before re-detecting DAC: {}",
                                    connect_attempts,
                                );
                                // Sleep for a moment to avoid spamming the socket.
                                std::thread::sleep(std::time::Duration::from_millis(16));
                            }
                            // If we failed to prepare the stream or submit data, retry.
                            EtherDreamStreamError::FailedToPrepareStream { .. }
                            | EtherDreamStreamError::FailedToBeginStream { .. }
                            | EtherDreamStreamError::FailedToSubmitData { .. }
                            | EtherDreamStreamError::FailedToSubmitPointRate { .. } => {
                                eprintln!("{} - will now attempt to reconnect", err);
                            }
                            // Return all other errors.
                            err => return Err(RawStreamError::EtherDreamStream { err }),
                        }
                    }
                }
                Ok(())
            })?;
        let is_paused = AtomicBool::new(false);
        let shared = Arc::new(Shared { model, is_paused, is_closed });
        let stream = Stream { shared, state_update_tx, model_update_tx };
        Ok(stream)
    }
pub fn build ( self ) -> io :: Result < Stream < M > > where M : ' static + Send , F : ' static + RenderFn < M > + Send , { let Builder { api_inner , builder , model , render } = self ; // Prepare the model for sharing between the laser thread and stream handle. let model = Arc :: new ( Mutex :: new ( Some ( model ) ) ) ; let model_2 = model . clone ( ) ; // The channels used for sending updates to the model via the stream handle. let ( model_update_tx , m_rx ) = mpsc :: channel ( ) ; let ( state_update_tx , s_rx ) = mpsc :: channel ( ) ; // Retrieve the specified point rate or use a default. let point_hz = builder . point_hz . unwrap_or ( super :: DEFAULT_POINT_HZ ) ; // Retrieve the latency as a number of points. let latency_points = builder . latency_points . unwrap_or_else ( | | default_latency_points ( point_hz ) ) ; // The raw laser stream state to live on the laser thread. let state = Arc :: new ( Mutex :: new ( State { point_hz , latency_points } ) ) ; // Retrieve whether or not the user specified a detected DAC. let mut maybe_dac = builder . dac ; // A flag for tracking whether or not the stream has been closed. let is_closed = Arc :: new ( AtomicBool :: new ( false ) ) ; let is_closed2 = is_closed . clone ( ) ; // Spawn the thread for communicating with the DAC. std :: thread :: Builder :: new ( ) . name ( " " . into ( ) ) . spawn ( move | | { let mut connect_attempts = 3 ; while ! is_closed2 . load ( atomic :: Ordering :: Relaxed ) { // If there are no more remaining connection attempts, try to redetect the DAC // if a specific DAC was specified by the user. if connect_attempts == 0 { connect_attempts = 3 ; if let Some ( ref mut dac ) = maybe_dac { let dac_id = dac . id ( ) ; eprintln ! ( " " dac_id ) ; * dac = match api_inner . detect_dac ( dac_id ) { Ok ( dac ) => dac , Err ( err ) => { let err = EtherDreamStreamError :: FailedToDetectDacs { err } ; return Err ( RawStreamError :: EtherDreamStream { err } ) ; } } ; } } // Retrieve the DAC or find one. let dac = match maybe_dac { Some ( ref dac ) => dac . clone ( ) , None => api_inner . detect_dacs ( ) . map_err ( | err | EtherDreamStreamError :: FailedToDetectDacs { err } ) ? . next ( ) . expect ( " " ) . map_err ( | err | EtherDreamStreamError :: FailedToDetectDacs { err } ) ? , } ; // Connect and run the laser stream. match run_laser_stream ( & dac , & state , & model_2 , & render , & s_rx , & m_rx , & is_closed2 ) { Ok ( ( ) ) => return Ok ( ( ) ) , Err ( RawStreamError :: EtherDreamStream { err } ) => match err { // If we failed to connect to the DAC, keep track of attempts. EtherDreamStreamError :: FailedToConnectStream { err } => { eprintln ! ( " " err ) ; connect_attempts -= 1 ; eprintln ! ( " " connect_attempts ) ; // Sleep for a moment to avoid spamming the socket. std :: thread :: sleep ( std :: time :: Duration :: from_millis ( 16 ) ) ; } // If we failed to prepare the stream or submit data, retry. EtherDreamStreamError :: FailedToPrepareStream { .. } | EtherDreamStreamError :: FailedToBeginStream { .. } | EtherDreamStreamError :: FailedToSubmitData { .. } | EtherDreamStreamError :: FailedToSubmitPointRate { .. } => { eprintln ! ( " " err ) ; } // Return all other errors. err => return Err ( RawStreamError :: EtherDreamStream { err } ) , } } } Ok ( ( ) ) } ) ? ; let is_paused = AtomicBool :: new ( false ) ; let shared = Arc :: new ( Shared { model , is_paused , is_closed } ) ; let stream = Stream { shared , state_update_tx , model_update_tx } ; Ok ( stream ) }
Build the stream with the specified parameters.
Build the stream with the specified parameters .
self