agentlink_sdk/events/

event_loop.rs

//! Generic Event Loop
//!
//! Provides type-safe event handling with generic callbacks.

use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use tokio::sync::{mpsc, RwLock};
use serde::de::DeserializeOwned;
use serde_json::Value;

use super::ServerEvent;



/// Type alias for event callbacks
pub type EventCallback<T> = Arc<
    dyn Fn(ServerEvent<T>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync,
>;

/// Type-erased callback wrapper
struct TypeErasedCallback {
    callback: Arc<dyn Fn(Value) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>,
}

/// Generic event loop with type-safe callbacks
pub struct EventLoop {
    /// Registered callbacks for specific event types
    callbacks: Arc<RwLock<HashMap<String, Vec<TypeErasedCallback>>>>,
    /// Channel sender for internal event distribution
    event_tx: mpsc::UnboundedSender<(String, Value)>,
    /// Running state
    running: Arc<RwLock<bool>>,
}

impl EventLoop {
    /// Create a new event loop
    pub fn new() -> (Self, mpsc::UnboundedReceiver<(String, Value)>) {
        let (tx, rx) = mpsc::unbounded_channel();
        
        let event_loop = Self {
            callbacks: Arc::new(RwLock::new(HashMap::new())),
            event_tx: tx,
            running: Arc::new(RwLock::new(false)),
        };
        
        (event_loop, rx)
    }
    
    /// Register a typed callback for a specific event type
    ///
    /// # Example
    /// ```rust,ignore
    /// event_loop.on_event(EVENT_MESSAGE_RECEIVED, |event: ServerEvent<MessageReceivedData>| {
    ///     Box::pin(async move {
    ///         println!("New message: {:?}", event.data.content);
    ///     })
    /// }).await;
    /// ```
    pub async fn on_event<T, F, Fut>(&self, event_type: &str, callback: F)
    where
        T: DeserializeOwned + Send + Sync + 'static,
        F: Fn(ServerEvent<T>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = ()> + Send + 'static,
    {
        let wrapped = TypeErasedCallback {
            callback: Arc::new(move |value: Value| {
                // Try to deserialize the full event
                match serde_json::from_value::<ServerEvent<T>>(value.clone()) {
                    Ok(event) => {
                        Box::pin(callback(event)) as Pin<Box<dyn Future<Output = ()> + Send>>
                    }
                    Err(e) => {
                        tracing::error!("[EventLoop] Failed to deserialize event: {}", e);
                        Box::pin(async {}) as Pin<Box<dyn Future<Output = ()> + Send>>
                    }
                }
            }),
        };
        
        let mut callbacks = self.callbacks.write().await;
        callbacks
            .entry(event_type.to_string())
            .or_insert_with(Vec::new)
            .push(wrapped);
        
        tracing::debug!("[EventLoop] Registered callback for event type: {}", event_type);
    }
    
    /// Register a handler function directly (simplified API)
    ///
    /// This is a convenience wrapper around `on_event` that allows direct
    /// registration of async handler functions without wrapping them in closures.
    ///
    /// # Example
    /// ```rust,ignore
    /// // 定义 handler 函数
    /// async fn on_message_received(event: ServerEvent<MessageReceivedData>) {
    ///     println!("New message: {:?}", event.data.content);
    /// }
    ///
    /// // 直接注册，无需包装闭包
    /// event_loop.on(EVENT_MESSAGE_RECEIVED, on_message_received).await;
    /// ```
    pub async fn on<T, F, Fut>(&self, event_type: &'static str, handler: F)
    where
        T: DeserializeOwned + Send + Sync + 'static,
        F: Fn(ServerEvent<T>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = ()> + Send + 'static,
    {
        self.on_event(event_type, handler).await;
    }
    
    /// Remove all callbacks for a specific event type
    pub async fn off_event(&self, event_type: &str) {
        let mut callbacks = self.callbacks.write().await;
        callbacks.remove(event_type);
        tracing::info!("[EventLoop] Removed callbacks for event type: {}", event_type);
    }
    
    /// Clear all callbacks
    pub async fn clear_callbacks(&self) {
        let mut callbacks = self.callbacks.write().await;
        callbacks.clear();
        tracing::info!("[EventLoop] Cleared all callbacks");
    }
    
    /// Process a single event
    pub async fn process_event(&self, event_type: &str, payload: Value) {
        let callbacks = self.callbacks.read().await;
        
        if let Some(handlers) = callbacks.get(event_type) {
            for handler in handlers {
                let callback = handler.callback.clone();
                let payload = payload.clone();
                tokio::spawn(async move {
                    callback(payload).await;
                });
            }
        }
    }
    
    /// Start the event loop
    pub async fn start(&self, mut event_rx: mpsc::UnboundedReceiver<(String, Value)>) {
        {
            let mut running = self.running.write().await;
            *running = true;
        }
        
        tracing::debug!("[EventLoop] Started");
        
        while let Some((event_type, payload)) = event_rx.recv().await {
            self.process_event(&event_type, payload).await;
        }
        
        {
            let mut running = self.running.write().await;
            *running = false;
        }
        
        tracing::info!("[EventLoop] Stopped");
    }
    
    /// Check if the event loop is running
    pub async fn is_running(&self) -> bool {
        *self.running.read().await
    }
    
    /// Get event sender
    pub fn event_sender(&self) -> mpsc::UnboundedSender<(String, Value)> {
        self.event_tx.clone()
    }
}

impl Clone for EventLoop {
    fn clone(&self) -> Self {
        Self {
            callbacks: self.callbacks.clone(),
            event_tx: self.event_tx.clone(),
            running: self.running.clone(),
        }
    }
}

impl Default for EventLoop {
    fn default() -> Self {
        let (event_loop, _) = Self::new();
        event_loop
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use super::super::{MessageReceivedData, EVENT_MESSAGE_RECEIVED};
    
    #[tokio::test]
    async fn test_typed_callback_registration() {
        let (event_loop, _) = EventLoop::new();
        
        let called = Arc::new(RwLock::new(false));
        let called_clone = called.clone();
        
        event_loop.on_event(EVENT_MESSAGE_RECEIVED, move |event: ServerEvent<MessageReceivedData>| {
            let called = called_clone.clone();
            Box::pin(async move {
                assert_eq!(event.data.content, Some("Hello".to_string()));
                let mut c = called.write().await;
                *c = true;
            })
        }).await;
        
        // Verify callback was registered
        let callbacks = event_loop.callbacks.read().await;
        assert!(callbacks.contains_key(EVENT_MESSAGE_RECEIVED));
    }
}