如何在Rust中实现观察者模式？

Question

我有一个可观察的集合和一个观察者.我希望观察者成为一个特质实现trait Observer.当某些事件发生时,可观察对象应该能够通知每个观察者.这应该解释我的意图:

struct A {
    observables: Vec<Observable>,
}

impl A {
    fn new() -> A {
        A {
            observables: vec![],
        }
    }
}

trait Observer {
    fn event(&mut self, _: &String);
}

impl Observer for A {
    fn event(&mut self, ev: &String) {
        println!("Got event from observable: {}", ev);
    }
}

struct Observable {
    observers: Vec<dyn Observer>, // How to contain references to observers? (this line is invalid)
}

impl Observable {
    fn new() -> Observable {
        Observable {
            observers: Vec::new(),
        }
    }

    fn add_observer(&mut self, o: &dyn Observer) {
        // incorrect line too
        self.observers.push(o);
    }

    fn remove_observer(&mut self, o: &dyn Observer) {
        // incorrect line too
        self.observers.remove(o);
    }

    fn notify_observers(&self, ev: &String) {
        for o in &mut self.observers {
            o.event(ev);
        }
    }
}

(游乐场)

我收到错误:

error[E0277]: the size for values of type `(dyn Observer + 'static)` cannot be known at compilation time
  --> src/lib.rs:24:5
   |
24 |     observers: Vec<dyn Observer>, // How to contain references to observers?
   |     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^ doesn't have a size known at compile-time
   |
   = help: the trait `std::marker::Sized` is not implemented for `(dyn Observer + 'static)`
   = note: to learn more, visit <https://doc.rust-lang.org/book/second-edition/ch19-04-advanced-types.html#dynamically-sized-types-and-the-sized-trait>
   = note: required by `std::vec::Vec`

这只是我想要做的模型.我在Java,Python和C++中有这样的代码,但我不知道如何在Rust中实现观察者模式.我相信我的问题是存储对可观察对象内的观察者对象的引用.

Answer 1

Observable根据实施选择,该模式可能构成所有权挑战.

在Garbage Collected语言中,通常Observer引用Observer(通知它)和Observable引用Observable(注销本身)......这会导致一些所有权方面的挑战(谁比谁更长？)并且存在整体"关于取消注册的通知"的事情.

在Rust(和C++)中,我建议避免循环.

简单解决方案

在Observer与Observer具有不同的寿命,无持有其他或被预期寿命比其他.

use std::rc::Weak;

struct Event;

trait Observable {
    fn register(&mut self, observer: Weak<dyn Observer>);
}

trait Observer {
    fn notify(&self, event: &Event);
}

关键是分配Rc到a Weak然后移交Observable(弱引用)到Observer.

如果Event需要修改需求RefCell,则要么需要内部可变性,要么需要将其包装成Weak<RefCell<dyn Observer>>(传递Observable给Observable).

当通知时,Observer意志会经常意识到存在死亡的弱引用(Observable已经消失),它可以随意删除那些,懒惰.

还有其他解决方案,例如使用Broker(非常类似于事件循环),从推模式转换到拉模式(即(1)生成所有事件,(2)处理所有事件),但这些有点偏离传统的观察者模式,有不同的优点/缺点,所以我不会试图在这里对待它们.

Answer 2

我使用了回调函数.它简单而强大,没有终身问题或类型擦除.

我试过了Weak<dyn Observer>,但是

1. 它需要一个 Rc
2. 您必须重复代码才能创建不同的观察者结构.
3. 它需要类型擦除

pub struct Notifier<E> {
    subscribers: Vec<Box<dyn Fn(&E)>>,
}

impl<E> Notifier<E> {
    pub fn new() -> Notifier<E> {
        Notifier {
            subscribers: Vec::new(),
        }
    }

    pub fn register<F>(&mut self, callback: F)
    where
        F: 'static + Fn(&E),
    {
        self.subscribers.push(Box::new(callback));
    }

    pub fn notify(&self, event: E) {
        for callback in &self.subscribers {
            callback(&event);
        }
    }
}

Answer 3

这是我基于这个问题的答案和许多痛苦和痛苦的实施.我使用弱引用来存储观察者,并使用a RefCell来调用mutable notify().

我正在使用,Arc因为我的侦听器可以从任何线程调用.如果您使用单个线程,则可以使用Rc.

每次dispatch()调用时,它都会检查是否有任何弱引用的侦听器已经消失.如果有,它将清理监听器列表.

pub enum Event {} // You make Event hold anything you want to fire 

pub trait Listener {
    fn notify(&mut self, event: &Event);
}

pub trait Dispatchable<T>
    where T: Listener
{
    fn register_listener(&mut self, listener: Arc<RefCell<T>>);
}

pub struct Dispatcher<T>
    where T: Listener
{
    /// A list of synchronous weak refs to listeners
    listeners: Vec<Weak<RefCell<T>>>,
}

impl<T> Dispatchable<T> for Dispatcher<T>
    where T: Listener
{
    /// Registers a new listener
    fn register_listener(&mut self, listener: Arc<RefCell<T>>) {
        self.listeners.push(Arc::downgrade(&listener));
    }
}

impl<T> Dispatcher<T>
    where T: Listener
{
    pub fn new() -> Dispatcher<T> {
        Dispatcher { listeners: Vec::new() }
    }

    pub fn num_listeners(&self) -> usize {
        self.listeners.len()
    }

    pub fn dispatch(&mut self, event: Event) {
        let mut cleanup = false;
        // Call the listeners
        for l in self.listeners.iter() {
            if let Some(mut listener_rc) = l.upgrade() {
                let mut listener = listener_rc.borrow_mut();
                listener.notify(&event);
            } else {
                println!("Cannot get listener, cleanup necessary");
                cleanup = true;
            }
        }
        // If there were invalid weak refs, clean up the list
        if cleanup {
            println!("Dispatcher is cleaning up weak refs");
            self.listeners.retain(|ref l| {
                // Only retain valid weak refs
                let got_ref = l.clone().upgrade();
                match got_ref {
                    None => false,
                    _ => true,
                }
            });
        }
    }
}

这是一个练习它的单元测试代码片段.

测试来自卡片游戏库,我的Event枚举FlopDealt和GameFinished变体.测试创建并注册我的监听器,并确保在FlopDealt调度时调用它.范围部分是这样我可以在侦听器超出范围后测试弱引用行为.我触发另一个事件并计算侦听器的数量以确保清除列表.

use std::time::Instant;

#[derive(Debug)]
pub enum Event {
    FlopDealt,
    GameFinished { ended: Instant },
}

struct MyListener {
    pub flop_dealt: bool,
}

impl Listener for MyListener {
    fn notify(&mut self, event: &Event) {
        println!("Notify called with {:?}", event);
        if let Event::FlopDealt = event {
            println!("Flop dealt");
            self.flop_dealt = true;
        }
    }
}

#[test]
fn events_register() {
    let mut d: Dispatcher<MyListener> = Dispatcher::new();

    {
        let listener_rc = Arc::new(RefCell::new(MyListener { flop_dealt: false }));
        d.register_listener(listener_rc.clone());
        d.dispatch(Event::FlopDealt);

        let flop_dealt = listener_rc.borrow().flop_dealt;
        println!("Flop was {}dealt", if flop_dealt { "" } else { "not " });
        assert_eq!(flop_dealt, true);
        assert_eq!(d.num_listeners(), 1);
    }

    // Listener should disappear
    d.dispatch(Event::GameFinished {
        ended: Instant::now(),
    });
    assert_eq!(d.num_listeners(), 0);
}