#stream #reactive #iterable #observable #frp

callbag

Rust implementation of the callbag spec for reactive/iterable programming

13 breaking releases

0.14.0 Jan 14, 2022
0.13.0 Dec 28, 2021
0.12.0 Dec 25, 2021
0.6.0 Nov 23, 2021

#652 in Asynchronous

35 downloads per month

MIT/Apache

110KB
1.5K SLoC

callbag-rs

Rust implementation of the callbag spec for reactive/iterable programming.

Basic callbag factories and operators to get started with.

Highlights:

  • Supports reactive stream programming
  • Supports iterable programming (also!)
  • Same operator works for both of the above
  • Extensible

Imagine a hybrid between an Observable and an (Async)Iterable, that's what callbags are all about. It's all done with a few simple callbacks, following the callbag spec.

CI Crates.io Documentation MIT OR Apache-2.0 licensed

Examples

Reactive programming examples

Pick the first 5 odd numbers from a clock that ticks every second, then start observing them:

use async_nursery::Nursery;
use crossbeam_queue::SegQueue;
use std::{sync::Arc, time::Duration};

use callbag::{filter, for_each, interval, map, pipe, take};

let (nursery, nursery_out) = Nursery::new(async_executors::AsyncStd);

let actual = Arc::new(SegQueue::new());

pipe!(
    interval(Duration::from_millis(1_000), nursery.clone()),
    map(|x| x + 1),
    filter(|x| x % 2 == 1),
    take(5),
    for_each({
        let actual = Arc::clone(&actual);
        move |x| {
            println!("{}", x);
            actual.push(x);
        }
    }),
);

drop(nursery);
async_std::task::block_on(nursery_out);

assert_eq!(
    &{
        let mut v = vec![];
        for _i in 0..actual.len() {
            v.push(actual.pop().unwrap());
        }
        v
    }[..],
    [1, 3, 5, 7, 9]
);

Iterable programming examples

From a range of numbers, pick 5 of them and divide them by 4, then start pulling those one by one:

use crossbeam_queue::SegQueue;
use std::sync::Arc;

use callbag::{for_each, from_iter, map, pipe, take};

#[derive(Clone)]
struct Range {
    i: usize,
    to: usize,
}

impl Range {
    fn new(from: usize, to: usize) -> Self {
        Range { i: from, to }
    }
}

impl Iterator for Range {
    type Item = usize;

    fn next(&mut self) -> Option<Self::Item> {
        let i = self.i;
        if i <= self.to {
            self.i += 1;
            Some(i)
        } else {
            None
        }
    }
}

let actual = Arc::new(SegQueue::new());

pipe!(
    from_iter(Range::new(40, 99)),
    take(5),
    map(|x| x as f64 / 4.0),
    for_each({
        let actual = Arc::clone(&actual);
        move |x| {
            println!("{}", x);
            actual.push(x);
        }
    }),
);

assert_eq!(
    &{
        let mut v = vec![];
        for _i in 0..actual.len() {
            v.push(actual.pop().unwrap());
        }
        v
    }[..],
    [10.0, 10.25, 10.5, 10.75, 11.0]
);

Ok::<(), Box<dyn std::error::Error>>(())

API

The list below shows what's included.

Source factories

Sink factories

Transformation operators

Filtering operators

Combination operators

Utilities

Terminology

  • source: a callbag that delivers data
  • sink: a callbag that receives data
  • puller sink: a sink that actively requests data from the source
  • pullable source: a source that delivers data only on demand (on receiving a request)
  • listener sink: a sink that passively receives data from the source
  • listenable source: source which sends data to the sink without waiting for requests
  • operator: a callbag based on another callbag which applies some operation

License

Licensed under either of

at your option.

Contribution

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.

Acknowledgements

Thanks to André Staltz (@staltz) for creating the callbag spec.

This library is a port of https://github.com/staltz/callbag-basics. Some inspiration was taken from https://github.com/f5io/callbag.rs.

Many thanks to the awesome folks on the Rust Users Forum for their help, especially:

Dependencies

~185–460KB