#stream #tokio #genstage #pipeline

laststage

快速、耐用、高并发的 HashMap

3 个版本 (1 个稳定版)

1.0.0 2022 年 4 月 22 日
0.1.1 2022 年 4 月 22 日
0.1.0 2022 年 4 月 22 日

#845并发

Apache-2.0

22KB
317 行 代码

LastStage

LastStage 是一个用于在生产者和消费者之间交换事件的规范,采用 基于推的模式

此项目目前提供

  • 生产者 产生事件并通过调度器分发到订阅者(无输入 - 一个/多个输出)

  • 生产者消费者 它从上游获取事件并通过调度器分发到订阅者(一个/多个输入 - 一个/多个输出)

  • 消费者 它从上游获取事件并消费它(一个/多个输入 - 无输出)

  • 调度器 首先获取一个到多个订阅者,然后通过两种模式(广播/循环)开始分发事件

示例

示例目录

安装

laststage = "1.0.0"

快速示例


#[tokio::main]
async fn main() {

    let(shutdown_sender, shutdown_recv) = channel();


    // -----------------------------------
    //
    // Producer -> ProducerConsumer -> Consumer 
    //
    // ------------------------------------



    // Run Consumer
    let log_chan = ConsumerRunnable::new(Box::new(Log)).run(100);


    // Run ProducerConsumer
    let filter_chan = ProducerConsumerRunnable::new(Box::new(FilterByAge), 
                                                    vec![log_chan], 
                                                    Some(DispatcherType::RoundRobin)
                                                    ).unwrap().run(100);

    // Run Producer
    let _ = ProducerRunnable::new(Box::new(Prod), 
                                  vec![filter_chan], 
                                  None, 
                                  100, 
                                  shutdown_recv).unwrap().run();

    
    tokio::time::sleep(Duration::from_secs(10)).await;
}


#[derive(Clone)]
struct ProdEvent {
    pub funame: String,
    pub age: i32
}

struct Prod;

#[async_trait]
impl Producer<ProdEvent> for Prod {
    async fn init(&mut self) {

    }

    async fn handle_demand(&mut self, max_demand: usize) -> Vec<ProdEvent> {
        (0..max_demand as i32)
            .into_iter()
            .map(|i| {
                
                ProdEvent { 
                    funame: format!("DanyalMh-{}", i), 
                    age: (i + 30) % 35 
                }

            })
            .collect()
    }

    async fn terminate(&mut self) {

    }
} 


// -------------------------------------------


struct FilterByAge;

#[async_trait]
impl ProducerConsumer<ProdEvent, ProdEvent> for FilterByAge {
    async fn init(&mut self) {
        
    }

    async fn handle_events(&mut self, events: Vec<ProdEvent>) -> Vec<ProdEvent> {
        events
            .into_iter()
            .filter(|pe| pe.age > 25 && pe.age < 32)
            .collect()
    }

    async fn terminate(&mut self) {

    }


} 



struct Log;

#[async_trait]
impl Consumer<ProdEvent> for Log {
    async fn init(&mut self) {

    }

    async fn handle_events(&mut self, events: Vec<ProdEvent>) -> State<ProdEvent> {
        events
            .into_iter()
            .for_each(|pe| {
                println!("==> {} -> {}", pe.funame, pe.age)
            });
        
        State::Continue
    }  


    async fn terminate(&mut self) {
        
    }
}


依赖项

~2.3–4MB
~66K SLoC