flare_core/common/message/
processor.rs

1//! 消息处理器实现
2//!
3//! 提供常用的消息处理器实现
4
5use super::pipeline::{MessageContext, MessageProcessor};
6use crate::common::error::Result;
7use crate::common::protocol::Frame;
8use async_trait::async_trait;
9use std::sync::Arc;
10
11/// 函数式消息处理器
12///
13/// 使用闭包处理消息，适合简单场景
14pub struct FunctionProcessor {
15    name: String,
16    #[allow(clippy::type_complexity)]
17    handler: Arc<
18        dyn Fn(
19                &MessageContext,
20            )
21                -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Option<Frame>>> + Send>>
22            + Send
23            + Sync,
24    >,
25}
26
27impl FunctionProcessor {
28    /// 创建新的函数式处理器
29    pub fn new<F, Fut>(name: impl Into<String>, handler: F) -> Self
30    where
31        F: Fn(&MessageContext) -> Fut + Send + Sync + 'static,
32        Fut: std::future::Future<Output = Result<Option<Frame>>> + Send + 'static,
33    {
34        Self {
35            name: name.into(),
36            handler: Arc::new(move |ctx| Box::pin(handler(ctx))),
37        }
38    }
39}
40
41#[async_trait]
42impl MessageProcessor for FunctionProcessor {
43    async fn process(&self, ctx: &MessageContext) -> Result<Option<Frame>> {
44        (self.handler)(ctx).await
45    }
46
47    fn name(&self) -> &str {
48        &self.name
49    }
50}
51
52/// 委托处理器
53///
54/// 将消息委托给其他处理器（如 ConnectionHandler）
55pub struct DelegateProcessor {
56    name: String,
57    #[allow(clippy::type_complexity)]
58    handler: Arc<
59        dyn Fn(
60                &Frame,
61                Option<&str>,
62            )
63                -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<Option<Frame>>> + Send>>
64            + Send
65            + Sync,
66    >,
67}
68
69impl DelegateProcessor {
70    /// 创建新的委托处理器
71    pub fn new<F, Fut>(name: impl Into<String>, handler: F) -> Self
72    where
73        F: Fn(&Frame, Option<&str>) -> Fut + Send + Sync + 'static,
74        Fut: std::future::Future<Output = Result<Option<Frame>>> + Send + 'static,
75    {
76        Self {
77            name: name.into(),
78            handler: Arc::new(move |frame, conn_id| Box::pin(handler(frame, conn_id))),
79        }
80    }
81}
82
83#[async_trait]
84impl MessageProcessor for DelegateProcessor {
85    async fn process(&self, ctx: &MessageContext) -> Result<Option<Frame>> {
86        (self.handler)(&ctx.frame, ctx.connection_id.as_deref()).await
87    }
88
89    fn name(&self) -> &str {
90        &self.name
91    }
92}
flare_core/common/message/processor.rs

flare_core/common/message/
processor.rs
