Event Bus и паттерн Observer в Rust: руководство

Event Bus и паттерн Observer в Rust: реализация и примеры

Event Bus (шина событий) — это архитектурный паттерн, реализующий принцип publish-subscribe (издатель-подписчик). В Rust этот паттерн особенно полезен для построения слабосвязанных систем, где компоненты общаются через события, не зная друг о друге напрямую. В этой статье мы разберём, как создать собственный Event Bus с использованием паттерна Observer, и рассмотрим готовые решения из экосистемы Rust.

Паттерн Observer: основы

Паттерн Observer определяет отношение «один-ко-многим» между объектами. Когда один объект (издатель) изменяет своё состояние, все зависимые объекты (подписчики) автоматически уведомляются. В Rust это часто реализуется через трейты и динамическую диспетчеризацию.

Базовая реализация Observer в Rust

use std::collections::HashMap;

// Трейт для наблюдателя
trait Observer {
    fn update(&self, event: &str);
}

// Структура издателя
struct EventManager {
    observers: Vec<Box<dyn Observer>>,
}

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

    fn subscribe(&mut self, observer: Box<dyn Observer>) {
        self.observers.push(observer);
    }

    fn notify(&self, event: &str) {
        for observer in &self.observers {
            observer.update(event);
        }
    }
}

// Конкретный наблюдатель
struct Logger;

impl Observer for Logger {
    fn update(&self, event: &str) {
        println!("[Logger] Получено событие: {}", event);
    }
}

fn main() {
    let mut manager = EventManager::new();
    manager.subscribe(Box::new(Logger));
    manager.notify("user.login");
}

Event Bus: расширенная версия Observer

Event Bus — это централизованная шина, которая маршрутизирует события по типам. В отличие от простого Observer, здесь поддерживаются разные типы событий и фильтрация подписок.

Реализация Event Bus с типами событий

use std::any::Any;
use std::collections::HashMap;
use std::sync::{Arc, Mutex};

// Тип события
#[derive(Clone, Debug)]
enum EventType {
    UserLogin,
    UserLogout,
    DataUpdate,
}

// Структура события
#[derive(Clone, Debug)]
struct Event {
    event_type: EventType,
    payload: String,
}

// Обработчик события
type Handler = Arc<dyn Fn(Event) + Send + Sync>;

// Шина событий
struct EventBus {
    handlers: HashMap<EventType, Vec<Handler>>,
}

impl EventBus {
    fn new() -> Self {
        EventBus {
            handlers: HashMap::new(),
        }
    }

    fn subscribe<F>(&mut self, event_type: EventType, handler: F)
    where
        F: Fn(Event) + 'static + Send + Sync,
    {
        self.handlers
            .entry(event_type)
            .or_insert_with(Vec::new)
            .push(Arc::new(handler));
    }

    fn publish(&self, event: Event) {
        if let Some(handlers) = self.handlers.get(&event.event_type) {
            for handler in handlers {
                handler(event.clone());
            }
        }
    }
}

fn main() {
    let mut bus = EventBus::new();

    bus.subscribe(EventType::UserLogin, |event| {
        println!("Пользователь вошёл: {}", event.payload);
    });

    bus.subscribe(EventType::UserLogin, |event| {
        // Другой обработчик
        println!("Логин зафиксирован: {}", event.payload);
    });

    bus.publish(Event {
        event_type: EventType::UserLogin,
        payload: "user123".to_string(),
    });
}

Асинхронный Event Bus с Tokio

В реальных приложениях часто требуется асинхронная обработка событий. Для этого отлично подходит библиотека tokio с её каналами broadcast.

Использование tokio::sync::broadcast

use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    // Создаём канал broadcast с буфером на 16 сообщений
    let (tx, mut rx1) = broadcast::channel::<String>(16);
    let mut rx2 = tx.subscribe();

    // Первый подписчик
    tokio::spawn(async move {
        while let Ok(msg) = rx1.recv().await {
            println!("Подписчик 1 получил: {}", msg);
        }
    });

    // Второй подписчик
    tokio::spawn(async move {
        while let Ok(msg) = rx2.recv().await {
            println!("Подписчик 2 получил: {}", msg);
        }
    });

    // Публикуем события
    tx.send("Событие 1".to_string()).unwrap();
    tx.send("Событие 2".to_string()).unwrap();

    // Даём время на обработку
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
}