Lib.rs

Async Bi-directional Pipe

sync example:

use anyhow::{Result,ensure};
use bi_directional_pipe::sync::pipe;
use tokio::task::JoinHandle;

#[tokio::main]
async fn main()->Result<()> {
    let (left,right)=pipe();
    let join:JoinHandle<Result<()>>= tokio::spawn(async move{
        for i in 0..1000000 {
            right.send(i);
            let ok=right.recv().await?;
            ensure!(ok==i+1,"left return v error!!")
        }
        Ok(())
    });

    loop {
        if let Ok(v)=left.recv().await{
            left.send(v+1);
        }else {
            break;
        }
    }

    join.await??;
    Ok(())
}

unsync example:

use anyhow::{Result,ensure};
use std::time::Instant;
use bi_directional_pipe::unsync::pipe;
use tokio::task::JoinHandle;

#[tokio::main]
async fn main()->Result<()> {
    let single_runtime=tokio::task::LocalSet::new();
    let (left,right)=pipe();
    
    let res:Result<()>= single_runtime.run_until(async move{
        let join:JoinHandle<Result<()>>= tokio::task::spawn_local(async move{
            for i in 0..1000000 {
                right.send(i);
                let ok=right.recv().await?;
                ensure!(ok==i+1,"left return v error!!")
            }
            Ok(())
        });

        let start=Instant::now();
        loop {
            if let Ok(v)=left.recv().await{
                left.send(v+1);
            }else {
                break;
            }
        }
        println!("time {}ms",start.elapsed().as_millis());

        join.await??;
        Ok(())
    }).await;

    res?;
    Ok(())
}