3 个版本 (1 个稳定版)
1.0.0 | 2022 年 4 月 22 日 |
---|---|
0.1.1 | 2022 年 4 月 22 日 |
0.1.0 | 2022 年 4 月 22 日 |
#845 在 并发
22KB
317 行 代码
LastStage
LastStage 是一个用于在生产者和消费者之间交换事件的规范,采用 基于推的模式
此项目目前提供
-
生产者 产生事件并通过调度器分发到订阅者(无输入 - 一个/多个输出)
-
生产者消费者 它从上游获取事件并通过调度器分发到订阅者(一个/多个输入 - 一个/多个输出)
-
消费者 它从上游获取事件并消费它(一个/多个输入 - 无输出)
-
调度器 首先获取一个到多个订阅者,然后通过两种模式(广播/循环)开始分发事件
示例
示例目录
-
[简单] (https://github.com/Rustixir/last_stage/blob/master/examples/simple.rs)
-
[多] (https://github.com/Rustixir/last_stage/blob/master/examples/multi.rs)
安装
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