FRP in Rust without reckless cloning

frp.rs

#![allow(dead_code)]
use std::rc::Rc;
use std::cell::RefCell;
use std::borrow::Cow;
use std::ptr;

// callbacks use a Cow<T> argument so we can choose at runtime if
// we will send a ref or an owned value
struct Callbacks<T: Clone>
{
    fs: Vec<Box<Fn(Cow<T>) -> bool>>,
}

impl<T: Clone> Callbacks<T>
{
    fn new() -> Self
    {
        Callbacks{ fs: Vec::new() }
    }

    fn push<F>(&mut self, cb: F)
        where F: Fn(Cow<T>) -> bool + 'static
    {
        self.fs.push(Box::new(cb));
    }

    // sends a ref to the first N-1 callbacks, and the owned value to the last
    // this way we prevent tons of cloning
    fn call(&mut self, arg: T)
    {
        let maybe_last = self.fs.pop();
        self.fs.retain(|f| f(Cow::Borrowed(&arg)));

        if let Some(last) = maybe_last
        {
            if last(Cow::Owned(arg))
            {
                self.fs.push(last);
            }
        }
    }
}

impl<T: Clone> Default for Callbacks<T>
{
    fn default() -> Self
    {
        Callbacks::new()
    }
}

// type erasure
trait Untyped {}
impl<T> Untyped for T {}

/// Represents a stream of discrete values sent over time
#[derive(Clone)]
struct Stream<T: Clone>
{
    cbs: Rc<RefCell<Callbacks<T>>>,
    source: Option<Rc<Untyped>>,  // strong reference to a parent Stream
}

impl<T: Clone> Stream<T>
{
    fn new() -> Self
    {
        Stream{ cbs: Default::default(), source: None }
    }

    fn send(&self, arg: T)
    {
        self.cbs.borrow_mut().call(arg)
    }

    /// Creates a new Stream that contains the transformed the value of this Stream
    fn map<F, R>(&self, f: F) -> Stream<R>
        where F: Fn(Cow<T>) -> R + 'static, R: Clone + 'static, T: 'static
    {
        let new_cbs = Rc::new(RefCell::new(Callbacks::new()));
        let weak = Rc::downgrade(&new_cbs);
        self.cbs.borrow_mut().push(move |arg| {
            weak.upgrade()
                .map(|cb| cb.borrow_mut().call(f(arg)))
                .is_some()
        });
        Stream{ cbs: new_cbs, source: Some(Rc::new(self.clone())) }
    }

    /// Read the value without modifying it
    fn inspect<F>(self, f: F) -> Self
        where F: Fn(Cow<T>) + 'static
    {
        self.cbs.borrow_mut().push(move |arg| { f(arg); true });
        self
    }

    /// Creates a Signal that holds the last value sent to this Stream
    fn hold_last(&self, initial: T) -> Signal<T>
        where T: 'static
    {
        let storage = Rc::new(RefCell::new(initial));
        let weak = Rc::downgrade(&storage);
        self.cbs.borrow_mut().push(move |arg| {
            weak.upgrade()
                .map(|st| *st.borrow_mut() = arg.into_owned())
                .is_some()
        });
        Signal{
            val: SigVal::Shared(storage),
            source: Some(Rc::new(self.clone()))
        }
    }

    /// Accumulates the values sent over a Stream
    fn fold<A, F>(&self, initial: A, f: F) -> Signal<A>
        where F: Fn(A, Cow<T>) -> A + 'static, A: Clone + 'static, T: 'static
    {
        let storage = Rc::new(RefCell::new(initial));
        let weak = Rc::downgrade(&storage);
        self.cbs.borrow_mut().push(move |arg| {
            weak.upgrade()
                .map(|st| {
                    let acc = &mut *st.borrow_mut();
                    let old = unsafe { ptr::read(acc) };
                    let new = f(old, arg);  // maybe should catch_unwind this
                    unsafe { ptr::write(acc, new) };
                })
                .is_some()
        });
        Signal{
            val: SigVal::Shared(storage),
            source: Some(Rc::new(self.clone()))
        }
    }
}

#[derive(Clone)]
enum SigVal<T: Clone>
{
    Constant(T),
    Shared(Rc<RefCell<T>>),
    Dynamic(Rc<Fn() -> T>),
}

impl<T: Clone> SigVal<T>
{
    fn from_fn<F>(f: F) -> Self
        where F: Fn() -> T + 'static
    {
        SigVal::Dynamic(Rc::new(f))
    }
}

// Represents a continuous value that changes over time
#[derive(Clone)]
struct Signal<T: Clone>
{
    val: SigVal<T>,
    source: Option<Rc<Untyped>>,
}

impl<T: Clone> Signal<T>
{
    fn constant(val: T) -> Self
    {
        Signal{ val: SigVal::Constant(val), source: None }
    }

    fn from_fn<F>(f: F) -> Self
        where F: Fn() -> T + 'static
    {
        Signal{ val: SigVal::from_fn(f), source: None }
    }

    /// Sample by value.
    /// This clones the content of the signal
    fn sample(&self) -> T
    {
        match self.val
        {
            SigVal::Constant(ref v) => v.clone(),
            SigVal::Shared(ref s) => s.borrow().clone(),
            SigVal::Dynamic(ref f) => f(),
        }
    }

    /// Sample by reference.
    /// This is meant to be the most efficient way when cloning is undesirable,
    /// but it requires a callback to prevent outliving the RefCell borrow
    fn sample_with<F, R>(&self, cb: F) -> R
        where F: FnOnce(Cow<T>) -> R
    {
        match self.val
        {
            SigVal::Constant(ref v) => cb(Cow::Borrowed(v)),
            SigVal::Shared(ref s) => cb(Cow::Borrowed(&s.borrow())),
            SigVal::Dynamic(ref f) => cb(Cow::Owned(f())),
        }
    }

    /// Transform the signal value
    fn map<F, R>(&self, f: F) -> Signal<R>
        where F: Fn(Cow<T>) -> R + 'static, R: Clone, T: 'static
    {
        let cloned = self.clone();
        Signal{
            val: SigVal::from_fn(move || cloned.sample_with(|r| f(r))),
            source: None
        }
    }

    /// Takes a snapshot of the Signal every time the trigger signal fires
    fn snapshot<S, F, R>(&self, trigger: &Stream<S>, f: F) -> Stream<R>
        where F: Fn(Cow<T>, Cow<S>) -> R + 'static, S: Clone + 'static, R: Clone + 'static, T: 'static
    {
        let cloned = self.clone();
        trigger.map(move |b| cloned.sample_with(|a| f(a, b)))
    }
}


// test case for the cloning issue
use std::fmt::Debug;

#[derive(Debug)]
struct Storage<T>
{
    vec: Vec<T>,
}

impl<T> Storage<T>
{
    fn new() -> Self
    {
        Storage{ vec: Vec::new() }
    }

    fn push(mut self, item: T) -> Self
    {
        self.vec.push(item);
        self
    }
}

impl<T: Clone + Debug> Clone for Storage<T>
{
    fn clone(&self) -> Self
    {
        println!("storage cloned! {:?}", self.vec);
        Storage{ vec: self.vec.clone() }
    }
}

fn main()
{
    let stream = Stream::new();
    let sig = stream.fold(Storage::new(), |a, v| a.push(*v));

    stream.send(11);
    stream.send(22);
    stream.send(33);

    sig.sample_with(|val| println!("result: {:?}", val));
}