+//! Streams of discrete events
+
+use std::sync::{ Arc, RwLock, Mutex, Weak };
+use std::sync::mpsc::{ Receiver, channel };
+use std::thread;
+use source::{ Source, CallbackError, CallbackResult, with_weak };
+use signal::{ self, Signal, SignalMut, sample_raw };
+use transaction::{ commit, later };
+
+
+/// An event sink.
+///
+/// This primitive is a way of generating streams of events. One can send
+/// input values into a sink and generate a stream that fires all these inputs
+/// as events:
+///
+/// ```
+/// # use carboxyl::Sink;
+/// // A new sink
+/// let sink = Sink::new();
+///
+/// // Make an iterator over a stream.
+/// let mut events = sink.stream().events();
+///
+/// // Send a value into the sink
+/// sink.send(5);
+///
+/// // The stream
+/// assert_eq!(events.next(), Some(5));
+/// ```
+///
+/// You can also feed a sink with an iterator:
+///
+/// ```
+/// # use carboxyl::Sink;
+/// # let sink = Sink::new();
+/// # let mut events = sink.stream().events();
+/// sink.feed(20..40);
+/// assert_eq!(events.take(4).collect::<Vec<_>>(), vec![20, 21, 22, 23]);
+/// ```
+///
+/// # Asynchronous calls
+///
+/// It is possible to send events into the sink asynchronously using the methods
+/// `send_async` and `feed_async`. Note though, that this will void some
+/// guarantees on the order of events. In the following example, it is unclear,
+/// which event is the first in the stream:
+///
+/// ```
+/// # use carboxyl::Sink;
+/// let sink = Sink::new();
+/// let mut events = sink.stream().events();
+/// sink.send_async(13);
+/// sink.send_async(22);
+/// let first = events.next().unwrap();
+/// assert!(first == 13 || first == 22);
+/// ```
+///
+/// `feed_async` provides a workaround, as it preserves the order of events from
+/// the iterator. However, any event sent into the sink after a call to it, may
+/// come at any point between the iterator events.
+pub struct Sink<A> {
+    source: Arc<RwLock<Source<A>>>,
+}
+
+impl<A> Clone for Sink<A> {
+    fn clone(&self) -> Sink<A> {
+        Sink { source: self.source.clone() }
+    }
+}
+
+impl<A: Send + Sync> Sink<A> {
+    /// Create a new sink.
+    pub fn new() -> Sink<A> {
+        Sink { source: Arc::new(RwLock::new(Source::new())) }
+    }
+
+    /// Generate a stream that fires all events sent into the sink.
+    pub fn stream(&self) -> Stream<A> {
+        Stream { source: self.source.clone(), keep_alive: Box::new(()), }
+    }
+}
+
+impl<A: Send + Sync + Clone + 'static> Sink<A> {
+    /// Asynchronous send.
+    ///
+    /// Same as `send`, but it spawns a new thread to process the updates to
+    /// dependent streams and signals.
+    pub fn send_async(&self, a: A) {
+        let clone = self.clone();
+        thread::spawn(move || clone.send(a));
+    }
+
+    /// Feed values from an iterator into the sink.
+    ///
+    /// This method feeds events into the sink from an iterator.
+    pub fn feed<I: IntoIterator<Item=A>>(&self, iterator: I) {
+        for event in iterator {
+            self.send(event);
+        }
+    }
+
+    /// Asynchronous feed.
+    ///
+    /// This is the same as `feed`, but it does not block, since it spawns the
+    /// feeding as a new task. This is useful, if the provided iterator is large
+    /// or even infinite (e.g. an I/O event loop).
+    pub fn feed_async<I: IntoIterator<Item=A> + Send + 'static>(&self, iterator: I) {
+        let clone = self.clone();
+        thread::spawn(move || clone.feed(iterator));
+    }
+
+    /// Send a value into the sink.
+    ///
+    /// When a value is sent into the sink, an event is fired in all dependent
+    /// streams.
+    pub fn send(&self, a: A) {
+        commit(|| self.source.write().unwrap().send(a))
+    }
+}
+
+
+/// Trait to wrap cloning of boxed values in a object-safe manner
+pub trait BoxClone: Sync + Send {
+    /// Clone the object as a boxed trait object
+    fn box_clone(&self) -> Box<BoxClone>; 
+}
+
+impl<T: Sync + Send + Clone + 'static> BoxClone for T {
+    fn box_clone(&self) -> Box<BoxClone> {
+        Box::new(self.clone())
+    }
+}
+
+
+/// Access a stream's source.
+///
+/// This is not defined as a method, so that it can be public to other modules
+/// in this crate while being private outside the crate.
+pub fn source<A>(stream: &Stream<A>) -> &Arc<RwLock<Source<A>>> {
+    &stream.source
+}
+
+
+/// A stream of events.
+///
+/// Conceptually a stream can be thought of as a series of discrete events that
+/// occur at specific times. They are ordered by a transaction system. This
+/// means that firings of disjoint events can not interfere with each other. The
+/// consequences of one event are atomically reflected in dependent quantities.
+///
+/// Streams provide a number of primitive operations. These can be used to
+/// compose streams and combine them with signals. For instance, streams can be
+/// mapped over with a function, merged with another stream of the same type or
+/// filtered by some predicate.
+///
+/// # Algebraic laws
+///
+/// Furthermore, streams satisfy certain algebraic properties that are useful to
+/// reason about them.
+///
+/// ## Monoid
+///
+/// For once, streams of the same type form a **monoid** under merging. The
+/// neutral element in this context is `Stream::never()`. So the following laws
+/// always hold for streams `a`, `b` and `c` of the same type:
+///
+/// - Left identity: `Stream::never().merge(&a) == a`,
+/// - Right identity: `a.merge(&Stream::never()) == a`,
+/// - Associativity: `a.merge(&b).merge(&c) == a.merge(&b.merge(&c))`.
+///
+/// *Note that equality in this context is not actually implemented as such,
+/// since comparing two (potentially infinite) streams is a prohibitive
+/// operation. Instead, the expressions above can be used interchangably and
+/// behave identically.*
+///
+/// ## Functor
+///
+/// Under the mapping operation streams also become a functor. A functor is a
+/// generic type like `Stream` with some mapping operation that takes a function
+/// `Fn(A) -> B` to map a `Stream<A>` to a `Stream<B>`. Algebraically it
+/// satisfies the following laws:
+///
+/// - The identity function is preserved: `a.map(|x| x) == a`,
+/// - Function composition is respected: `a.map(f).map(g) == a.map(|x| g(f(x)))`.
+pub struct Stream<A> {
+    source: Arc<RwLock<Source<A>>>,
+    #[allow(dead_code)]
+    keep_alive: Box<BoxClone>,
+}
+
+impl<A> Clone for Stream<A> {
+    fn clone(&self) -> Stream<A> {
+        Stream {
+            source: self.source.clone(),
+            keep_alive: self.keep_alive.box_clone(),
+        }
+    }
+}
+
+impl<A: Clone + Send + Sync + 'static> Stream<A> {
+    /// Create a stream that never fires. This can be useful in certain
+    /// situations, where a stream is logically required, but no events are
+    /// expected.
+    pub fn never() -> Stream<A> {
+        Stream {
+            source: Arc::new(RwLock::new(Source::new())),
+            keep_alive: Box::new(()) 
+        }
+    }
+
+    /// Map the stream to another stream using a function.
+    ///
+    /// `map` applies a function to every event fired in this stream to create a
+    /// new stream of type `B`.
+    ///
+    /// ```
+    /// # use carboxyl::Sink;
+    /// let sink: Sink<i32> = Sink::new();
+    /// let mut events = sink.stream().map(|x| x + 4).events();
+    /// sink.send(3);
+    /// assert_eq!(events.next(), Some(7));
+    /// ```
+    pub fn map<B, F>(&self, f: F) -> Stream<B>
+        where B: Send + Sync + Clone + 'static,
+              F: Fn(A) -> B + Send + Sync + 'static,
+    {
+        commit(|| {
+            let src = Arc::new(RwLock::new(Source::new()));
+            let weak = src.downgrade();
+            self.source.write().unwrap()
+                .register(move |a| with_weak(&weak, |src| src.send(f(a))));
+            Stream {
+                source: src,
+                keep_alive: Box::new(self.clone()),
+            }
+        })
+    }
+
+    /// Filter a stream according to a predicate.
+    ///
+    /// `filter` creates a new stream that only fires those events from the
+    /// original stream that satisfy the predicate.
+    ///
+    /// ```
+    /// # use carboxyl::Sink;
+    /// let sink: Sink<i32> = Sink::new();
+    /// let mut events = sink.stream()
+    ///     .filter(|&x| (x >= 4) && (x <= 10))
+    ///     .events();
+    /// sink.send(2); // won't arrive
+    /// sink.send(5); // will arrive
+    /// assert_eq!(events.next(), Some(5));
+    /// ```
+    pub fn filter<F>(&self, f: F) -> Stream<A>
+        where F: Fn(&A) -> bool + Send + Sync + 'static,
+    {
+        self.filter_map(move |a| if f(&a) { Some(a) } else { None })
+    }
+
+    /// Both filter and map a stream.
+    ///
+    /// This is equivalent to `.map(f).filter_some()`.
+    ///
+    /// ```
+    /// # use carboxyl::Sink;
+    /// let sink = Sink::new();
+    /// let mut events = sink.stream()
+    ///     .filter_map(|i| if i > 3 { Some(i + 2) } else { None })
+    ///     .events();
+    /// sink.send(2);
+    /// sink.send(4);
+    /// assert_eq!(events.next(), Some(6));
+    /// ```
+    pub fn filter_map<B, F>(&self, f: F) -> Stream<B>
+        where B: Send + Sync + Clone + 'static,
+              F: Fn(A) -> Option<B> + Send + Sync + 'static,
+    {
+        self.map(f).filter_some()
+    }
+
+    /// Merge with another stream.
+    ///
+    /// `merge` takes two streams and creates a new stream that fires events
+    /// from both input streams.
+    ///
+    /// ```
+    /// # use carboxyl::Sink;
+    /// let sink_1 = Sink::<i32>::new();
+    /// let sink_2 = Sink::<i32>::new();
+    /// let mut events = sink_1.stream().merge(&sink_2.stream()).events();
+    /// sink_1.send(2);
+    /// assert_eq!(events.next(), Some(2));
+    /// sink_2.send(4);
+    /// assert_eq!(events.next(), Some(4));
+    /// ```
+    pub fn merge(&self, other: &Stream<A>) -> Stream<A> {
+        commit(|| {
+            let src = Arc::new(RwLock::new(Source::new()));
+            for parent in [self, other].iter() {
+                let weak = src.downgrade();
+                parent.source.write().unwrap()
+                    .register(move |a| with_weak(&weak, |src| src.send(a)));
+            }
+            Stream {
+                source: src,
+                keep_alive: Box::new((self.clone(), other.clone())),
+            }
+        })
+    }
+
+    /// Coalesce multiple event firings within the same transaction into a
+    /// single event.
+    ///
+    /// The function should ideally commute, as the order of events within a
+    /// transaction is not well-defined.
+    pub fn coalesce<F>(&self, f: F) -> Stream<A>
+        where F: Fn(A, A) -> A + Send + Sync + 'static,
+    {
+        commit(|| {
+            let src = Arc::new(RwLock::new(Source::new()));
+            let weak = src.downgrade();
+            self.source.write().unwrap().register({
+                let mutex = Arc::new(Mutex::new(None));
+                move |a| {
+                    let mut inner = mutex.lock().unwrap();
+                    *inner = Some(match inner.take() {
+                        Some(b) => f(a, b),
+                        None => a,
+                    });
+                    // Send the updated value later
+                    later({
+                        let mutex = mutex.clone();
+                        let weak = weak.clone();
+                        move || {
+                            let mut inner = mutex.lock().unwrap();
+                            // Take it out and map, so that it does not happen twice
+                            inner.take().map(|value|
+                                with_weak(&weak, |src| src.send(value))
+                            );
+                        }
+                    });
+                    Ok(())
+                }
+            });
+            Stream { source: src, keep_alive: Box::new(self.clone()) }
+        })
+    }
+
+    /// Hold an event in a signal.
+    ///
+    /// The resulting signal `hold`s the value of the last event fired by the
+    /// stream.
+    ///
+    /// ```
+    /// # use carboxyl::Sink;
+    /// let sink = Sink::new();
+    /// let signal = sink.stream().hold(0);
+    /// assert_eq!(signal.sample(), 0);
+    /// sink.send(2);
+    /// assert_eq!(signal.sample(), 2);
+    /// ```
+    pub fn hold(&self, initial: A) -> Signal<A> {
+        signal::hold(initial, self)
+    }
+
+    /// A blocking iterator over the stream.
+    pub fn events(&self) -> Events<A> { Events::new(self) }
+
+    /// Scan a stream and accumulate its event firings in a signal.
+    ///
+    /// Starting at some initial value, each new event changes the value of the
+    /// resulting signal as prescribed by the supplied function.
+    ///
+    /// ```
+    /// # use carboxyl::Sink;
+    /// let sink = Sink::new();
+    /// let sum = sink.stream().scan(0, |a, b| a + b);
+    /// assert_eq!(sum.sample(), 0);
+    /// sink.send(2);
+    /// assert_eq!(sum.sample(), 2);
+    /// sink.send(4);
+    /// assert_eq!(sum.sample(), 6);
+    /// ```
+    pub fn scan<B, F>(&self, initial: B, f: F) -> Signal<B>
+        where B: Send + Sync + Clone + 'static,
+              F: Fn(B, A) -> B + Send + Sync + 'static,
+    {
+        Signal::cyclic(|scan| scan.snapshot(self, f).hold(initial))
+    }
+
+    /// Scan a stream and accumulate its event firings in some mutable state.
+    ///
+    /// Semantically this is equivalent to `scan`. However, it allows one to use
+    /// a non-Clone type as an accumulator and update it with efficient in-place
+    /// operations.
+    ///
+    /// The resulting `SignalMut` does have a slightly different API from a
+    /// regular `Signal` as it does not allow clones.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// # use carboxyl::{ Sink, Signal };
+    /// let sink: Sink<i32> = Sink::new();
+    /// let sum = sink.stream()
+    ///     .scan_mut(0, |sum, a| *sum += a)
+    ///     .combine(&Signal::new(()), |sum, ()| *sum);
+    /// assert_eq!(sum.sample(), 0);
+    /// sink.send(2);
+    /// assert_eq!(sum.sample(), 2);
+    /// sink.send(4);
+    /// assert_eq!(sum.sample(), 6);
+    /// ```
+    pub fn scan_mut<B, F>(&self, initial: B, f: F) -> SignalMut<B>
+        where B: Send + Sync + 'static,
+              F: Fn(&mut B, A) + Send + Sync + 'static,
+    {
+        signal::scan_mut(self, initial, f)
+    }
+}
+
+impl<A: Clone + Send + Sync + 'static> Stream<Option<A>> {
+    /// Filter a stream of options.
+    ///
+    /// `filter_some` creates a new stream that only fires the unwrapped
+    /// `Some(…)` events from the original stream omitting any `None` events.
+    ///
+    /// ```
+    /// # use carboxyl::Sink;
+    /// let sink = Sink::new();
+    /// let mut events = sink.stream().filter_some().events();
+    /// sink.send(None); // won't arrive
+    /// sink.send(Some(5)); // will arrive
+    /// assert_eq!(events.next(), Some(5));
+    /// ```
+    pub fn filter_some(&self) -> Stream<A> {
+        commit(|| {
+            let src = Arc::new(RwLock::new(Source::new()));
+            let weak = src.downgrade();
+            self.source.write().unwrap()
+                .register(move |a| a.map_or(
+                    Ok(()),
+                    |a| with_weak(&weak, |src| src.send(a))
+                ));
+            Stream {
+                source: src,
+                keep_alive: Box::new(self.clone())
+            }
+        })
+    }
+}
+
+impl<A: Send + Sync + Clone + 'static> Stream<Stream<A>> {
+    /// Switch between streams.
+    ///
+    /// This takes a stream of streams and maps it to a new stream, which fires
+    /// all events from the most recent stream fired into it.
+    ///
+    /// # Example
+    ///
+    /// ```
+    /// # use carboxyl::{ Sink, Stream };
+    /// // Create sinks
+    /// let stream_sink: Sink<Stream<i32>> = Sink::new();
+    /// let sink1: Sink<i32> = Sink::new();
+    /// let sink2: Sink<i32> = Sink::new();
+    ///
+    /// // Switch and listen
+    /// let switched = stream_sink.stream().switch();
+    /// let mut events = switched.events();
+    ///
+    /// // Should not receive events from either sink
+    /// sink1.send(1); sink2.send(2);
+    ///
+    /// // Now switch to sink 2
+    /// stream_sink.send(sink2.stream());
+    /// sink1.send(3); sink2.send(4);
+    /// assert_eq!(events.next(), Some(4));
+    ///
+    /// // And then to sink 1
+    /// stream_sink.send(sink1.stream());
+    /// sink1.send(5); sink2.send(6);
+    /// assert_eq!(events.next(), Some(5));
+    /// ```
+    pub fn switch(&self) -> Stream<A> {
+        fn rewire_callbacks<A>(new_stream: Stream<A>, source: Weak<RwLock<Source<A>>>,
+                               terminate: &mut Arc<()>)
+            -> CallbackResult
+            where A: Send + Sync + Clone + 'static,
+        {
+            *terminate = Arc::new(());
+            let weak = terminate.downgrade();
+            new_stream.source.write().unwrap().register(move |a|
+                weak.upgrade()
+                    .ok_or(CallbackError::Disappeared)
+                    .and_then(|_| with_weak(&source, |src| src.send(a)))
+            );
+            Ok(())
+        }
+        commit(|| {
+            let src = Arc::new(RwLock::new(Source::new()));
+            let weak = src.downgrade();
+            self.source.write().unwrap().register({
+                let mut terminate = Arc::new(());
+                move |stream| rewire_callbacks(stream, weak.clone(), &mut terminate)
+            });
+            Stream {
+                source: src,
+                keep_alive: Box::new(self.clone()),
+            }
+        })
+    }
+}
+
+
+/// Make a snapshot of a signal, whenever a stream fires an event.
+pub fn snapshot<A, B, C, F>(signal: &Signal<A>, stream: &Stream<B>, f: F) -> Stream<C>
+    where A: Clone + Send + Sync + 'static,
+          B: Clone + Send + Sync + 'static,
+          C: Clone + Send + Sync + 'static,
+          F: Fn(A, B) -> C + Send + Sync + 'static,
+{
+    commit(|| {
+        let src = Arc::new(RwLock::new(Source::new()));
+        let weak = src.downgrade();
+        stream.source.write().unwrap().register({
+            let signal = signal.clone();
+            move |b| with_weak(&weak, |src| src.send(f(sample_raw(&signal), b)))
+        });
+        Stream {
+            source: src,
+            keep_alive: Box::new((stream.clone(), signal.clone())),
+        }
+    })
+}
+
+
+/// A blocking iterator over events in a stream.
+pub struct Events<A> {
+    receiver: Receiver<A>,
+    #[allow(dead_code)]
+    keep_alive: Box<BoxClone>,
+}
+
+impl<A: Send + Sync + 'static> Events<A> {
+    /// Create a new events iterator.
+    fn new(stream: &Stream<A>) -> Events<A> {
+        commit(|| {
+            let (tx, rx) = channel();
+            let tx = Mutex::new(tx);
+            stream.source.write().unwrap().register(
+                move |a| tx.lock().unwrap().send(a).map_err(|_| CallbackError::Disappeared)
+            );
+            Events {
+                receiver: rx,
+                keep_alive: Box::new(stream.clone()),
+            }
+        })
+    }
+}
+
+impl<A: Send + Sync + 'static> Iterator for Events<A> {
+    type Item = A;
+    fn next(&mut self) -> Option<A> { self.receiver.recv().ok() }
+}
+
+
+#[cfg(test)]
+mod test {
+    use std::thread;
+    use quickcheck::quickcheck;
+
+    use testing::{ id, stream_eq };
+    use super::*;
+
+    #[test]
+    fn sink() {
+        let sink = Sink::new();
+        let mut events = sink.stream().events();
+        sink.send(1);
+        sink.send(2);
+        assert_eq!(events.next(), Some(1));
+        assert_eq!(events.next(), Some(2));
+    }
+
+    #[test]
+    fn map() {
+        let sink = Sink::new();
+        let triple = sink.stream().map(|x| 3 * x);
+        let mut events = triple.events();
+        sink.send(1);
+        assert_eq!(events.next(), Some(3));
+    }
+
+    #[test]
+    fn filter_some() {
+        let sink = Sink::new();
+        let small = sink.stream().filter_some();
+        let mut events = small.events();
+        sink.send(None);
+        sink.send(Some(9));
+        assert_eq!(events.next(), Some(9));
+    }
+
+    #[test]
+    fn chain_1() {
+        let sink: Sink<i32> = Sink::new();
+        let chain = sink.stream()
+            .map(|x| x / 2)
+            .filter(|&x| x < 3);
+        let mut events = chain.events();
+        sink.send(7);
+        sink.send(4);
+        assert_eq!(events.next(), Some(2));
+    }
+
+    #[test]
+    fn merge() {
+        let sink1 = Sink::new();
+        let sink2 = Sink::new();
+        let mut events = sink1.stream().merge(&sink2.stream()).events();
+        sink1.send(12);
+        sink2.send(9);
+        assert_eq!(events.next(), Some(12));
+        assert_eq!(events.next(), Some(9));
+    }
+
+    #[test]
+    fn chain_2() {
+        let sink1: Sink<i32> = Sink::new();
+        let sink2: Sink<i32> = Sink::new();
+        let mut events = sink1.stream().map(|x| x + 4)
+            .merge(
+                &sink2.stream()
+                .filter_map(|x| if x < 4 { Some(x) } else { None })
+                .map(|x| x * 5))
+            .events();
+        sink1.send(12);
+        sink2.send(3);
+        assert_eq!(events.next(), Some(16));
+        assert_eq!(events.next(), Some(15));
+    }
+
+    #[test]
+    fn move_closure() {
+        let sink = Sink::<i32>::new();
+        let x = 3;
+        sink.stream().map(move |y| y + x);
+    }
+
+    #[test]
+    fn scan_race_condition() {
+        let sink = Sink::new();
+        // Feed the sink in the background
+        sink.feed_async(0..100000);
+        // Try it multiple times to increase failure probability, when a data
+        // race can potentially happen.
+        for _ in 0..10 {
+            let _sum = sink.stream().scan(0, |a, b| a + b);
+        }
+    }
+
+    #[test]
+    fn sink_send_async() {
+        let sink = Sink::new();
+        let mut events = sink.stream().events();
+        sink.send_async(1);
+        assert_eq!(events.next(), Some(1));
+    }
+
+    #[test]
+    fn sink_feed() {
+        let sink = Sink::new();
+        let events = sink.stream().events();
+        sink.feed(0..10);
+        for (n, m) in events.take(10).enumerate() {
+            assert_eq!(n as i32, m);
+        }
+    }
+
+    #[test]
+    fn sink_feed_async() {
+        let sink = Sink::new();
+        let events = sink.stream().events();
+        sink.feed_async(0..10);
+        for (n, m) in events.take(10).enumerate() {
+            assert_eq!(n as i32, m);
+        }
+    }
+
+    #[test]
+    fn coalesce() {
+        let sink = Sink::new();
+        let stream = sink.stream()
+            .merge(&sink.stream())
+            .coalesce(|a, b| a + b);
+        let mut events = stream.events();
+
+        sink.send(1);
+        assert_eq!(events.next(), Some(2));
+    }
+
+    #[test]
+    fn monoid_left_identity() {
+        fn check(input: Vec<i32>) -> Result<bool, String> {
+            let sink = Sink::new();
+            let a = sink.stream();
+            let eq = stream_eq(&Stream::never().merge(&a), &a);
+            sink.feed(input.into_iter());
+            eq.sample()
+        }
+        quickcheck(check as fn(Vec<i32>) -> Result<bool, String>);
+    }
+
+    #[test]
+    fn monoid_right_identity() {
+        fn check(input: Vec<i32>) -> Result<bool, String> {
+            let sink = Sink::new();
+            let a = sink.stream();
+            let eq = stream_eq(&a.merge(&Stream::never()), &a);
+            sink.feed(input.into_iter());
+            eq.sample()
+        }
+        quickcheck(check as fn(Vec<i32>) -> Result<bool, String>);
+    }
+
+    #[test]
+    fn monoid_associative() {
+        fn check(input_a: Vec<i32>, input_b: Vec<i32>, input_c: Vec<i32>) -> Result<bool, String> {
+            let sink_a = Sink::new();
+            let sink_b = Sink::new();
+            let sink_c = Sink::new();
+            let a = sink_a.stream();
+            let b = sink_b.stream();
+            let c = sink_c.stream();
+            let eq = stream_eq(&a.merge(&b.merge(&c)), &a.merge(&b).merge(&c));
+            /* feed in parallel */ {
+                let _g1 = thread::scoped(|| sink_a.feed(input_a.into_iter()));
+                let _g2 = thread::scoped(|| sink_b.feed(input_b.into_iter()));
+                let _g3 = thread::scoped(|| sink_c.feed(input_c.into_iter()));
+            }
+            eq.sample()
+        }
+        quickcheck(check as fn(Vec<i32>, Vec<i32>, Vec<i32>) -> Result<bool, String>);
+    }
+
+    #[test]
+    fn functor_identity() {
+        fn check(input: Vec<i32>) -> Result<bool, String> {
+            let sink = Sink::new();
+            let a = sink.stream();
+            let eq = stream_eq(&a.map(id), &a);
+            sink.feed(input.into_iter());
+            eq.sample()
+        }
+        quickcheck(check as fn(Vec<i32>) -> Result<bool, String>);
+    }
+
+    #[test]
+    fn functor_composition() {
+        fn check(input: Vec<i32>) -> Result<bool, String> {
+            fn f(n: i32) -> i64 { (n + 3) as i64 }
+            fn g(n: i64) -> f64 { n as f64 / 2.5 }
+
+            let sink = Sink::new();
+            let a = sink.stream();
+            let eq = stream_eq(&a.map(f).map(g), &a.map(|n| g(f(n))));
+            sink.feed(input.into_iter());
+            eq.sample()
+        }
+        quickcheck(check as fn(Vec<i32>) -> Result<bool, String>);
+    }
+}
+

+

Keyboard shortcuts

Search tricks