Dispatcher POC impl

Cargo.toml

@@ -13,4 +13,6 @@ lazy_static = "1.3"
 apply = "0.2.2"
 derive_more = "0.15.0"
 tokio = "0.2.0-alpha.4"
-bytes = "0.4.12"
+bytes = "0.4.12"
+log = "0.4.8"
+futures-util-preview = "0.3.0-alpha.18"

src/dispatcher/filter.rs

@@ -0,0 +1,243 @@
+/// Filter that determines that particular event
+/// is suitable for particular handler.
+pub trait Filter<T> {
+    /// Passes (return true) if event is suitable (otherwise return false)
+    fn test(&self, value: &T) -> bool;
+}
+
+/// ```
+/// use async_telegram_bot::dispatcher::filter::Filter;
+///
+/// let closure = |i: &i32| -> bool { *i >= 42 };
+/// assert!(closure.test(&42));
+/// assert!(closure.test(&100));
+///
+/// assert_eq!(closure.test(&41), false);
+/// assert_eq!(closure.test(&0), false);
+/// ```
+impl<T, F: Fn(&T) -> bool> Filter<T> for F {
+    fn test(&self, value: &T) -> bool {
+        (self)(value)
+    }
+}
+
+/// ```
+/// use async_telegram_bot::dispatcher::filter::Filter;
+///
+/// assert!(true.test(&()));
+/// assert_eq!(false.test(&()), false);
+/// ```
+impl<T> Filter<T> for bool {
+    fn test(&self, _: &T) -> bool { *self }
+}
+
+/// And filter.
+///
+/// Passes if both underlying filters pass.
+///
+/// **NOTE**: if one of filters don't pass
+/// it is **not** guaranteed that other will be executed.
+///
+/// ## Examples
+/// ```
+/// use async_telegram_bot::dispatcher::filter::{And, Filter};
+///
+/// // Note: bool can be treated as `Filter` that always return self.
+/// assert_eq!(And::new(true, false).test(&()), false);
+/// assert_eq!(And::new(true, false).test(&()), false);
+/// assert!(And::new(true, true).test(&()));
+/// assert!(And::new(true, And::new(|_: &()| true, true)).test(&()));
+/// ```
+#[derive(Debug, Clone, Copy)]
+pub struct And<A, B>(A, B);
+
+impl<A, B> And<A, B> {
+    pub fn new(a: A, b: B) -> Self {
+        And(a, b)
+    }
+}
+
+impl<T, A, B> Filter<T> for And<A, B>
+where
+    A: Filter<T>,
+    B: Filter<T>,
+{
+    fn test(&self, value: &T) -> bool {
+        self.0.test(value) && self.1.test(value)
+    }
+}
+
+/// Alias for [`And::new`]
+///
+/// ## Examples
+/// ```
+/// use async_telegram_bot::dispatcher::filter::{and, Filter};
+///
+/// assert!(and(true, true).test(&()));
+/// assert_eq!(and(true, false).test(&()), false);
+/// ```
+///
+/// [`And::new`]: crate::dispatcher::filter::And::new
+pub fn and<A, B>(a: A, b: B) -> And<A, B> {
+    And::new(a, b)
+}
+
+
+/// Or filter.
+///
+/// Passes if at least one underlying filters passes.
+///
+/// **NOTE**: if one of filters passes
+/// it is **not** guaranteed that other will be executed.
+///
+/// ## Examples
+/// ```
+/// use async_telegram_bot::dispatcher::filter::{Or, Filter};
+///
+/// // Note: bool can be treated as `Filter` that always return self.
+/// assert!(Or::new(true, false).test(&()));
+/// assert!(Or::new(false, true).test(&()));
+/// assert!(Or::new(false, Or::new(|_: &()| true, false)).test(&()));
+/// assert_eq!(Or::new(false, false).test(&()), false);
+/// ```
+#[derive(Debug, Clone, Copy)]
+pub struct Or<A, B>(A, B);
+
+impl<A, B> Or<A, B> {
+    pub fn new(a: A, b: B) -> Self {
+        Or(a, b)
+    }
+}
+
+impl<T, A, B> Filter<T> for Or<A, B>
+where
+    A: Filter<T>,
+    B: Filter<T>,
+{
+    fn test(&self, value: &T) -> bool {
+        self.0.test(value) || self.1.test(value)
+    }
+}
+
+/// Alias for [`Or::new`]
+///
+/// ## Examples
+/// ```
+/// use async_telegram_bot::dispatcher::filter::{or, Filter};
+///
+/// assert!(or(true, false).test(&()));
+/// assert_eq!(or(false, false).test(&()), false);
+/// ```
+///
+/// [`Or::new`]: crate::dispatcher::filter::Or::new
+pub fn or<A, B>(a: A, b: B) -> Or<A, B> {
+    Or::new(a, b)
+}
+
+
+/// Not filter.
+///
+/// Passes if underlying filter don't pass.
+///
+/// ## Examples
+/// ```
+/// use async_telegram_bot::dispatcher::filter::{Not, Filter};
+///
+/// // Note: bool can be treated as `Filter` that always return self.
+/// assert!(Not::new(false).test(&()));
+/// assert_eq!(Not::new(true).test(&()), false);
+/// ```
+#[derive(Debug, Clone, Copy)]
+pub struct Not<A>(A);
+
+impl<A> Not<A> {
+    pub fn new(a: A) -> Self {
+        Not(a)
+    }
+}
+
+impl<T, A> Filter<T> for Not<A>
+where
+    A: Filter<T>,
+{
+    fn test(&self, value: &T) -> bool {
+        !self.0.test(value)
+    }
+}
+
+/// Alias for [`Not::new`]
+///
+/// ## Examples
+/// ```
+/// use async_telegram_bot::dispatcher::filter::{not, Filter};
+///
+/// assert!(not(false).test(&()));
+/// assert_eq!(not(true).test(&()), false);
+/// ```
+///
+/// [`Not::new`]: crate::dispatcher::filter::Not::new
+pub fn not<A>(a: A) -> Not<A> {
+    Not::new(a)
+}
+
+/// Return [filter] that passes if and only if all given filters passes.
+///
+/// **NOTE**: if one of filters don't pass
+/// it is **not** guaranteed that other will be executed.
+///
+/// ## Examples
+/// ```
+/// use async_telegram_bot::{all, dispatcher::filter::Filter};
+///
+/// assert!(all![true].test(&()));
+/// assert!(all![true, true].test(&()));
+/// assert!(all![true, true, true].test(&()));
+///
+/// assert_eq!(all![false].test(&()), false);
+/// assert_eq!(all![true, false].test(&()), false);
+/// assert_eq!(all![false, true].test(&()), false);
+/// assert_eq!(all![false, false].test(&()), false);
+/// ```
+///
+/// [filter]: crate::dispatcher::filter::Filter
+#[macro_export]
+macro_rules! all {
+    ($one:expr) => { $one };
+    ($head:expr, $($tail:tt)+) => {
+        $crate::dispatcher::filter::And::new(
+            $head,
+            $crate::all!($($tail)+)
+        )
+    };
+}
+
+/// Return [filter] that passes if and only if any given filters passes.
+///
+/// **NOTE**: if one of filters passes
+/// it is **not** guaranteed that other will be executed.
+///
+/// ## Examples
+/// ```
+/// use async_telegram_bot::{any, dispatcher::filter::Filter};
+///
+/// assert!(any![true].test(&()));
+/// assert!(any![true, true].test(&()));
+/// assert!(any![false, true].test(&()));
+/// assert!(any![true, false, true].test(&()));
+///
+/// assert_eq!(any![false].test(&()), false);
+/// assert_eq!(any![false, false].test(&()), false);
+/// assert_eq!(any![false, false, false].test(&()), false);
+/// ```
+///
+/// [filter]: crate::dispatcher::filter::Filter
+#[macro_export]
+macro_rules! any {
+    ($one:expr) => { $one };
+    ($head:expr, $($tail:tt)+) => {
+        $crate::dispatcher::filter::Or::new(
+            $head,
+            $crate::all!($($tail)+)
+        )
+    };
+}

src/dispatcher/handler.rs

@@ -0,0 +1,18 @@
+use std::pin::Pin;
+use std::future::Future;
+
+
+/// Asynchronous handler for event `I` (like `&self, I -> Future` fn)
+pub trait Handler<I> {
+    fn handle(&self, value: I) -> Pin<Box<dyn Future<Output=()> + 'static>>;
+}
+
+impl<Fut, T, F> Handler<T> for F
+where
+    Fut: Future<Output = ()> + 'static,
+    F: Fn(T) -> Fut,
+{
+    fn handle(&self, value: T) -> Pin<Box<dyn Future<Output=()> + 'static>> {
+        Box::pin((self)(value))
+    }
+}

src/dispatcher/mod.rs

@@ -0,0 +1,8 @@
+/// POC implementation of update dispatching
+
+pub mod filter;
+pub mod handler;
+pub mod simple;
+
+pub use filter::Filter;
+pub use handler::Handler;

src/dispatcher/simple/mod.rs

@@ -0,0 +1,254 @@
+use std::{future::Future, pin::Pin};
+
+use crate::{
+    dispatcher::{
+        filter::Filter,
+        handler::Handler,
+    },
+    core::types::{
+        Update,
+        Message,
+        UpdateKind,
+        CallbackQuery,
+        ChosenInlineResult,
+    },
+};
+
+use tokio::stream::Stream;
+
+
+pub type Handlers<T> = Vec<(Box<dyn Filter<T>>, Box<dyn Handler<T>>)>;
+
+pub struct Dispatcher {
+    message_handlers: Handlers<Message>,
+    edited_message_handlers: Handlers<Message>,
+    channel_post_handlers: Handlers<Message>,
+    edited_channel_post_handlers: Handlers<Message>,
+    inline_query_handlers: Handlers<()>,
+    chosen_inline_result_handlers: Handlers<ChosenInlineResult>,
+    callback_query_handlers: Handlers<CallbackQuery>,
+}
+
+impl Dispatcher {
+    pub fn new() -> Self {
+        Dispatcher {
+            message_handlers: Vec::new(),
+            edited_message_handlers: Vec::new(),
+            channel_post_handlers: Vec::new(),
+            edited_channel_post_handlers: Vec::new(),
+            inline_query_handlers: Vec::new(),
+            chosen_inline_result_handlers: Vec::new(),
+            callback_query_handlers: Vec::new()
+        }
+    }
+
+    pub fn message_handler<F, H>(mut self, filter: F, handler: H) -> Self
+    where
+        F: Filter<Message> + 'static,
+        H: Handler<Message> + 'static,
+    {
+        self.message_handlers.push((Box::new(filter), Box::new(handler)));
+        self
+    }
+
+    pub fn edited_message_handler<F, H>(mut self, filter: F, handler: H) -> Self
+    where
+        F: Filter<Message> + 'static,
+        H: Handler<Message> + 'static,
+    {
+        self.edited_message_handlers.push((Box::new(filter), Box::new(handler)));
+        self
+    }
+
+    pub fn channel_post_handler<F, H>(mut self, filter: F, handler: H) -> Self
+    where
+        F: Filter<Message> + 'static,
+        H: Handler<Message> + 'static,
+    {
+        self.channel_post_handlers.push((Box::new(filter), Box::new(handler)));
+        self
+    }
+
+    pub fn edited_channel_post_handler<F, H>(mut self, filter: F, handler: H) -> Self
+    where
+        F: Filter<Message> + 'static,
+        H: Handler<Message> + 'static,
+    {
+        self.edited_channel_post_handlers.push((Box::new(filter), Box::new(handler)));
+        self
+    }
+
+    pub fn inline_query_handler<F, H>(mut self, filter: F, handler: H) -> Self
+    where
+        F: Filter<()> + 'static,
+        H: Handler<()> + 'static,
+    {
+        self.inline_query_handlers.push((Box::new(filter), Box::new(handler)));
+        self
+    }
+
+    pub fn chosen_inline_result_handler<F, H>(mut self, filter: F, handler: H) -> Self
+    where
+        F: Filter<ChosenInlineResult> + 'static,
+        H: Handler<ChosenInlineResult> + 'static,
+    {
+        self.chosen_inline_result_handlers.push((Box::new(filter), Box::new(handler)));
+        self
+    }
+
+    pub fn callback_query_handler<F, H>(mut self, filter: F, handler: H) -> Self
+    where
+        F: Filter<CallbackQuery> + 'static,
+        H: Handler<CallbackQuery> + 'static,
+    {
+        self.callback_query_handlers.push((Box::new(filter), Box::new(handler)));
+        self
+    }
+
+    // TODO: Can someone simplify this?
+    pub async fn dispatch<S>(&mut self, updates: S)
+    where
+        S: Stream<Item=Update>
+    {
+        use futures_util::stream::StreamExt;
+
+        let dp = &*self;
+
+        updates.for_each(|Update { id, kind }| async move {
+            log::debug!("Handled update#{id:?}: {kind:?}", id = id, kind = kind);
+
+            match kind {
+                UpdateKind::Message(mes) => {
+                    call_handler(
+                        find_handler(&dp.message_handlers, &mes),
+                        mes
+                    )
+                        .await;
+                },
+                UpdateKind::EditedMessage(mes) => {
+                    call_handler(
+                        find_handler(&dp.edited_message_handlers, &mes),
+                        mes
+                    )
+                        .await;
+                },
+                UpdateKind::ChannelPost(post) => {
+                    call_handler(
+                        find_handler(&dp.channel_post_handlers, &post),
+                        post
+                    )
+                        .await;
+                },
+                UpdateKind::EditedChannelPost(post) => {
+                    call_handler(
+                        find_handler(&dp.edited_channel_post_handlers, &post),
+                        post
+                    )
+                        .await;
+                },
+                UpdateKind::InlineQuery(query) => {
+                    call_handler(
+                        find_handler(&dp.inline_query_handlers, &query),
+                        query
+                    )
+                        .await;
+                },
+                UpdateKind::ChosenInlineResult(result) => {
+                    call_handler(
+                        find_handler(&dp.chosen_inline_result_handlers, &result),
+                        result
+                    )
+                        .await;
+                },
+                UpdateKind::CallbackQuery(callback) => {
+                    call_handler(
+                        find_handler(&dp.callback_query_handlers, &callback),
+                        callback
+                    )
+                        .await;
+                },
+            }
+        })
+            .await;
+    }
+}
+
+/// Helper function
+fn find_handler<'a, T: std::fmt::Debug>(handlers: &'a Handlers<T>, value: &T) -> Option<&'a Box<Handler<T>>> {
+    let handler = handlers.iter().find_map(|e| {
+        let (filter, handler) = e;
+        if filter.test(value) {
+            Some(handler)
+        } else {
+            None
+        }
+    });
+
+    handler
+}
+
+/// Helper function
+async fn call_handler<T: std::fmt::Debug>(handler: Option<&Box<Handler<T>>>, value: T) {
+    match handler {
+        Some(handler) => handler.handle(value).await,
+        None => log::warn!("Unhandled update: {:?}", value)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test() {
+        use crate::{
+            core::types::{
+                Message, ChatKind, MessageKind, Sender, ForwardKind, MediaKind, Chat, User, Update, UpdateKind
+            },
+            dispatcher::simple::Dispatcher,
+        };
+
+        let mes = Message {
+            id: 6534,
+            date: 1567898953,
+            chat: Chat {
+                id: 218485655,
+                photo: None,
+                kind: ChatKind::Private {
+                    type_: (),
+                    first_name: Some("W".to_string()),
+                    last_name: None,
+                    username: Some("WaffleLapkin".to_string()),
+                },
+            },
+            message_kind: MessageKind::IncomingMessage {
+                from: Sender::User(User {
+                    id: 457569668,
+                    is_bot: true,
+                    first_name: "BT".to_string(),
+                    last_name: None,
+                    username: Some("BloodyTestBot".to_string()),
+                    language_code: None,
+                }),
+                forward_kind: ForwardKind::Origin {
+                    reply_to_message: None,
+                },
+                edit_date: None,
+                media_kind: MediaKind::Text {
+                    text: "text".to_string(),
+                    entities: vec![],
+                },
+                reply_markup: None,
+            },
+        };
+
+        async fn handler(mes: Message) {
+            println!("{:#?}", mes)
+        }
+
+        let mut dp = Dispatcher::new()
+            .message_handler(true, handler);
+
+        dp.dispatch(tokio::stream::iter(vec![Update { id: 0, kind: UpdateKind::Message(mes) }])).await;
+    }
+}

src/lib.rs

@@ -6,3 +6,4 @@ extern crate serde;
 pub mod bot;
 pub mod core;
 pub mod keyboards;
+pub mod dispatcher;