Struct reactive_mutiny::uni::Uni

pub struct Uni<ItemType: Send + Sync + Debug + 'static, UniChannelType: FullDuplexUniChannel<'static, ItemType, DerivedItemType> + Send + Sync + 'static, const INSTRUMENTS: usize, DerivedItemType: Send + Sync + Debug + 'static = ItemType> {
    pub channel: Arc<UniChannelType>,
    pub stream_executors: Vec<Arc<StreamExecutor>>,
    pub finished_executors_count: AtomicU32,
    /* private fields */
}

Contains the producer-side Uni handle used to interact with the uni event – for closing the stream, requiring stats, …

Fields

channel: Arc<UniChannelType>

stream_executors: Vec<Arc<StreamExecutor>>

finished_executors_count: AtomicU32

Implementations

impl<ItemType: Send + Sync + Debug + 'static, UniChannelType: FullDuplexUniChannel<'static, ItemType, DerivedItemType> + Send + Sync + 'static, const INSTRUMENTS: usize, DerivedItemType: Send + Sync + Debug> Uni<ItemType, UniChannelType, INSTRUMENTS, DerivedItemType>

pub fn send(&self, item: ItemType) -> RetryConsumerResult<(), ItemType, ()>

pub fn send_with<F: FnOnce(&mut ItemType)>( &self, setter: F ) -> RetryConsumerResult<(), F, ()>

pub fn consumer_stream( self ) -> (Arc<Self>, Vec<MutinyStream<'static, ItemType, UniChannelType, DerivedItemType>>)

Sets this Uni to return Streams instead of executing them

pub async fn flush(&self, duration: Duration) -> u32

pub async fn close(&self, timeout: Duration) -> bool

Closes this Uni, in isolation – flushing pending events, closing the producers, waiting for all events to be fully processed and calling the “on close” callback.
Returns false if the timeout kicked-in before it could be attested that the closing was complete.
If this Uni share resources with another one (which will get dumped by the “on close” callback), most probably you want to close them atomically – see unis_close_async!()

pub fn spawn_executors<OutItemType: Send + Debug, OutStreamType: Stream<Item = OutType> + Send + 'static, OutType: Future<Output = Result<OutItemType, Box<dyn Error + Send + Sync>>> + Send, ErrVoidAsyncType: Future<Output = ()> + Send + 'static, CloseVoidAsyncType: Future<Output = ()> + Send + 'static>( self, concurrency_limit: u32, futures_timeout: Duration, pipeline_builder: impl Fn(MutinyStream<'static, ItemType, UniChannelType, DerivedItemType>) -> OutStreamType, on_err_callback: impl Fn(Box<dyn Error + Send + Sync>) -> ErrVoidAsyncType + Send + Sync + 'static, on_close_callback: impl FnOnce(Arc<StreamExecutor>) -> CloseVoidAsyncType + Send + Sync + 'static ) -> Arc<Uni<ItemType, UniChannelType, INSTRUMENTS, DerivedItemType>>

Spawns an optimized executor for the Stream returned by pipeline_builder(), provided it produces elements which are Future & fallible (Actually, as many consumers as MAX_STREAMS will be spawned).
on_close_callback(stats) is called when this Uni (and all Streams) are closed.
on_err_callback(error) is called whenever the Stream returns an Err element.

pub fn spawn_fallibles_executors<OutItemType: Send + Debug, OutStreamType: Stream<Item = Result<OutItemType, Box<dyn Error + Send + Sync>>> + Send + 'static, CloseVoidAsyncType: Future<Output = ()> + Send + 'static>( self, concurrency_limit: u32, pipeline_builder: impl Fn(MutinyStream<'static, ItemType, UniChannelType, DerivedItemType>) -> OutStreamType, on_err_callback: impl Fn(Box<dyn Error + Send + Sync>) + Send + Sync + 'static, on_close_callback: impl FnOnce(Arc<StreamExecutor>) -> CloseVoidAsyncType + Send + Sync + 'static ) -> Arc<Uni<ItemType, UniChannelType, INSTRUMENTS, DerivedItemType>>

Spawns an optimized executor for the Stream returned by pipeline_builder(), provided it produces elements which are fallible & non-future (Actually, as many consumers as MAX_STREAMS will be spawned).
on_close_callback(stats) is called when this Uni (and all Streams) are closed.
on_err_callback(error) is called whenever the Stream returns an Err element.

pub fn spawn_futures_executors<OutItemType: Send + Debug, OutStreamType: Stream<Item = OutType> + Send + 'static, OutType: Future<Output = OutItemType> + Send, CloseVoidAsyncType: Future<Output = ()> + Send + 'static>( self, concurrency_limit: u32, futures_timeout: Duration, pipeline_builder: impl Fn(MutinyStream<'static, ItemType, UniChannelType, DerivedItemType>) -> OutStreamType, on_close_callback: impl FnOnce(Arc<StreamExecutor>) -> CloseVoidAsyncType + Send + Sync + 'static ) -> Arc<Uni<ItemType, UniChannelType, INSTRUMENTS, DerivedItemType>>

Spawns an optimized executor for the Stream returned by pipeline_builder(), provided it produces elements which are Future & non-fallible (Actually, as many consumers as MAX_STREAMS will be spawned).
on_close_callback(stats) is called when this Uni (and all Streams) are closed.

pub fn spawn_non_futures_non_fallibles_executors<OutItemType: Send + Debug, OutStreamType: Stream<Item = OutItemType> + Send + 'static, CloseVoidAsyncType: Future<Output = ()> + Send + 'static>( self, concurrency_limit: u32, pipeline_builder: impl Fn(MutinyStream<'static, ItemType, UniChannelType, DerivedItemType>) -> OutStreamType, on_close_callback: impl FnOnce(Arc<StreamExecutor>) -> CloseVoidAsyncType + Send + Sync + 'static ) -> Arc<Uni<ItemType, UniChannelType, INSTRUMENTS, DerivedItemType>>

Spawns an optimized executor for the Stream returned by pipeline_builder(), provided it produces elements which are non-future & non-fallible (Actually, as many consumers as MAX_STREAMS will be spawned).
on_close_callback(stats) is called when this Uni (and all Streams) are closed.

Trait Implementations

impl<ItemType: Send + Sync + Debug + 'static, UniChannelType: FullDuplexUniChannel<'static, ItemType, DerivedItemType> + Send + Sync + 'static, const INSTRUMENTS: usize, DerivedItemType: Send + Sync + Debug> GenericUni for Uni<ItemType, UniChannelType, INSTRUMENTS, DerivedItemType>

const INSTRUMENTS: usize = INSTRUMENTS

The instruments this Uni will collect/report

type ItemType = ItemType

The payload type this Uni’s producers will receive

type UniChannelType = UniChannelType

The channel through which payloads will travel from producers to consumers (see Uni for more info)

type DerivedItemType = DerivedItemType

The payload type this Uni’s Streams will yield

type MutinyStreamType = MutinyStream<'static, ItemType, UniChannelType, DerivedItemType>

Defined as MutinyStream<'static, ItemType, UniChannelType, DerivedItemType>,
the concrete type for the Stream of DerivedItemTypes to be given to consumers

fn new<IntoString: Into<String>>(uni_name: IntoString) -> Self

Creates a Uni, which implements the consumer pattern, capable of: