From 175e7572b224802500d3d907d9b04e7a5f42b460 Mon Sep 17 00:00:00 2001
From: Waffle <wafflewafflerov@gmail.com>
Date: Wed, 16 Oct 2019 14:47:09 +0300
Subject: [PATCH] Add dispatcher trait

---
 src/dispatcher/mod.rs        |  7 +++++++
 src/dispatcher/simple/mod.rs | 22 ++++++++++++++++------
 src/dispatcher/updater.rs    | 12 ++++++++----
 3 files changed, 31 insertions(+), 10 deletions(-)

diff --git a/src/dispatcher/mod.rs b/src/dispatcher/mod.rs
index b957411a..2011c220 100644
--- a/src/dispatcher/mod.rs
+++ b/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);
+}
diff --git a/src/dispatcher/simple/mod.rs b/src/dispatcher/simple/mod.rs
index 63ce3a28..16b3d6f6 100644
--- a/src/dispatcher/simple/mod.rs
+++ b/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 {
diff --git a/src/dispatcher/updater.rs b/src/dispatcher/updater.rs
index 8dd14da1..74b0164b 100644
--- a/src/dispatcher/updater.rs
+++ b/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 {