OpenSource/teloxide
OpenSource/teloxide
1
0
Fork 0
mirror of https://github.com/teloxide/teloxide.git synced 2024-12-22 22:46:39 +01:00
Code Issues Projects Releases Packages Wiki Activity

Fix merge conflicts

Browse source
This commit is contained in:
Hirrolot 2022-02-03 20:39:32 +06:00
commit 242bfef885
22 changed files with 208 additions and 215 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
.github/ISSUE_TEMPLATE/bug_report.md vendored
View file

@ -18,8 +18,3 @@ Instead, this happened: _explanation_
## Meta
- `teloxide` version: <!-- (e.g.: `0.3.1`) -->
- rustc version:
```
<version>
```
<!-- use `rustc --version --verbose` to get it -->

5
.github/ISSUE_TEMPLATE/parse-error.md vendored
View file

@ -19,11 +19,6 @@ When using `<...>` method I've got `RequestError::InvalidJson` error with the f
## Meta
- `teloxide` version: <!-- (e.g.: `0.3.1`) -->
- rustc version:
```
<version>
```
<!-- use `rustc --version --verbose` to get it -->
### Additional context

5
.github/ISSUE_TEMPLATE/unknown-telegram-error.md vendored
View file

@ -19,11 +19,6 @@ When using `<...>` method I've got `ApiError::Unknown` error with the following
## Meta
- `teloxide` version: <!-- (e.g.: `0.3.1`) -->
- rustc version:
```
<version>
```
<!-- use `rustc --version --verbose` to get it -->
### Additional context

5
CHANGELOG.md
View file

@ -10,6 +10,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Require that `AsUpdateStream::Stream` is `Send`
### Fixed
- Infinite retries while stopping polling listener ([issue 496](https://github.com/teloxide/teloxide/issues/496))
- `polling{,_default}` and it's `Stream` and `StopToken` not being `Send` (and by extension fix the same problem with `repl`s)
## 0.5.3 - 2021-10-25
### Fixed

9
Cargo.toml
View file

@ -57,10 +57,10 @@ full = [
]
[dependencies]
teloxide-core = { version = "0.3.3", default-features = false }
#teloxide-core = { git = "https://github.com/teloxide/teloxide-core.git", rev = "...", default-features = false }
#teloxide-macros = { version = "0.4", optional = true }
teloxide-macros = { path = "../teloxide-macros", optional = true }
#teloxide-core = { version = "0.3.3", default-features = false }
teloxide-core = { git = "https://github.com/teloxide/teloxide-core.git", rev = "3ccb8f0", default-features = false }
teloxide-macros = { git = "https://github.com/teloxide/teloxide-macros.git", branch = "dispatching2", optional = true }
serde_json = "1.0"
serde = { version = "1.0", features = ["derive"] }
@ -91,6 +91,7 @@ redis = { version = "0.20", features = ["tokio-comp"], optional = true }
serde_cbor = { version = "0.11", optional = true }
bincode = { version = "1.3", optional = true }
frunk = { version = "0.4", optional = true }
aquamarine = "0.1.11"
[dev-dependencies]
smart-default = "0.6.0"

2
README.md
View file

@ -392,6 +392,7 @@ Feel free to propose your own bot to our collection!
- [WaffleLapkin/crate_upd_bot](https://github.com/WaffleLapkin/crate_upd_bot) -- A bot that notifies about crate updates.
- [mxseev/logram](https://github.com/mxseev/logram) -- Utility that takes logs from anywhere and sends them to Telegram.
- [alexkonovalov/PedigreeBot](https://github.com/alexkonovalov/PedigreeBot) -- A Telegram bot for building family trees.
- [Hermitter/tepe](https://github.com/Hermitter/tepe) -- A CLI to command a bot to send messages and files over Telegram.
- [dracarys18/grpmr-rs](https://github.com/dracarys18/grpmr-rs) -- A Telegram group manager bot with variety of extra features.
- [steadylearner/subreddit_reader](https://github.com/steadylearner/Rust-Full-Stack/tree/master/commits/teloxide/subreddit_reader) -- A bot that shows the latest posts at Rust subreddit.
@ -409,6 +410,5 @@ Feel free to propose your own bot to our collection!
- [crapstone/hsctt](https://codeberg.org/crapstones-bots/hsctt) -- A Telegram bot that searches for HTTP status codes in all messages and replies with the text form.
- [alenpaul2001/AurSearchBot](https://gitlab.com/alenpaul2001/aursearchbot) -- Telegram bot for searching AUR in inline mode.
## Contributing
See [CONRIBUTING.md](https://github.com/teloxide/teloxide/blob/master/CONTRIBUTING.md).

14
examples/admin.rs
View file

@ -1,6 +1,6 @@
use std::{error::Error, str::FromStr};
use chrono::{DateTime, Duration, NaiveDateTime, Utc};
use chrono::Duration;
use teloxide::{prelude2::*, types::ChatPermissions, utils::command::BotCommand};
// Derive BotCommand to parse text with a command into this enumeration.
@ -90,12 +90,9 @@ async fn mute_user(
bot.restrict_chat_member(
msg.chat_id(),
replied.from().expect("Must be MessageKind::Common").id,
ChatPermissions::default(),
)
.until_date(
DateTime::<Utc>::from_utc(NaiveDateTime::from_timestamp(msg.date as i64, 0), Utc)
+ time,
ChatPermissions::empty(),
)
.until_date(msg.date + time)
.await?;
}
None => {
@ -118,10 +115,7 @@ async fn ban_user(
msg.chat_id(),
replied.from().expect("Must be MessageKind::Common").id,
)
.until_date(
DateTime::<Utc>::from_utc(NaiveDateTime::from_timestamp(msg.date as i64, 0), Utc)
+ time,
)
.until_date(msg.date + time)
.await?;
}
None => {

6
examples/heroku_ping_pong.rs
View file

@ -59,10 +59,8 @@ pub async fn webhook(bot: AutoSend<Bot>) -> impl update_listeners::UpdateListene
let server = warp::post()
.and(warp::path(path))
.and(warp::body::json())
.map(move |json: serde_json::Value| {
if let Ok(update) = Update::try_parse(&json) {
tx.send(Ok(update)).expect("Cannot send an incoming update from the webhook")
}
.map(move |update: Update| {
tx.send(Ok(update)).expect("Cannot send an incoming update from the webhook");
StatusCode::OK
})

4
examples/inline.rs
View file

@ -42,8 +42,8 @@ async fn main() {
))),
)
.description("DuckDuckGo Search")
.thumb_url("https://duckduckgo.com/assets/logo_header.v108.png")
.url("https://duckduckgo.com/about"); // Note: This is the url that will open if they click the thumbnail
.thumb_url("https://duckduckgo.com/assets/logo_header.v108.png".parse().unwrap())
.url("https://duckduckgo.com/about".parse().unwrap()); // Note: This is the url that will open if they click the thumbnail
let results = vec![
InlineQueryResult::Article(google_search),

6
examples/ngrok_ping_pong.rs
View file

@ -51,10 +51,8 @@ pub async fn webhook(bot: AutoSend<Bot>) -> impl update_listeners::UpdateListene
let server = warp::post()
.and(warp::body::json())
.map(move |json: serde_json::Value| {
if let Ok(update) = Update::try_parse(&json) {
tx.send(Ok(update)).expect("Cannot send an incoming update from the webhook")
}
.map(move |update: Update| {
tx.send(Ok(update)).expect("Cannot send an incoming update from the webhook");
StatusCode::OK
})

4
rust-toolchain.toml Normal file
View file

@ -0,0 +1,4 @@
[toolchain]
channel = "nightly-2021-10-24"
components = ["rustfmt", "clippy"]
profile = "minimal"

23
src/dispatching/dispatcher.rs
View file

@ -20,8 +20,9 @@ use futures::{stream::FuturesUnordered, Future, StreamExt};
use teloxide_core::{
requests::Requester,
types::{
AllowedUpdate, CallbackQuery, ChatMemberUpdated, ChosenInlineResult, InlineQuery, Message,
Poll, PollAnswer, PreCheckoutQuery, ShippingQuery, Update, UpdateKind,
AllowedUpdate, CallbackQuery, ChatJoinRequest, ChatMemberUpdated, ChosenInlineResult,
InlineQuery, Message, Poll, PollAnswer, PreCheckoutQuery, ShippingQuery, Update,
UpdateKind,
},
};
use tokio::{
@ -52,6 +53,7 @@ pub struct Dispatcher<R> {
poll_answers_queue: Tx<R, PollAnswer>,
my_chat_members_queue: Tx<R, ChatMemberUpdated>,
chat_members_queue: Tx<R, ChatMemberUpdated>,
chat_join_requests_queue: Tx<R, ChatJoinRequest>,
running_handlers: FuturesUnordered<JoinHandle<()>>,
@ -81,6 +83,7 @@ where
poll_answers_queue: None,
my_chat_members_queue: None,
chat_members_queue: None,
chat_join_requests_queue: None,
running_handlers: FuturesUnordered::new(),
state: <_>::default(),
shutdown_notify_back: <_>::default(),
@ -264,7 +267,7 @@ where
pub async fn dispatch(&mut self)
where
R: Requester + Clone,
<R as Requester>::GetUpdatesFaultTolerant: Send,
<R as Requester>::GetUpdates: Send,
{
let listener = update_listeners::polling_default(self.requester.clone()).await;
let error_handler =
@ -447,6 +450,20 @@ where
chat_member_updated,
"UpdateKind::MyChatMember",
),
UpdateKind::ChatJoinRequest(chat_join_request) => send(
&self.requester,
&self.chat_join_requests_queue,
chat_join_request,
"UpdateKind::ChatJoinRequest",
),
UpdateKind::Error(err) => {
log::error!(
"Cannot parse an update.\nError: {:?}\n\
This is a bug in teloxide-core, please open an issue here: \
https://github.com/teloxide/teloxide-core/issues.",
err,
);
}
}
}
}

2
src/dispatching/repls/commands_repl.rs
View file

@ -31,7 +31,7 @@ where
HandlerE: Debug + Send,
N: Into<String> + Send + 'static,
R: Requester + Send + Clone + 'static,
<R as Requester>::GetUpdatesFaultTolerant: Send,
<R as Requester>::GetUpdates: Send,
{
let cloned_requester = requester.clone();

2
src/dispatching/repls/dialogues_repl.rs
View file

@ -29,7 +29,7 @@ where
D: Clone + Default + Send + 'static,
Fut: Future<Output = DialogueStage<D>> + Send + 'static,
R: Requester + Send + Clone + 'static,
<R as Requester>::GetUpdatesFaultTolerant: Send,
<R as Requester>::GetUpdates: Send,
{
let cloned_requester = requester.clone();

11
src/dispatching/repls/repl.rs
View file

@ -28,7 +28,7 @@ where
Result<(), E>: OnError<E>,
E: Debug + Send,
R: Requester + Send + Clone + 'static,
<R as Requester>::GetUpdatesFaultTolerant: Send,
<R as Requester>::GetUpdates: Send,
{
let cloned_requester = requester.clone();
repl_with_listener(
@ -83,3 +83,12 @@ pub async fn repl_with_listener<'a, R, H, Fut, E, L, ListenerE>(
)
.await;
}
#[test]
fn repl_is_send() {
let bot = crate::Bot::new("");
let repl = crate::repl(bot, |_| async { crate::respond(()) });
assert_send(&repl);
fn assert_send(_: &impl Send) {}
}

116
src/dispatching/update_listeners.rs
View file

@ -1,107 +1,27 @@
//! Receiving updates from Telegram.
//!
//! The key trait here is [`UpdateListener`]. You can get it by these functions:
//! The key trait here is [`UpdateListener`]. You can get its implementation
//! using one these functions:
//!
//! - [`polling_default`], which returns a default long polling listener.
//! - [`polling`], which returns a long/short polling listener with your
//! - [`polling`], which returns a long polling listener with your
//! configuration.
//!
//! And then you can extract updates from it and pass them directly to a
//! dispatcher.
//! And then you can extract updates from it or pass them directly to a
//! [`Dispatcher`].
//!
//! Telegram supports two ways of [getting updates]: [long]/[short] polling and
//! [webhook].
//!
//! # Long Polling
//!
//! In long polling, you just call [`Box::get_updates`] every N seconds.
//!
//! ## Example
//!
//! <pre>
//! tg bot
//! | |
//! |<---------------------------| Updates? (Bot::get_updates call)
//! ↑ ↑
//! | timeout<a id="1b" href="#1">^1</a> |
//! ↓ ↓
//! Nope |--------------------------->|
//! ↑ ↑
//! | delay between Bot::get_updates<a id="2b" href="#2">^2</a> |
//! ↓ ↓
//! |<---------------------------| Updates?
//! ↑ ↑
//! | timeout<a id="3b" href="#3">^3</a> |
//! ↓ ↓
//! Yes |-------[updates 0, 1]------>|
//! ↑ ↑
//! | delay |
//! ↓ ↓
//! |<-------[offset = 1]--------| Updates?<a id="4b" href="#4">^4</a>
//! ↑ ↑
//! | timeout |
//! ↓ ↓
//! Yes |---------[update 2]-------->|
//! ↑ ↑
//! | delay |
//! ↓ ↓
//! |<-------[offset = 2]--------| Updates?
//! ↑ ↑
//! | timeout |
//! ↓ ↓
//! Nope |--------------------------->|
//! ↑ ↑
//! | delay |
//! ↓ ↓
//! |<-------[offset = 2]--------| Updates?
//! ↑ ↑
//! | timeout |
//! ↓ ↓
//! Nope |--------------------------->|
//! ↑ ↑
//! | delay |
//! ↓ ↓
//! |<-------[offset = 2]--------| Updates?
//! ↑ ↑
//! | timeout |
//! ↓ ↓
//! Yes |-------[updates 2..5]------>|
//! ↑ ↑
//! | delay |
//! ↓ ↓
//! |<-------[offset = 5]--------| Updates?
//! ↑ ↑
//! | timeout |
//! ↓ ↓
//! Nope |--------------------------->|
//! | |
//! ~ and so on, and so on ~
//! </pre>
//!
//! <a id="1" href="#1b">^1</a> A timeout can be even 0
//! (this is also called short polling),
//! but you should use it **only** for testing purposes.
//!
//! <a id="2" href="#2b">^2</a> Large delays will cause in bot lags,
//! so delay shouldn't exceed second.
//!
//! <a id="3" href="#3b">^3</a> Note that if Telegram already have updates for
//! you it will answer you **without** waiting for a timeout.
//!
//! <a id="4" href="#4b">^4</a> `offset = N` means that we've already received
//! updates `0..=N`.
//!
//! # Webhooks
//! See the [README FAQ about webhooks](https://github.com/teloxide/teloxide/blob/master/README.md#faq).
//! Telegram supports two ways of [getting updates]: [long polling] and
//! [webhooks]. Currently, only the former one is implemented (see [`polling()`]
//! and [`polling_default`]). See also [README FAQ about webhooks](https://github.com/teloxide/teloxide/blob/master/README.md#faq).
//!
//! [`UpdateListener`]: UpdateListener
//! [`polling_default`]: polling_default
//! [`polling`]: polling()
//! [`Dispatcher`]: crate::dispatching::Dispatcher
//! [`Box::get_updates`]: crate::requests::Requester::get_updates
//! [getting updates]: https://core.telegram.org/bots/api#getting-updates
//! [long]: https://en.wikipedia.org/wiki/Push_technology#Long_polling
//! [short]: https://en.wikipedia.org/wiki/Polling_(computer_science)
//! [webhook]: https://en.wikipedia.org/wiki/Webhook
//! [long polling]: https://en.wikipedia.org/wiki/Push_technology#Long_polling
//! [webhooks]: https://en.wikipedia.org/wiki/Webhook
use futures::Stream;
@ -122,19 +42,15 @@ pub use self::{
/// An update listener.
///
/// Implementors of this trait allow getting updates from Telegram.
///
/// Currently Telegram has 2 ways of getting updates -- [polling] and
/// [webhooks]. Currently, only the former one is implemented (see [`polling()`]
/// and [`polling_default`])
/// Implementors of this trait allow getting updates from Telegram. See
/// [module-level documentation] for more.
///
/// Some functions of this trait are located in the supertrait
/// ([`AsUpdateStream`]), see also:
/// - [`AsUpdateStream::Stream`]
/// - [`AsUpdateStream::as_stream`]
///
/// [polling]: self#long-polling
/// [webhooks]: self#webhooks
/// [module-level documentation]: mod@self
pub trait UpdateListener<E>: for<'a> AsUpdateStream<'a, E> {
/// The type of token which allows to stop this listener.
type StopToken: StopToken;
@ -150,8 +66,8 @@ pub trait UpdateListener<E>: for<'a> AsUpdateStream<'a, E> {
/// Implementors of this function are encouraged to stop listening for
/// updates as soon as possible and return `None` from the update stream as
/// soon as all cached updates are returned.
#[must_use = "This function doesn't stop listening, to stop listening you need to call stop on \
the returned token"]
#[must_use = "This function doesn't stop listening, to stop listening you need to call `stop` \
on the returned token"]
fn stop_token(&mut self) -> Self::StopToken;
/// Hint which updates should the listener listen for.

177
src/dispatching/update_listeners/polling.rs
View file

@ -7,12 +7,12 @@ use futures::{
use crate::{
dispatching::{
stop_token::{AsyncStopFlag, AsyncStopToken},
stop_token::{AsyncStopFlag, AsyncStopToken, StopToken},
update_listeners::{stateful_listener::StatefulListener, UpdateListener},
},
payloads::GetUpdates,
payloads::{GetUpdates, GetUpdatesSetters as _},
requests::{HasPayload, Request, Requester},
types::{AllowedUpdate, SemiparsedVec, Update},
types::{AllowedUpdate, Update},
};
/// Returns a long polling update listener with `timeout` of 10 seconds.
@ -22,36 +22,118 @@ use crate::{
/// ## Notes
///
/// This function will automatically delete a webhook if it was set up.
pub async fn polling_default<R>(requester: R) -> impl UpdateListener<R::Err>
pub async fn polling_default<R>(
requester: R,
) -> impl UpdateListener<R::Err, StopToken = impl Send + StopToken>
where
R: Requester + Send + 'static,
<R as Requester>::GetUpdatesFaultTolerant: Send,
<R as Requester>::GetUpdates: Send,
{
delete_webhook_if_setup(&requester).await;
polling(requester, Some(Duration::from_secs(10)), None, None)
}
/// Returns a long/short polling update listener with some additional options.
#[cfg_attr(doc, aquamarine::aquamarine)]
/// Returns a long polling update listener with some additional options.
///
/// - `bot`: Using this bot, the returned update listener will receive updates.
/// - `timeout`: A timeout for polling.
/// - `timeout`: A timeout in seconds for polling.
/// - `limit`: Limits the number of updates to be retrieved at once. Values
/// between 1—100 are accepted.
/// - `allowed_updates`: A list the types of updates you want to receive.
///
/// See [`GetUpdates`] for defaults.
///
/// See also: [`polling_default`](polling_default).
///
/// [`GetUpdates`]: crate::payloads::GetUpdates
/// ## Notes
///
/// - `timeout` should not be bigger than http client timeout, see
/// [`default_reqwest_settings`] for default http client settings.
/// - [`repl`]s and [`Dispatcher`] use [`hint_allowed_updates`] to set
/// `allowed_updates`, so you rarely need to pass `allowed_updates`
/// explicitly.
///
/// [`default_reqwest_settings`]: teloxide::net::default_reqwest_settings
/// [`repl`]: fn@crate::repl
/// [`Dispatcher`]: crate::dispatching::Dispatcher
/// [`hint_allowed_updates`]:
/// crate::dispatching::update_listeners::UpdateListener::hint_allowed_updates
///
/// ## How it works
///
/// Long polling works by repeatedly calling [`Bot::get_updates`][get_updates].
/// If telegram has any updates, it returns them immediately, otherwise it waits
/// until either it has any updates or `timeout` expires.
///
/// Each [`get_updates`][get_updates] call includes an `offset` parameter equal
/// to the latest update id + one, that allows to only receive updates that has
/// not been received before.
///
/// When telegram receives a [`get_updates`][get_updates] request with `offset =
/// N` it forgets any updates with id < `N`. When `polling` listener is stopped,
/// it sends [`get_updates`][get_updates] with `timeout = 0, limit = 1` and
/// appropriate `offset`, so future bot restarts won't see updates that were
/// already seen.
///
/// Consumers of a `polling` update listener then need to repeatedly call
/// [`futures::StreamExt::next`] to get the updates.
///
/// Here is an example diagram that shows these interactions between consumers
/// like [`Dispatcher`], `polling` update listener and telegram.
///
/// ```mermaid
/// sequenceDiagram
/// participant C as Consumer
/// participant P as polling
/// participant T as Telegram
///
/// link C: Dispatcher @ ../struct.Dispatcher.html
/// link C: repl @ ../../fn.repl.html
///
/// C->>P: next
///
/// P->>+T: Updates? (offset = 0)
/// Note right of T: timeout
/// T->>-P: None
///
/// P->>+T: Updates? (offset = 0)
/// Note right of T: <= timeout
/// T->>-P: updates with ids [3, 4]
///
/// P->>C: update(3)
///
/// C->>P: next
/// P->>C: update(4)
///
/// C->>P: next
///
/// P->>+T: Updates? (offset = 5)
/// Note right of T: <= timeout
/// T->>-P: updates with ids [5]
///
/// C->>P: stop signal
///
/// P->>C: update(5)
///
/// C->>P: next
///
/// P->>T: *Acknolegment of update(5)*
/// T->>P: ok
///
/// P->>C: None
/// ```
///
/// [get_updates]: crate::requests::Requester::get_updates
pub fn polling<R>(
requester: R,
bot: R,
timeout: Option<Duration>,
limit: Option<u8>,
allowed_updates: Option<Vec<AllowedUpdate>>,
) -> impl UpdateListener<R::Err>
) -> impl UpdateListener<R::Err, StopToken = impl Send + StopToken>
where
R: Requester + Send + 'static,
<R as Requester>::GetUpdatesFaultTolerant: Send,
<R as Requester>::GetUpdates: Send,
{
struct State<B: Requester> {
bot: B,
@ -61,74 +143,57 @@ where
offset: i32,
flag: AsyncStopFlag,
token: AsyncStopToken,
force_stop: bool,
}
fn stream<B>(st: &mut State<B>) -> impl Stream<Item = Result<Update, B::Err>> + Send + '_
where
B: Requester + Send,
<B as Requester>::GetUpdatesFaultTolerant: Send,
<B as Requester>::GetUpdates: Send,
{
stream::unfold(st, move |state| async move {
let State { timeout, limit, allowed_updates, bot, offset, flag, .. } = &mut *state;
let State { timeout, limit, allowed_updates, bot, offset, flag, force_stop, .. } =
&mut *state;
if *force_stop {
return None;
}
if flag.is_stopped() {
let mut req = bot.get_updates_fault_tolerant();
req.payload_mut().0 = GetUpdates {
offset: Some(*offset),
timeout: Some(0),
limit: Some(1),
allowed_updates: allowed_updates.take(),
};
let mut req = bot.get_updates().offset(*offset).timeout(0).limit(1);
req.payload_mut().allowed_updates = allowed_updates.take();
return match req.send().await {
Ok(_) => None,
Err(err) => Some((Either::Left(stream::once(ready(Err(err)))), state)),
Err(err) => {
// Prevents infinite retries, see https://github.com/teloxide/teloxide/issues/496
*force_stop = true;
Some((Either::Left(stream::once(ready(Err(err)))), state))
}
};
}
let mut req = bot.get_updates_fault_tolerant();
req.payload_mut().0 = GetUpdates {
let mut req = bot.get_updates();
*req.payload_mut() = GetUpdates {
offset: Some(*offset),
timeout: *timeout,
limit: *limit,
allowed_updates: allowed_updates.take(),
};
let updates = match req.send().await {
Err(err) => return Some((Either::Left(stream::once(ready(Err(err)))), state)),
Ok(SemiparsedVec(updates)) => {
match req.send().await {
Ok(updates) => {
// Set offset to the last update's id + 1
if let Some(upd) = updates.last() {
let id: i32 = match upd {
Ok(ok) => ok.id,
Err((value, _)) => value["update_id"]
.as_i64()
.expect("The 'update_id' field must always exist in Update")
.try_into()
.expect("update_id must be i32"),
};
*offset = id + 1;
*offset = upd.id + 1;
}
for update in &updates {
if let Err((value, e)) = update {
log::error!(
"Cannot parse an update.\nError: {:?}\nValue: {}\n\
This is a bug in teloxide-core, please open an issue here: \
https://github.com/teloxide/teloxide-core/issues.",
e,
value
);
}
}
updates.into_iter().filter_map(Result::ok).map(Ok)
}
};
let updates = updates.into_iter().map(Ok);
Some((Either::Right(stream::iter(updates)), state))
}
Err(err) => Some((Either::Left(stream::once(ready(Err(err)))), state)),
}
})
.flatten()
}
@ -136,13 +201,14 @@ where
let (token, flag) = AsyncStopToken::new_pair();
let state = State {
bot: requester,
bot,
timeout: timeout.map(|t| t.as_secs().try_into().expect("timeout is too big")),
limit,
allowed_updates,
offset: 0,
flag,
token,
force_stop: false,
};
let stop_token = |st: &mut State<_>| st.token.clone();
@ -170,7 +236,7 @@ where
}
};
let is_webhook_setup = !webhook_info.url.is_empty();
let is_webhook_setup = webhook_info.url.is_some();
if is_webhook_setup {
if let Err(e) = requester.delete_webhook().send().await {
@ -188,6 +254,7 @@ fn polling_is_send() {
assert_send(&polling);
assert_send(&polling.as_stream());
assert_send(&polling.stop_token());
fn assert_send(_: &impl Send) {}
}

12
src/dispatching/update_listeners/stateful_listener.rs
View file

@ -25,25 +25,23 @@ pub struct StatefulListener<St, Assf, Sf, Hauf, Thf> {
/// The function used as [`AsUpdateStream::as_stream`].
///
/// Must be of type `for<'a> &'a mut St -> impl Stream + 'a` and callable by
/// `&mut`.
/// Must implement `for<'a> FnMut(&'a mut St) -> impl Stream + 'a`.
pub stream: Assf,
/// The function used as [`UpdateListener::stop_token`].
///
/// Must be of type `for<'a> &'a mut St -> impl StopToken`.
/// Must implement `FnMut(&mut St) -> impl StopToken`.
pub stop_token: Sf,
/// The function used as [`UpdateListener::hint_allowed_updates`].
///
/// Must be of type `for<'a, 'b> &'a mut St, &'b mut dyn Iterator<Item =
/// AllowedUpdate> -> ()`.
/// Must implement `FnMut(&mut St, &mut dyn Iterator<Item =
/// AllowedUpdate>)`.
pub hint_allowed_updates: Option<Hauf>,
/// The function used as [`UpdateListener::timeout_hint`].
///
/// Must be of type `for<'a> &'a St -> Option<Duration>` and callable by
/// `&`.
/// Must implement `Fn(&St) -> Option<Duration>`.
pub timeout_hint: Option<Thf>,
}

2
src/dispatching2/dispatcher.rs
View file

@ -132,7 +132,7 @@ where
pub async fn dispatch(&mut self)
where
R: Requester + Clone,
<R as Requester>::GetUpdatesFaultTolerant: Send,
<R as Requester>::GetUpdates: Send,
{
let listener = update_listeners::polling_default(self.bot.clone()).await;
let error_handler =

2
src/dispatching2/repls/commands_repl.rs
View file

@ -30,7 +30,7 @@ where
Cmd: BotCommand + Send + Sync + 'static,
H: Injectable<DependencyMap, Result<(), E>, Args> + Send + Sync + 'static,
R: Requester + Clone + Send + Sync + 'static,
<R as Requester>::GetUpdatesFaultTolerant: Send,
<R as Requester>::GetUpdates: Send,
E: Debug + Send + Sync + 'static,
{
let cloned_bot = bot.clone();

2
src/dispatching2/repls/repl.rs
View file

@ -26,7 +26,7 @@ where
Result<(), E>: OnError<E>,
E: Debug + Send + Sync + 'static,
R: Requester + Send + Sync + Clone + 'static,
<R as Requester>::GetUpdatesFaultTolerant: Send,
<R as Requester>::GetUpdates: Send,
{
let cloned_bot = bot.clone();
repl_with_listener(bot, handler, update_listeners::polling_default(cloned_bot).await).await;

5
src/lib.rs
View file

@ -44,8 +44,6 @@
html_logo_url = "https://github.com/teloxide/teloxide/raw/master/ICON.png",
html_favicon_url = "https://github.com/teloxide/teloxide/raw/master/ICON.png"
)]
#![allow(clippy::match_bool)]
#![forbid(unsafe_code)]
// We pass "--cfg docsrs" when building docs to add `This is supported on
// feature="..." only.`
//
@ -56,6 +54,9 @@
// $ RUSTFLAGS="--cfg dep_docsrs" RUSTDOCFLAGS="--cfg docsrs -Znormalize-docs" cargo +nightly doc --open --all-features
// ```
#![cfg_attr(all(docsrs, feature = "nightly"), feature(doc_cfg))]
#![forbid(unsafe_code)]
#![warn(rustdoc::broken_intra_doc_links)]
#![allow(clippy::match_bool)]
#![allow(clippy::redundant_pattern_matching)]
// https://github.com/rust-lang/rust-clippy/issues/7422
#![allow(clippy::nonstandard_macro_braces)]