lasprs 0.14.3

use super::super::*;
use crate::{
    daq::error::StreamMgrError,
    siggen::{self, Siggen, SiggenCommand, SiggenError},
    tools::find_unused_buf,
    *,
};
use crossbeam::channel::{Receiver, Sender, TrySendError, unbounded};
use dasp_sample::{Sample, ToSample};
use snafu::ResultExt;
use std::{
    any::Any,
    collections::{HashMap, VecDeque},
    mem::replace,
    sync::{Arc, Mutex, Weak, atomic::AtomicBool},
    time::Duration,
};
type Result<T> = std::result::Result<T, StreamMgrError>;

/// Start signal generator thread
///
/// # Args
///
/// - `meta` - Stream metadata
/// - `siggen` - Signal generator to use
/// - `tx` - The channel where the generated signal needs to be send to.
/// - `mon_queues` - Queues where generated signal should also be send to, for
///   monitoring purposes
///
///
/// # Returns
///
/// - Signal generator join handle (to get the Siggen struct back after the stream ends)
/// - Channel part where messages can be send to, to control the signal
///   generator (change source, set gains etc)
/// - Result receiver for messages from sending part.
///
#[allow(clippy::type_complexity)]
pub(crate) fn startSiggenThread(
    meta: Arc<StreamMetaData>,
    mut siggen: Siggen,
    tx: Sender<Arc<RawStreamData>>,
    mut mon_queues: Vec<SharedInQueue>,
) -> Result<(
    JoinHandle<(Siggen, InQueues)>,
    Sender<OutputStreamCommand>,
    Receiver<Result<()>>,
)> {
    // Communicating with
    let (commtx_res, commrx) = unbounded();
    let (commtx, commrx_res) = unbounded();

    // Number of channels to output for
    let nchannels = meta.nchannels();

    siggen.setAllMute(true);
    if siggen.nchannels() != nchannels {
        // Updating number of channels
        siggen.setNChannels(nchannels);
    }
    siggen.reset(meta.samplerate).context(SiggenSnafu)?;

    let threadhandle = spawn(
        move || {
            // What is a good sleep time? We have made sure that there are
            // two buffers available for the output stream. We choose to wake up twice per frame.
            let sleep_time_us = Duration::from_micros(
                (0.1 * 1e6 * meta.framesPerBlock as Flt / *meta.samplerate) as u64,
            );

            let mut bufs: VecDeque<Arc<RawStreamData>> = VecDeque::with_capacity(10);

            let mut floatbuf: Vec<Flt> = vec![0.; nchannels * meta.framesPerBlock];
            let mut ctr = 0;
            'infy: loop {
                if let Ok(streamcommand) = commrx.try_recv() {
                    match streamcommand {
                        // Stop this thread. Returns the queue
                        OutputStreamCommand::StopThread => {
                            commtx.send(Ok(())).unwrap();
                            // Send all monitor queues the fact that the stream stops
                            mon_queues.retain(|q| q.send(InStreamMsg::StreamStopped).is_ok());
                            break 'infy;
                        }
                        OutputStreamCommand::SiggenCommand(cmd) => {
                            // Apply command to signal generator.
                            let res = siggen.applyCommand(cmd);
                            commtx
                                .send(res.map_err(|source| StreamMgrError::SiggenError { source }))
                                .unwrap();
                        }
                        OutputStreamCommand::AddMonitorQueue(tx) => {
                            if let Ok(()) = tx.send(InStreamMsg::StreamStarted(
                                meta.clone(),
                                // Pre capture buffer not used for monitoring
                                PreCaptureBuffer::NotLoaded,
                            )) {
                                mon_queues.push(tx);
                            }
                        }
                    }
                }
                if tx.is_empty() {
                    // Obtain signal from signal generator
                    siggen.genSignal(&mut floatbuf);

                    // Search for a buffer to send over in the list of buffers. Pops
                    // one off in case Arc::get_mut is some. If that is true, the
                    // buffer is not in used and only stored in the list of buffers.
                    // If no unused buffers can be found, we create a new one.
                    let mut buftouse = find_unused_buf(&mut bufs).
                        unwrap_or_else(||

                        // Create a new buffer based on the required metadata
                         match meta.rawDatatype {
                        DataType::I8 => Arc::new(RawStreamData::Datai8(vec![0; floatbuf.len()])),
                        DataType::F32 => Arc::new(RawStreamData::Dataf32(vec![0.; floatbuf.len()])),
                        DataType::F64 => Arc::new(RawStreamData::Dataf64(vec![0.; floatbuf.len()])),
                        DataType::I16 => Arc::new(RawStreamData::Datai16(vec![0; floatbuf.len()])),
                        DataType::I32 => Arc::new(RawStreamData::Datai32(vec![0; floatbuf.len()])),
                        DataType::I24 => Arc::new(RawStreamData::Datai24(vec![dasp_sample::I24::EQUILIBRIUM; floatbuf.len()])),
                    });
                    let mutbuf = Arc::get_mut(&mut buftouse)
                        .expect("Buffer taken tat is in use. Not possible");

                    // Convert signal generator data to raw data and push to the stream thread
                    // The code below works but is a lot of repetition. This
                    // should be re placed by a macro, but I cannot get this
                    // working.
                    match meta.rawDatatype {
                        DataType::I8 => {
                            if let RawStreamData::Datai8(v) = mutbuf {
                                v.iter_mut()
                                    .zip(floatbuf.iter())
                                    .for_each(|(v, f)| *v = (*f).to_sample_());
                            } else {
                                unreachable!("Buffer is of wrong type");
                            }
                        }
                        DataType::I16 => {
                            if let RawStreamData::Datai16(v) = mutbuf {
                                v.iter_mut()
                                    .zip(floatbuf.iter())
                                    .for_each(|(v, f)| *v = (*f).to_sample_());
                            } else {
                                unreachable!("Buffer is of wrong type");
                            }
                        }
                        DataType::I24 => {
                            if let RawStreamData::Datai32(v) = mutbuf {
                                v.iter_mut()
                                    .zip(floatbuf.iter())
                                    .for_each(|(v, f)| *v = (*f).to_sample_());
                            } else {
                                unreachable!("Buffer is of wrong type");
                            }
                        }
                        DataType::I32 => {
                            if let RawStreamData::Datai32(v) = mutbuf {
                                v.iter_mut()
                                    .zip(floatbuf.iter())
                                    .for_each(|(v, f)| *v = (*f).to_sample_());
                            } else {
                                unreachable!("Buffer is of wrong type");
                            }
                        }
                        DataType::F32 => {
                            if let RawStreamData::Dataf32(v) = mutbuf {
                                v.iter_mut()
                                    .zip(floatbuf.iter())
                                    .for_each(|(v, f)| *v = (*f).to_sample_());
                            } else {
                                unreachable!("Buffer is of wrong type");
                            }
                        }
                        DataType::F64 => {
                            if let RawStreamData::Dataf64(v) = mutbuf {
                                v.iter_mut()
                                    .zip(floatbuf.iter())
                                    .for_each(|(v, f)| *v = (*f).to_sample_());
                            } else {
                                unreachable!("Buffer is of wrong type");
                            }
                        }
                    }
                    // Store buffer in list of buffers to avoid allocation in thread
                    bufs.push_back(buftouse.clone());
                    if let Err(_e) = tx.send(buftouse) {
                        // An error occured while trying to send the raw data to
                        // the stream. This might be because the stream has
                        // stopped or has an error.

                        // There is nothing we can do here, but we should not stop the thread.
                    }
                    if !mon_queues.is_empty() {
                        let floatdat = Array2::from_shape_fn(
                            (meta.framesPerBlock, meta.nchannels()).f(),
                            |(frame, channel)| floatbuf[frame * meta.nchannels() + channel],
                        );
                        let msg = InStreamMsg::InStreamData(Arc::new(
                            InStreamData::newFromConverted(ctr, meta.clone(), floatdat, false),
                        ));
                        mon_queues.retain(|q| q.send(msg.clone()).is_ok());
                    }

                    // Increment block counter
                    ctr += 1;
                } else {
                    // dbg!("Nothing to be filled");
                }
            }
            std::thread::sleep(sleep_time_us);
            (siggen, mon_queues)
        },
        ThreadPriority::Normal,
    );
    Ok((threadhandle, commtx_res, commrx_res))
}