1
\$\begingroup\$

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)),
    });
}
```
\$\endgroup\$
2
\$\begingroup\$

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.

\$\endgroup\$
4
  • \$\begingroup\$ Thanks a lot for your review! You say the callback can't do anything useful but (1) it's a lambda that can still access its environment through closures right? and (2) the value provided can be mutable. You say the observer is borrowed, but I though that it would depend on T (wether a reference, ref mut or a value). Finally, my understanding of Box and Rc are that they precisely allow you to do thing like that, it's even describe in the rust book. Thanks again! \$\endgroup\$ Jul 22 '21 at 5:22
  • \$\begingroup\$ And you are right, this does not fully implement the Observer pattern. My intention is to implement a Subject based on the Observable. The subscription function is used to append the observer to a list and the cleanup function to remove it from the list. So this is just a building block of the Observer pattern. \$\endgroup\$ Jul 22 '21 at 7:05
  • 1
    \$\begingroup\$ @LukeSkywalker, (1) no, it can't access the environment directly because that would involve holding a reference to the environment, which is disallowed because the type is 'static. It will only be able to access things either moved into the closure or things through a Rc or similiar. 2) True. 3) Yes, Box and Rc let you do such things. But needing to do such things means you are running afoul of the borrow checker and thus probably not writing very Rusty code. \$\endgroup\$ Jul 24 '21 at 22:12
  • 1
    \$\begingroup\$ The observer is borrowed because the type is &FullObserver<T, E>, the & makes it borrowed regardless of the type of T. I believe you'll find it impossible to implement a full observer pattern on this building block, because the lifetimes you have going won't allow it. Probably, the best way to really understand that is to try anyways. \$\endgroup\$ Jul 24 '21 at 22:23
0
\$\begingroup\$

@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.

\$\endgroup\$
3
  • \$\begingroup\$ Thanks for your review! It's an interesting implementation. However my use case is not multi-threaded but associated to an event loop and the additional complexity and, probably, execution cost are not worth it. \$\endgroup\$ Jul 22 '21 at 7:03
  • \$\begingroup\$ @LukeSkywalker If you are using an event loop, then you can definitly use my implementation, as it supports async wakeup upon update to the value. \$\endgroup\$
    – JiaHao Xu
    Jul 22 '21 at 10:14
  • \$\begingroup\$ @LukeSkywalker The RwLock I used in Observed can be easily removed if it is used inside a single-thread program. \$\endgroup\$
    – JiaHao Xu
    Jul 22 '21 at 10:15

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.