Merge branch 'dev' into update_macros

CHANGELOG.md

@@ -10,6 +10,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 <!-- TODO: use a version once teloxide-macros is released -->
 - Updated `teloxide-macros` see its [changelog](https://github.com/teloxide/teloxide-macros/blob/master/CHANGELOG.md#unreleased) for more
+- `UpdateListener` now has an associated type `Err` instead of a generic
+- `AsUpdateStream` now has an associated type `StreamErr` instead of a generic
 
 ## 0.10.1 - 2022-07-22
 

src/dispatching/dispatcher.rs

@@ -287,14 +287,14 @@ where
     /// This method adds the same dependencies as [`Dispatcher::dispatch`].
     ///
     /// [`shutdown`]: ShutdownToken::shutdown
-    pub async fn dispatch_with_listener<'a, UListener, ListenerE, Eh>(
+    pub async fn dispatch_with_listener<'a, UListener, Eh>(
         &'a mut self,
         mut update_listener: UListener,
         update_listener_error_handler: Arc<Eh>,
     ) where
-        UListener: UpdateListener<ListenerE> + 'a,
-        Eh: ErrorHandler<ListenerE> + 'a,
-        ListenerE: Debug,
+        UListener: UpdateListener + 'a,
+        Eh: ErrorHandler<UListener::Err> + 'a,
+        UListener::Err: Debug,
     {
         // FIXME: there should be a way to check if dependency is already inserted
         let me = self.bot.get_me().send().await.expect("Failed to retrieve 'me'");

src/dispatching/repls/commands_repl.rs

@@ -78,7 +78,7 @@ where
 /// [`commands_repl`]: crate::dispatching::repls::commands_repl()
 /// [`UpdateListener`]: crate::dispatching::update_listeners::UpdateListener
 #[cfg(feature = "ctrlc_handler")]
-pub async fn commands_repl_with_listener<'a, R, Cmd, H, L, ListenerE, E, Args>(
+pub async fn commands_repl_with_listener<'a, R, Cmd, H, L, E, Args>(
     bot: R,
     handler: H,
     listener: L,
@@ -86,8 +86,8 @@ pub async fn commands_repl_with_listener<'a, R, Cmd, H, L, ListenerE, E, Args>(
 ) where
     Cmd: BotCommands + Send + Sync + 'static,
     H: Injectable<DependencyMap, Result<(), E>, Args> + Send + Sync + 'static,
-    L: UpdateListener<ListenerE> + Send + 'a,
-    ListenerE: Debug + Send + 'a,
+    L: UpdateListener + Send + 'a,
+    L::Err: Debug + Send + 'a,
     R: Requester + Clone + Send + Sync + 'static,
     E: Debug + Send + Sync + 'static,
 {

src/dispatching/repls/repl.rs

@@ -60,11 +60,11 @@ where
 /// [`repl`]: crate::dispatching::repls::repl()
 /// [`UpdateListener`]: crate::dispatching::update_listeners::UpdateListener
 #[cfg(feature = "ctrlc_handler")]
-pub async fn repl_with_listener<'a, R, H, E, L, ListenerE, Args>(bot: R, handler: H, listener: L)
+pub async fn repl_with_listener<'a, R, H, E, L, Args>(bot: R, handler: H, listener: L)
 where
     H: Injectable<DependencyMap, Result<(), E>, Args> + Send + Sync + 'static,
-    L: UpdateListener<ListenerE> + Send + 'a,
-    ListenerE: Debug,
+    L: UpdateListener + Send + 'a,
+    L::Err: Debug,
     Result<(), E>: OnError<E>,
     E: Debug + Send + Sync + 'static,
     R: Requester + Clone + Send + Sync + 'static,

src/dispatching/update_listeners.rs

@@ -59,7 +59,12 @@ pub use self::{
 /// - [`AsUpdateStream::as_stream`]
 ///
 /// [module-level documentation]: mod@self
-pub trait UpdateListener<E>: for<'a> AsUpdateStream<'a, E> {
+pub trait UpdateListener:
+    for<'a> AsUpdateStream<'a, StreamErr = <Self as UpdateListener>::Err>
+{
+    /// The type of errors that can be returned from this listener.
+    type Err;
+
     /// The type of token which allows to stop this listener.
     type StopToken: StopToken + Send;
 
@@ -110,16 +115,19 @@ pub trait UpdateListener<E>: for<'a> AsUpdateStream<'a, E> {
 /// [`UpdateListener`]'s supertrait/extension.
 ///
 /// This trait is a workaround to not require GAT.
-pub trait AsUpdateStream<'a, E> {
+pub trait AsUpdateStream<'a> {
+    /// Error that can be returned from the [`Stream`]
+    ///
+    /// [`Stream`]: AsUpdateStream::Stream
+    // NB: This should be named differently to `UpdateListener::Err`, so that it's
+    // unambiguous
+    type StreamErr;
+
     /// The stream of updates from Telegram.
-    // HACK: There is currently no way to write something like
-    // `-> impl for<'a> AsUpdateStream<'a, E, Stream: Send>`. Since we return
-    // `impl UpdateListener<E>` from `polling`, we need to have `Send` bound here,
-    // to make the stream `Send`.
-    //
-    // Without this it's, for example, impossible to spawn a tokio task with
-    // teloxide polling.
-    type Stream: Stream<Item = Result<Update, E>> + Send + 'a;
+    // NB: `Send` is not strictly required here, but it makes it easier to return
+    //     `impl AsUpdateStream` and also you want `Send` streams almost (?) always
+    //     anyway.
+    type Stream: Stream<Item = Result<Update, Self::StreamErr>> + Send + 'a;
 
     /// Creates the update [`Stream`].
     ///
@@ -128,9 +136,9 @@ pub trait AsUpdateStream<'a, E> {
 }
 
 #[inline(always)]
-pub(crate) fn assert_update_listener<L, E>(listener: L) -> L
+pub(crate) fn assert_update_listener<L>(listener: L) -> L
 where
-    L: UpdateListener<E>,
+    L: UpdateListener,
 {
     listener
 }

src/dispatching/update_listeners/polling.rs

@@ -291,7 +291,8 @@ pub struct PollingStream<'a, B: Requester> {
     in_flight: Option<<B::GetUpdates as Request>::Send>,
 }
 
-impl<B: Requester + Send + 'static> UpdateListener<B::Err> for Polling<B> {
+impl<B: Requester + Send + 'static> UpdateListener for Polling<B> {
+    type Err = B::Err;
     type StopToken = AsyncStopToken;
 
     fn stop_token(&mut self) -> Self::StopToken {
@@ -309,7 +310,8 @@ impl<B: Requester + Send + 'static> UpdateListener<B::Err> for Polling<B> {
     }
 }
 
-impl<'a, B: Requester + Send + 'a> AsUpdateStream<'a, B::Err> for Polling<B> {
+impl<'a, B: Requester + Send + 'a> AsUpdateStream<'a> for Polling<B> {
+    type StreamErr = B::Err;
     type Stream = PollingStream<'a, B>;
 
     fn as_stream(&'a mut self) -> Self::Stream {

src/dispatching/update_listeners/stateful_listener.rs

@@ -5,7 +5,7 @@ use futures::Stream;
 use crate::{
     dispatching::{
         stop_token::{self, StopToken},
-        update_listeners::{AsUpdateStream, UpdateListener},
+        update_listeners::{assert_update_listener, AsUpdateStream, UpdateListener},
     },
     types::{AllowedUpdate, Update},
 };
@@ -103,7 +103,7 @@ where
     }
 }
 
-impl<'a, St, Assf, Sf, Hauf, Thf, Strm, E> AsUpdateStream<'a, E>
+impl<'a, St, Assf, Sf, Hauf, Thf, Strm, E> AsUpdateStream<'a>
     for StatefulListener<St, Assf, Hauf, Sf, Thf>
 where
     (St, Strm): 'a,
@@ -111,6 +111,7 @@ where
     Assf: FnMut(&'a mut St) -> Strm,
     Strm: Stream<Item = Result<Update, E>>,
 {
+    type StreamErr = E;
     type Stream = Strm;
 
     fn as_stream(&'a mut self) -> Self::Stream {
@@ -118,15 +119,15 @@ where
     }
 }
 
-impl<St, Assf, Sf, Hauf, Stt, Thf, E> UpdateListener<E>
-    for StatefulListener<St, Assf, Sf, Hauf, Thf>
+impl<St, Assf, Sf, Hauf, Stt, Thf, E> UpdateListener for StatefulListener<St, Assf, Sf, Hauf, Thf>
 where
-    Self: for<'a> AsUpdateStream<'a, E>,
+    Self: for<'a> AsUpdateStream<'a, StreamErr = E>,
     Sf: FnMut(&mut St) -> Stt,
     Stt: StopToken + Send,
     Hauf: FnMut(&mut St, &mut dyn Iterator<Item = AllowedUpdate>),
     Thf: Fn(&St) -> Option<Duration>,
 {
+    type Err = E;
     type StopToken = Stt;
 
     fn stop_token(&mut self) -> Stt {
@@ -143,10 +144,3 @@ where
         self.timeout_hint.as_ref().and_then(|f| f(&self.state))
     }
 }
-
-fn assert_update_listener<L, E>(l: L) -> L
-where
-    L: UpdateListener<E>,
-{
-    l
-}

src/dispatching/update_listeners/webhooks/axum.rs

@@ -38,7 +38,10 @@ use crate::{
 ///
 /// [`axum_to_router`] and [`axum_no_setup`] for lower-level versions of this
 /// function.
-pub async fn axum<R>(bot: R, options: Options) -> Result<impl UpdateListener<Infallible>, R::Err>
+pub async fn axum<R>(
+    bot: R,
+    options: Options,
+) -> Result<impl UpdateListener<Err = Infallible>, R::Err>
 where
     R: Requester + Send + 'static,
     <R as Requester>::DeleteWebhook: Send,
@@ -107,7 +110,10 @@ where
 pub async fn axum_to_router<R>(
     bot: R,
     mut options: Options,
-) -> Result<(impl UpdateListener<Infallible>, impl Future<Output = ()> + Send, axum::Router), R::Err>
+) -> Result<
+    (impl UpdateListener<Err = Infallible>, impl Future<Output = ()> + Send, axum::Router),
+    R::Err,
+>
 where
     R: Requester + Send,
     <R as Requester>::DeleteWebhook: Send,
@@ -148,7 +154,7 @@ where
 /// function.
 pub fn axum_no_setup(
     options: Options,
-) -> (impl UpdateListener<Infallible>, impl Future<Output = ()>, axum::Router) {
+) -> (impl UpdateListener<Err = Infallible>, impl Future<Output = ()>, axum::Router) {
     use crate::{
         dispatching::{
             stop_token::AsyncStopToken,

src/utils/shutdown_token.rs

@@ -93,7 +93,7 @@ impl fmt::Display for IdleShutdownError {
 
 impl std::error::Error for IdleShutdownError {}
 
-pub(crate) fn shutdown_check_timeout_for<E>(update_listener: &impl UpdateListener<E>) -> Duration {
+pub(crate) fn shutdown_check_timeout_for(update_listener: &impl UpdateListener) -> Duration {
     const MIN_SHUTDOWN_CHECK_TIMEOUT: Duration = Duration::from_secs(1);
     const DZERO: Duration = Duration::ZERO;