#![no_std]
#![allow(clippy::needless_range_loop)]
#![allow(clippy::type_complexity)]
#![allow(async_fn_in_trait)]
//! Abstractions for asynchronously working with series of data (“streams” and “sinks”).
//!
//! See the [`producer`] and [`consumer`] modules for thorough introductions to the designs. See the [ufotofu website](https://worm-blossom.org/ufotofu/) for a discussion of our motivating design choices — these crate docs stay focussed on the *what*, not the *why*.
//!
//! ## Caveats
//!
//! Ufotofu adheres to the principles of [frugal async rust](https://worm-blossom.org/frugal_async_rust.html):
//!
//! - The futures returned by async ufotofu methods are `!Send`, they cannot be run on multi-threaded executors.
//! - Dropping any method-returned future before polling it to completion will leave the original object in an undefined state; subsequent method calls may display arbitrary (but always safe) behaviour.
//! - Unwinding any panic may leave ufotofu values in an undefined state. Do not attempt to recover from panics when using ufotofu.
//!
//! ## Module Overview
//!
//! The two central modules are [`producer`] and [`consumer`], they define the core abstractions of the crate.
//!
//! The [`queues`] module provides the [`Queue`](queues::Queue) trait for infallible in-memory queues with bulk push and pop operations, and some types implementing it. These power the buffered producer and consumer implementations of ufotofu.
//!
//! The [`channels`] module provides ordered in-memory communication channels, which implement [`BulkProducer`] and [`BulkConsumer`] on their endpoints.
//!
//! The [`codec`] module provides traits and functionality around asynchronous encoding and decoding. The [`codec_relative`] module provides analogous functionality for settings where encodings can make use of shared context between encoder and decoder.
//!
//! The [`fuzz_testing_tutorial`] demonstrates the utility types for [fuzz testing](https://rust-fuzz.github.io/book/introduction.html) that ufotofu provides.
//!
//! ## Actual I/O
//!
//! The ufotofu abstractions are nice and all, but how do you actually do any I/O with ufotofu? The [`producer::compat::reader::reader_to_bulk_producer`] function turns any [`AsyncRead`](futures_lite::AsyncRead) into a [`BulkProducer`], and the [`consumer::compat::writer::writer_to_bulk_consumer`] function turns any [`AsyncWrite`](futures_lite::AsyncWrite) into a [`BulkConsumer`] (both functions require the `compat_futures_io` feature to be activated). This way, you can use established crates such as [`smol`](https://crates.io/crates/smol) or [`tokio`](https://crates.io/crates/tokio) with ufotofu abstractions.
//!
//! ## Compared to `futures-rs`
//!
//! The traits of ufotofu roughly correspond to the traits of the [`futures`](https://crates.io/crates/futures) crate in the following way:
//!
//! | `ufotofu` | `futures-rs` |
//! |---|---|
//! | [`Producer`] | [`Stream`](https://docs.rs/futures/0.3.31/futures/prelude/trait.Stream.html) |
//! | [`BulkProducer`] | [`AsyncBufRead`](https://docs.rs/futures/0.3.31/futures/io/trait.AsyncBufRead.html), but for arbitrary item and error types |
//! | [`Consumer`] | [`Sink`](https://docs.rs/futures/0.3.31/futures/prelude/trait.Sink.html) |
//! | [`BulkConsumer`] | [`AsyncWrite`](https://docs.rs/futures/0.3.31/futures/io/trait.AsyncWrite.html) (because there is no `AsyncBufWrite`), but for arbitrary item and error types |
//!
//! For an overview of the core design differences between ufotofu and the [`futures`](https://crates.io/crates/futures) crate, see [the ufotofu website](https://worm-blossom.org/ufotofu/#design_choices): each listed design choice is a point of departure from how the [`futures`](https://crates.io/crates/futures) crate does things.
//!
//! Do note in particular that the traits of ufotofu are not [dyn-compatible](https://doc.rust-lang.org/1.88.0/reference/items/traits.html#dyn-compatibility), and nothing in ufotofu is [`Send`], [cancel-safe](https://rfd.shared.oxide.computer/rfd/400), or [exception-safe](https://doc.rust-lang.org/nomicon/exception-safety.html).
#[cfg(feature = "std")]
extern crate std;
#[cfg(feature = "alloc")]
extern crate alloc;
// We re-export Either here so we can reliably match against it in the macros we export. We hide it from our docs though.
#[doc(hidden)]
pub use either::Either;
use prelude::*;
/// Conveniently consume the output of a [`Producer`].
///
/// This macro provides a generalisation of a `for` loop. It first converts a value into a producer via [`IntoProducer`], and then repeatedly calls `produce` until the final value or an error is emitted. The macro has three branches to specify how to handle regular items, final valu, and/or error respectively.
///
/// ```
/// use ufotofu::prelude::*;
/// # fn main() {
/// # pollster::block_on(async{
///
/// // The macro converts `[1, 2, 4]` into a producer via `[1, 2, 4].into_producer()`.
/// consume![[1, 2, 4] {
/// item it => print!("{it}, "),
/// final () => {
/// print!("and ");
/// println!("done!");
/// }
/// error _err => unreachable!("this producer is infallible"),
/// }];
/// // Prints `1, 2, 4, and done!`.
/// # Result::<(), Infallible>::Ok(())
/// # });
/// # }
/// ```
///
/// The `error` branch is optional — without it, the macro implicitly applies the `?` to the result of calling `produce`.
///
/// ```
/// use ufotofu::prelude::*;
/// # fn main() {
/// # pollster::block_on(async{
///
/// consume![[1, 2, 4] {
/// item it => print!("{it}, "),
/// final () => println!("and done!"),
/// }];
/// # Result::<(), Infallible>::Ok(())
/// # });
/// # }
/// ```
///
/// The `final` branch is also optional, but only if [`Producer::Final`] is `()`.
///
/// ```
/// use ufotofu::prelude::*;
/// # fn main() {
/// # pollster::block_on(async{
///
/// consume![[1, 2, 4] {
/// item it => print!("{it}, "),
/// // Could also omit the `error` branch.
/// error _err => unreachable!("this producer is infallible"),
/// }];
/// // Prints `1, 2, 4, `.
/// # Result::<(), Infallible>::Ok(())
/// # });
/// # }
/// ```
///
/// The `item` branch is the only mandatory branch. Each of the `item`, `final`, and `error` "keywords" can be followed by an arbitrary pattern. The order of the three kinds of branches is arbitrary.
///
/// You can specify multiple branches of the same kind, in order to match different patterns.
///
/// ```
/// use ufotofu::prelude::*;
/// # fn main() {
/// # pollster::block_on(async{
///
/// consume![[1, 2, 4] {
/// item 2 => print!("quack, "),
/// item it => print!("{it}, "),
/// final () => {
/// print!("and ");
/// println!("done!");
/// }
/// error _err => unreachable!("this producer is infallible"),
/// }];
/// // Prints `1, quack, 4, and done!`.
/// # Result::<(), Infallible>::Ok(())
/// # });
/// # }
/// ```
///
/// Finally, here is a demonstration of what the macro expands to when all three kinds of cases are present, slightly simplified for readability:
///
/// ```
/// use ufotofu::prelude::*;
/// # fn main() {
/// # pollster::block_on(async{
/// # let some_value = [1, 2, 4];
///
/// consume![some_value {
/// item 42 => println!("42"),
/// item pattern_item => println!("non-42 item"),
/// final pattern_final => println!("final"),
/// error pattern_error => println!("error"),
/// }];
/// # Result::<(), Infallible>::Ok(())
/// # });
/// # }
///
/// // Roughly expands to:
///
/// # fn expanded() {
/// # pollster::block_on(async{
/// # let some_value = [1, 2, 4];
/// # fn handle_item(){}
/// # fn handle_final(){}
/// # fn handle_error(){}
/// let mut producer = some_value.into_producer();
///
/// loop {
/// match producer.produce().await {
/// Ok(Left(42)) => println!("42"),
/// Ok(Left(pattern_item)) => println!("non-42 item"),
/// Ok(Right(pattern_final)) => println!("final"),
/// Err(pattern_error) => println!("error"),
/// }
/// }
/// # Result::<(), Infallible>::Ok(())
/// # });
/// # }
/// ```
///
/// <br/>Counterpart: none, because Rust has no counterpart to the `for` loop. In a certain sense, generators are this counterpart, but we have not implemented generator-like macros. Yet.
pub use ufotofu_macros::consume;
mod errors;
pub use errors::*;
pub mod producer;
pub use producer::{
BulkProducer, BulkProducerExt, IntoBulkProducer, IntoProducer, Producer, ProducerExt,
};
pub mod consumer;
pub use consumer::{
BulkConsumer, BulkConsumerExt, Consumer, ConsumerExt, IntoBulkConsumer, IntoConsumer,
};
pub mod channels;
pub mod queues;
pub mod codec;
pub mod codec_relative;
#[cfg(all(feature = "dev", feature = "alloc"))]
mod test_yielder;
#[cfg(feature = "dev")]
pub mod fuzz_testing_tutorial;
/// A “prelude” for crates using the `ufotofu` crate.
///
/// This prelude is similar to the standard library’s prelude in that you’ll almost always want to import its entire contents, but unlike the standard library’s prelude you’ll have to do so manually:
///
/// `use ufotofu::prelude::*;`
///
/// The prelude may grow over time.
pub mod prelude {
pub use crate::{
consume, consumer, producer, BulkConsumer, BulkConsumerExt, BulkProducer, BulkProducerExt,
Consumer, ConsumerExt, IntoBulkConsumer, IntoBulkProducer, IntoConsumer, IntoProducer,
Producer, ProducerExt,
};
#[cfg(feature = "dev")]
pub use crate::{
consumer::{
build_test_consumer, BulkConsumerOperation, ConsumerOperation, TestConsumer,
TestConsumerBuilder, TestConsumerBuilderError,
},
producer::{
build_test_producer, BulkProducerOperation, ProducerOperation, TestProducer,
TestProducerBuilder, TestProducerBuilderError,
},
};
pub use either::Either::{self, Left, Right};
pub use core::convert::Infallible;
}
/// An additional “prelude” for crates using the [`codec`] and/or [`codec_relative`] modules of `ufotofu`.
///
/// `use ufotofu::codec_prelude::*;`
///
/// The prelude may grow over time.
pub mod codec_prelude {
pub use crate::codec::{
self, endian, Blame, Decodable, DecodableCanonic, DecodeError, Encodable, EncodableExt,
EncodableKnownLength, EncodableKnownLengthExt,
};
pub use crate::codec_relative::{
self, RelativeDecodable, RelativeDecodableCanonic, RelativeEncodable, RelativeEncodableExt,
RelativeEncodableKnownLength, RelativeEncodableKnownLengthExt,
};
pub use crate::prelude::*;
}
/// Pipes as many items as possible from a [`Producer`] into a [`Consumer`]. Then closes the consumer with the final value emitted by the producer.
pub async fn pipe<P, C>(producer: P, consumer: C) -> Result<(), PipeError<P::Error, C::Error>>
where
P: IntoProducer,
C: IntoConsumer<Item = P::Item, Final = P::Final>,
{
let mut consumer = consumer.into_consumer();
consume![producer {
item it => consumer.consume_item(it).await.map_err(PipeError::Consumer)?,
final fin => Ok(consumer.consume_final(fin).await.map_err(PipeError::Consumer)?),
error err => Err(PipeError::Producer(err)),
}]
}
/// Efficiently pipes as many items as possible from a [`BulkProducer`] into a [`BulkConsumer`], using [`BulkConsumerExt::bulk_consume`]. Then closes consumer with the final value emitted by the producer.
pub async fn bulk_pipe<P, C>(producer: P, consumer: C) -> Result<(), PipeError<P::Error, C::Error>>
where
P: IntoBulkProducer<Item: Clone>,
C: IntoBulkConsumer<Item = P::Item, Final = P::Final>,
{
let mut p = producer.into_producer();
let mut c = consumer.into_consumer();
loop {
match p
.expose_items(async |items| match c.bulk_consume(items).await {
Ok(amount) => (amount, Ok(())),
Err(consumer_error) => (0, Err(consumer_error)),
})
.await
.map_err(PipeError::Producer)?
{
Left(Ok(())) => {
// No-op, continues with next loop iteration.
}
Left(Err(consumer_err)) => return Err(PipeError::Consumer(consumer_err)),
Right(fin) => return c.consume_final(fin).await.map_err(PipeError::Consumer),
}
}
}