Merge pull request #62 from telebofr/dispatcher_trait

Add dispatcher trait

src/dispatcher/mod.rs

@@ -7,3 +7,10 @@ pub mod updater;
 
 pub use filter::Filter;
 pub use handler::Handler;
+
+use async_trait::async_trait;
+
+#[async_trait(?Send)]
+pub trait Dispatcher<'a, U> {
+    async fn dispatch(&'a mut self, updater: U);
+}

src/dispatcher/simple/mod.rs

@@ -1,10 +1,14 @@
 pub mod error_policy;
 
+use futures::StreamExt;
+use async_trait::async_trait;
+
 use crate::{
     dispatcher::{
         filter::Filter,
         handler::Handler,
         updater::Updater,
+        simple::error_policy::ErrorPolicy,
     },
     types::{
         Update,
@@ -15,10 +19,6 @@ use crate::{
     },
 };
 
-use futures::StreamExt;
-use crate::dispatcher::simple::error_policy::ErrorPolicy;
-
-
 type Handlers<'a, T, E> = Vec<(Box<dyn Filter<T> + 'a>, Box<dyn Handler<'a, T, E> + 'a>)>;
 
 /// Dispatcher that dispatches updates from telegram.
@@ -166,9 +166,9 @@ where
     }
 
     // TODO: Can someone simplify this?
-    pub async fn dispatch<U, UE>(&mut self, updates: U)
+    pub async fn dispatch<U>(&mut self, updates: U)
     where
-        U: Updater<UE> + 'a
+        U: Updater + 'a
     {
         updates.for_each(|res| {
             async {
@@ -219,6 +219,16 @@ where
     }
 }
 
+#[async_trait(?Send)]
+impl<'a, U, E> crate::dispatcher::Dispatcher<'a, U> for Dispatcher<'a, E>
+where
+    E: std::fmt::Debug,
+    U: Updater + 'a,
+{
+    async fn dispatch(&'a mut self, updater: U) {
+        Dispatcher::dispatch(self, updater).await
+    }
+}
 
 #[cfg(test)]
 mod tests {

src/dispatcher/updater.rs

@@ -99,7 +99,9 @@ use crate::{
 /// [GetUpdates]: crate::requests::GetUpdates
 /// [getting updates]: https://core.telegram.org/bots/api#getting-updates
 /// [wiki]: https://en.wikipedia.org/wiki/Push_technology#Long_polling
-pub trait Updater<E>: Stream<Item=Result<Update, E>> {}
+pub trait Updater: Stream<Item=Result<Update, <Self as Updater>::Error>> {
+    type Error;
+}
 
 #[pin_project]
 pub struct StreamUpdater<S> {
@@ -116,14 +118,16 @@ impl<S> StreamUpdater<S> {
 impl<S, E> Stream for StreamUpdater<S> where S: Stream<Item=Result<Update, E>> {
     type Item = Result<Update, E>;
 
-    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         self.project().stream.poll_next(cx)
     }
 }
 
-impl<S, E> Updater<E> for StreamUpdater<S> where S: Stream<Item=Result<Update, E>> {}
+impl<S, E> Updater for StreamUpdater<S> where S: Stream<Item=Result<Update, E>> {
+    type Error = E;
+}
 
-pub fn polling<'a>(bot: &'a Bot) -> impl Updater<RequestError> + 'a {
+pub fn polling<'a>(bot: &'a Bot) -> impl Updater<Error = RequestError> + 'a {
     let stream = stream::unfold((bot, 0), |(bot, mut offset)| async move {
         // this match converts Result<Vec<_>, _> -> Vec<Result<_, _>>
         let updates = match bot.get_updates().offset(offset).send().await {