1 unstable release
Uses old Rust 2015
0.0.4 | Apr 15, 2017 |
---|
#1896 in Rust patterns
14KB
315 lines
oyashio
A single producer, multiple consumer stream
- consumer streams can be cloned
- are consumed independently
- data sits in central arc
- polling and await style consumption
- plugs in to regular iterator functions
example
extern crate oyashio;
use oyashio::{Stream,StreamRMerge};
fn main () {
// stream stealing examples
let (mut st, mut sr) = Stream::default();
let mut sr2 = sr.clone();
for n in (0..10) { st.send(n) }
let r2 = sr2.take(4).collect::<Vec<&u8>>();
let r = sr.take(4).collect::<Vec<&u8>>();
println!("{:?},{:?}",r2,r); //[0, 1, 2, 3],[4, 5, 6, 7]
// broadcast examples
let (mut st, mut sr) = Stream::new_broadcast();
for n in (0..10) { st.send(n) }
// regular iterator, may wait for values to be sent
let r = sr.clone().filter(|x|*x%2==1)
.take(4) //only wait on up to 4 odd numbers
.collect::<Vec<&u8>>();
println!("{:?}",r); //[1, 3, 5, 7]
let mut vr = vec!();
for (i,n) in sr.clone().enumerate() { //iterates til break
vr.push(n);
assert_eq!(*n,i as u8);
if i == 9 {break;} //let's break out of this
}
println!("{:?}",vr); //[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
vr = vec!();
// polling iterator, never waits
for n in sr.poll() { vr.push(n); }
println!("{:?}",vr); //[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
//stream merge example
let (mut st2,mut sr2) = Stream::new_broadcast();
let (mut st, mut sr) = Stream::new_broadcast();
let (mut st3, mut sr3) = Stream::new_broadcast();
for n in (0i32..2) { st.send(n); } //0,1
let mut vr: Vec<i32> = vec!();
let mut vsr = vec![sr,sr2,sr3];
let sm = StreamRMerge::new(vsr);
for n in (2..4) { st2.send(n); } //2,3
for n in (4..6) { st3.send(n); } //4,5
for n in sm.clone() { vr.push(*n); }
assert_eq!(vr,&[0,2,4,1,3,5]);
}
todo
- create optional bounded size streams
- let streams consumption start and stop without 'moving' stream variable (eg: stream->filter->take->collect, then stream->enumerate() without necessary cloning)
- try in real world scenarios
Dependencies
~340–570KB