async fn merger(
    mut rx_mess: tokio::sync::mpsc::Receiver<SPState>,
    tx_runner: tokio::sync::mpsc::Sender<SPRunnerInput>,
) {
    let (tx, mut rx) = tokio::sync::watch::channel(false);
    let ms_arc = Arc::new(Mutex::new(MergedState::new()));

    let ms_in = ms_arc.clone();
    tokio::spawn(async move {
        loop {
            let s = rx_mess.recv().await.expect("The state channel should always work!");
            {
                ms_in.lock().unwrap().states.push(s);
            }
            tx.send(true).expect("internal channel in merge should always work!");
        }
    });

    let ms_out = ms_arc.clone();
    tokio::spawn(async move {
        loop {
            rx.changed().await;
            let mut states = {
                let mut x = ms_out.lock().unwrap();
                let res = x.states.clone();
                x.states = vec!();
                res
            };
            states.reverse();
            if !states.is_empty() {
                let mut x = states.pop().unwrap();
                for y in states {
                    if let Some(other) =  try_extend(&mut x, y) {
                        // Can not be merged so sending what we have
                        tx_runner.send(SPRunnerInput::StateChange(x.clone())).await;
                        x = other;
                    }
                }
                tx_runner.send(SPRunnerInput::StateChange(x)).await;
            }
        }
    });


}
/// Merging states if many states arrives at the same time
Merging states if many states arrives at the same time
Merging states if many states arrives at the same time
async fn merger(
    mut rx_mess: tokio::sync::mpsc::Receiver<SPState>,
    tx_runner: tokio::sync::mpsc::Sender<SPRunnerInput>,
) {
    let (tx, mut rx) = tokio::sync::watch::channel(false);
    let ms_arc = Arc::new(Mutex::new(MergedState::new()));
    let ms_in = ms_arc.clone();
    tokio::spawn(async move {
        loop {
            let s = rx_mess.recv().await.expect("The state channel should always work!");
            {
                ms_in.lock().unwrap().states.push(s);
            }
            tx.send(true).expect("internal channel in merge should always work!");
        }
    });
    let ms_out = ms_arc.clone();
    tokio::spawn(async move {
        loop {
            rx.changed().await;
            let mut states = {
                let mut x = ms_out.lock().unwrap();
                let res = x.states.clone();
                x.states = vec!();
                res
            };
            states.reverse();
            if !states.is_empty() {
                let mut x = states.pop().unwrap();
                for y in states {
                    if let Some(other) =  try_extend(&mut x, y) {
                        // Can not be merged so sending what we have
                        tx_runner.send(SPRunnerInput::StateChange(x.clone())).await;
                        x = other;
                    }
                }
                tx_runner.send(SPRunnerInput::StateChange(x)).await;
            }
        }
    });
}
async fn merger ( mut rx_mess : tokio :: sync :: mpsc :: Receiver < SPState > , tx_runner : tokio :: sync :: mpsc :: Sender < SPRunnerInput > , ) { let ( tx , mut rx ) = tokio :: sync :: watch :: channel ( false ) ; let ms_arc = Arc :: new ( Mutex :: new ( MergedState :: new ( ) ) ) ; let ms_in = ms_arc . clone ( ) ; tokio :: spawn ( async move { loop { let s = rx_mess . recv ( ) . await . expect ( " " ) ; { ms_in . lock ( ) . unwrap ( ) . states . push ( s ) ; } tx . send ( true ) . expect ( " " ) ; } } ) ; let ms_out = ms_arc . clone ( ) ; tokio :: spawn ( async move { loop { rx . changed ( ) . await ; let mut states = { let mut x = ms_out . lock ( ) . unwrap ( ) ; let res = x . states . clone ( ) ; x . states = vec ! ( ) ; res } ; states . reverse ( ) ; if ! states . is_empty ( ) { let mut x = states . pop ( ) . unwrap ( ) ; for y in states { if let Some ( other ) = try_extend ( & mut x , y ) { // Can not be merged so sending what we have tx_runner . send ( SPRunnerInput :: StateChange ( x . clone ( ) ) ) . await ; x = other ; } } tx_runner . send ( SPRunnerInput :: StateChange ( x ) ) . await ; } } } ) ; }
Merging states if many states arrives at the same time
Merging states if many states arrives at the same time
rx_mess
tokio::sync::mpsc::Receiver<SPState>
tx_runner
tokio::sync::mpsc::Sender<SPRunnerInput>