use futures::prelude::*;
use par_stream::prelude::*;

let data = vec![2, -1, 3, 5];
let stream = futures::stream::iter(data.clone());

// broadcast the stream into three receivers
let mut builder = stream.broadcast(None, true);
let rx1 = builder.register();
let rx2 = builder.register();
let rx3 = builder.register();
builder.build(); // finish the builder to start consuming items

// spawn a parallel processor for each receiver
let stream1 = rx1.map(|v| v * 2).spawned(None);
let stream2 = rx2.map(|v| v * 3).spawned(None);
let stream3 = rx3.map(|v| v * 5).spawned(None);

// collect output values
let vec: Vec<_> = stream1
    .zip(stream2)
    .zip(stream3)
    .map(|((v1, v2), v3)| (v1, v2, v3))
    .collect()
    .await;

// verify output values
assert_eq!(vec, [(4, 6, 10), (-2, -3, -5), (6, 9, 15), (10, 15, 25)]);

并行数据生成

以下组合器创建并行工作进程，每个工作进程独立生成项目。

• par_unfold 从未来生成值。
• par_unfold_blocking 从阻塞函数生成值。
• try_par_unfold 和 try_par_unfold_blocking 是它们的错误版本。

参数

组合器可能需要额外的参数来配置工作进程的数量和缓冲区大小。

• N: Into<NumWorkers> 用于 par_for_each<N, F>(n: N, f: F)
• B: Into<BufSize> 用于 scatter<B>(b: B)
• P: Into<ParParams> 用于 par_then<P, F>(p: P, f: F)

N: Into<NumWorkers> 接受以下值。

• None：默认值，设置为逻辑系统处理器的数量。
• 8（整数）：固定的工作线程数量。
• 2.0（浮点数）：设置为逻辑系统处理器数量的缩放值。

B: Into<BufSize> 接受以下值。

• None：默认值，设置为逻辑系统处理器的两倍。
• 8（整数）：固定缓冲区大小。
• 2.0（浮点数）：设置为逻辑系统处理器数量的缩放值。

P: Into<ParParms> 是工作线程大小和缓冲区大小的组合。它接受以下值。

• None：默认值，设置为工作线程大小和缓冲区大小的默认值。
• 8（整数）：固定工作线程大小，缓冲区大小为工作线程大小的常数倍。
• 2.0（浮点数）：将工作线程大小设置为逻辑系统处理器的缩放值，缓冲区大小为工作线程大小的常数倍。
• ParParamsConfig：手动配置。

实用组合器

该软件包提供了一些实用的流组合器，可以让你更轻松地完成任务 :).

• with_state 将一个流与一个状态值绑定。
• wait_until 允许流等待直到一个future解析。
• reduce 将流项目减少到单个值。
• batching 为每个输出项目消耗任意数量的输入项目。
• stateful_then、stateful_map、stateful_batching 是有状态的对应版本。
• take_until_error 在出现错误后停止流接受值。
• catch_error 将结果流拆分为未包装值流和一个可能解析为错误的future。

使用自定义运行时

要提供自定义运行时实现，声明一个实现 Runtime 的类型。然后，为该类型创建一个实例并将其传递给 set_global_runtime()。全局运行时最多只能设置一次，并且仅在未启用任何运行时Cargo功能时有效。否则 set_global_runtime() 返回错误。

use futures::future::BoxFuture;
use par_stream::rt::{Runtime, SleepHandle, SpawnHandle};
use std::{any::Any, time::Duration};

pub struct MyRuntime {/* omit */}

impl MyRuntime {
    pub fn new() -> Self {
        Self { /* omit */ }
    }
}

unsafe impl Runtime for MyRuntime {
    fn block_on<'a>(
        &self,
        fut: BoxFuture<'a, Box<dyn Send + Any + 'static>>,
    ) -> Box<dyn Send + Any + 'static> {
        todo!()
    }

    fn block_on_executor<'a>(
        &self,
        fut: BoxFuture<'a, Box<dyn Send + Any + 'static>>,
    ) -> Box<dyn Send + Any + 'static> {
        todo!()
    }

    fn spawn(
        &self,
        fut: BoxFuture<'static, Box<dyn Send + Any + 'static>>,
    ) -> Box<dyn SpawnHandle> {
        todo!()
    }

    fn spawn_blocking(
        &self,
        f: Box<dyn FnOnce() -> Box<dyn Send + Any + 'static> + Send>,
    ) -> Box<dyn SpawnHandle> {
        todo!()
    }

    fn sleep(&self, dur: Duration) -> Box<dyn SleepHandle> {
        todo!()
    }
}

par_stream::rt::set_global_runtime(MyRuntime::new()).unwrap();