Observer pattern in Rust

Question

A simple implementation of the Observer pattern in rust. The Observable constructor (::new) is called with a subscription function which will be called every time an observer subscribes.

The subscription function must return a cleanup function which is defined by the using code.

An observer subscribed to the Observable either by:

• passing a simple next function
• providing an Observer which provide a next, complete and error function
• providing a FullObserver which, in addition to the aforementioned function, provide also a start function which allows the Observer to be called before the processing start.

use std::error::Error;
use std::rc::Rc;

struct Observer<T, E> {
    next: Box<dyn Fn(T)>,
    complete: Box<dyn Fn()>,
    error: Box<dyn Fn(E)>,
}

struct FullObserver<T, E> {
    start: Box<dyn Fn(&mut Subscription)>,
    next: Box<dyn Fn(T)>,
    complete: Box<dyn Fn()>,
    error: Box<dyn Fn(E)>,
}

struct Observable<T, E> {
    subscription_fn: Rc<dyn Fn(&FullObserver<T, E>) -> Box<dyn Fn()>>,
}

struct Subscription {
    cleanup: Option<Box<dyn Fn()>>,
    closed: bool,
}

impl Subscription {
    fn new<T, E, F: Fn() + 'static>(
        observer: FullObserver<T, E>,
        subscription_fn: Rc<dyn Fn(&FullObserver<T, E>) -> F>,
    ) -> Subscription {
        let mut subscription = Subscription {
            cleanup: None,
            closed: false,
        };

        (observer.start)(&mut subscription);
        if !subscription.closed {
            let cleanup = Box::new((subscription_fn)(&observer));
            subscription.cleanup = Some(cleanup);
        }

        return subscription;
    }

    fn unsubscribe(&mut self) {
        if let Some(cleanup) = &self.cleanup {
            (cleanup)();
        }
        self.closed = true;
    }
}

impl<T, E> Observable<T, E> {
    fn new<F: Fn() + 'static>(
        subscription_fn: impl (Fn(&FullObserver<T, E>) -> F) + 'static,
    ) -> Self {
        Observable {
            subscription_fn: Rc::new(move |observer| Box::new(subscription_fn(observer))),
        }
    }

    fn subscribe(&self, next: impl Fn(T) + 'static) -> Subscription {
        Subscription::new(
            FullObserver {
                start: Box::new(|_s| {}),
                next: Box::new(next),
                complete: Box::new(|| {}),
                error: Box::new(|_err| {}),
            },
            self.subscription_fn.clone(),
        )
    }

    fn subscribe_observer(&self, observer: Observer<T, E>) -> Subscription {
        Subscription::new(
            FullObserver {
                start: Box::new(|_s| {}),
                next: observer.next,
                complete: observer.complete,
                error: observer.error,
            },
            self.subscription_fn.clone(),
        )
    }

    fn subscribe_full_observer(&self, observer: FullObserver<T, E>) -> Subscription {
        Subscription::new(observer, self.subscription_fn.clone())
    }
}

fn main() {
    let observable = Observable::<i32, Box<dyn Error>>::new(|observer| {
        (observer.next)(42);
        (observer.next)(666);
        (observer.complete)();
        return || {
            println!("cleanup");
        };
    });

    let mut subscription = observable.subscribe(|value| {
        println!("next {}", value);
    });
    subscription.unsubscribe();

    let some_closure = "yo!";
    observable.subscribe_full_observer(FullObserver {
        start: Box::new(|_subscription| println!("start")),
        next: Box::new(|value| println!("next {}", value)),
        complete: Box::new(move || println!("complete {}", some_closure)),
        error: Box::new(|error| eprintln!("error {:?}", error)),
    });
}
```

Answer 1

Firstly, this doesn't appear to follow the Observer pattern. Allow me to quote Wikipedia:

The observer pattern is a software design pattern in which an object, named the subject, maintains a list of its dependents, called observers, and notifies them automatically of any state changes, usually by calling one of their methods.

Your code lacks a list of dependents or observers which are notified in tandem whenever there is a state change. Instead, every call to subscribe invokes the "subscription function" which produces an independent stream of events.

Secondly, patterns involving invoking callbacks are a poor fit for Rust. They don't interact well with lifetimes. To see this, consider the basic callback function that you have:

let mut subscription = observable.subscribe(|value| {
    println!("next {}", value);
});

Here, you print something. But you can't actually do anything useful. The callback type is Fn(T) + 'static So this closure cannot contain any references to anything and cannot mutate any state. The only way to interact with the rest of the system would be to give the closure a Rc<RefCell<_>> (or equivalent) that it could use to get a mutable reference to the rest of the system.

We can also see it in your subscription function:

let observable = Observable::<i32, Box<dyn Error>>::new(|observer| {
    (observer.next)(42);
    (observer.next)(666);
    (observer.complete)();
    return || {
        println!("cleanup");
    };
});

Here, you generate a bunch of values and finish immediately. And that's pretty much the only thing you can do. Because the observer is borrowed, you can't keep it around to call later in response to any sort of event.

What should you do instead? That depends on the situation. But you should pretty much always be able to better a rust friendly pattern which is a better fit.

In general, I find that using Rc, Box, and dyn are signs that are trying to write code in a OO style which doesn't fit Rust well. Of course, you can do that if you really want, but Rust will give you a lot of friction. Furthermore, you won't notice this friction just printing out some test outputs. Try actually solving real problems and you'll see what I mean.

Answer 2

@Winston Ewert already points out several problems with your code, I however, would like to give you a naive implementation that works with asynchronous rust:

use std::ops::Deref;
use std::pin::Pin;
use std::future::Future;
use std::task::{Waker, Context, Poll};
use std::sync::{Arc, RwLock, RwLockReadGuard};
use std::marker::PhantomData;
use std::num::Wrapping;

/// Thread-safe observable, intended to be implemented on `&self`
trait Observable<'a> {
    type Item;
    /// (item, epoch)
    type Ref: Deref<Target = (Self::Item, usize)> + 'a;

    /// subscriber has to be renewed after the value is updated
    fn subscribe(self, waker: Waker);
    fn update(self, item: Self::Item);
    fn get_observed(self) -> Self::Ref;
    fn get_epoch(self) -> usize;
}

struct ObservableFuture<'a, T: Observable<'a>> {
    observable: T,
    epoch: usize,
    phantom: PhantomData<&'a ()>,
}
impl<'a, T: Copy + Observable<'a>> ObservableFuture<'a, T> {
    fn new(observable: T) -> Self {
        Self {
            observable,
            epoch: observable.get_epoch(),
            phantom: PhantomData,
        }
    }
}
impl<'a, T: Copy + Observable<'a>> Future for ObservableFuture<'a, T> {
    type Output = T::Ref;
    
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let observable = &self.observable;
        
        if self.epoch != observable.get_epoch() {
            Poll::Ready(observable.get_observed())
        } else {
            observable.subscribe(cx.waker().clone());
            Poll::Pending
        }
    }
}
fn wait_for_update<'a, T: Copy + Observable<'a>>(observable: T) -> ObservableFuture<'a, T> {
    ObservableFuture::new(observable)
}

struct Observed<T> {
    // If T is a primitive, then you can replace RwLock
    // with Atomic.
    // Or if T is `Sync`, then you may remove `RwLock`.
    // However, you still have to think of a way where
    // epoch can be updated atomically with T.
    item: RwLock<(T, usize)>,
    // You can consider replacing Vec with other
    // container with built-in concurrency support.
    // such as crossbeam::queue::SegQueue
    observers: RwLock<Vec<Waker>>,
}
impl<T> Observed<T> {
    fn new(item: T) -> Self {
        Self {
            item: RwLock::new((item, 0)),
            observers: RwLock::new(Vec::new()),
        }
    }
}
impl<'a, T: 'a> Observable<'a> for &'a Observed<T> {
    type Item = T;
    type Ref = RwLockReadGuard<'a, (T, usize)>;

   fn subscribe(self, waker: Waker) {
        self.observers.write().unwrap().push(waker);
    }

    fn update(self, item: Self::Item) {
        let epoch = Wrapping(self.get_epoch());
        let one = Wrapping(1);
        *self.item.write().unwrap() = (item, (epoch + one).0);
        let observers = std::mem::replace(&mut *self.observers.write().unwrap(), Vec::new());
        for observer in observers {
            observer.wake_by_ref();
        }
    }

    fn get_observed(self) -> Self::Ref {
        self.item.read().unwrap()
    }
    
    fn get_epoch(self) -> usize {
        self.item.read().unwrap().1
    }
}

#[tokio::main]
async fn main() {
    use tokio::time::{sleep, Duration};
    
    let observed = Arc::new(Observed::new(Vec::<i32>::new()));
    
    let observed_cloned = observed.clone();
    let handle = tokio::spawn(async move {
        let observed = observed_cloned;
        
        println!("{:#?}", *wait_for_update(& *observed).await);
    });

    let observed_cloned = observed.clone();
    let handle2 = tokio::spawn(async move {
        let observed = observed_cloned;
        
        println!("{:#?}", *wait_for_update(& *observed).await);
    });

    // sleep for 1s to make sure wait_for_update is executed
    sleep(Duration::from_millis(1000)).await;
    observed.update(vec![1, 2, 3, 4]);
    
    handle.await.unwrap();
    handle2.await.unwrap();
}

The Observable trait is designed to be thread-safe and work with async, requiring to be implemented on copyable reference (immutable reference).

The Observed struct provides a naive sample implementation which stores the list of observers in a Rwlock<Vec>, while a better implementation could use crossbeam::queue::SegQueue.

It also uses RwLock for safely updating the data, but for primitive you can use Atomic and for T that is already thread safe, you can remove RwLock completely, but you still need to account for updating ecpoch atomically with the data.

I have tested my code on rust playground and it worked as expected for two observers, though I haven’t tested it for more observers.

Edit:

If you are running in single-thread only, then you can simply remove the RwLock and there will be no synchronization overhead.