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

Merge pull request #566 from teloxide/remove-old-dispatching

Browse source 
Remove old dispatching
This commit is contained in:
Hirrolot 2022-04-02 08:12:27 -07:00 committed by GitHub
commit fd18f9b2dd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
46 changed files with 689 additions and 2784 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
CHANGELOG.md
View file

@ -6,11 +6,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## unreleased
### Removed
- The old dispatching system and related stuff: `dispatching`, `utils::UpState`, `prelude`, `repls2`, `crate::{dialogues_repl, dialogues_repl_with_listener}`, and `#[teloxide(subtransition)]` [**BC**].
### Changed
- `UpdateListener::StopToken` is now always `Send`
- Rename `BotCommand` trait to `BotCommands`
- `BotCommands::descriptions` now returns `CommandDescriptions` instead of `String`
- Rename `dispatching2` => `dispatching` [**BC**].
- Rename `prelude2` => `prelude` [**BC**].
- Move `update_listeners`, `stop_token`, `IdleShutdownError`, and `ShutdownToken` from the old `dispatching` to the new `dispatching` (previously `dispatching2`).
- Replace `crate::{commands_repl, commands_repl_with_listener, repl, repl_with_listener}` with those of the new `dispatching` [**BC**].
- `UpdateListener::StopToken` is now always `Send` [**BC**].
- Rename `BotCommand` trait to `BotCommands` [**BC**].
- `BotCommands::descriptions` now returns `CommandDescriptions` instead of `String` [**BC**].
## 0.7.2 - 2022-03-23

23
Cargo.toml
View file

@ -13,16 +13,13 @@ exclude = ["media"]
[features]
# FIXME: remove "cache-me" that was added by mistake here
default = ["native-tls", "ctrlc_handler", "teloxide-core/default", "auto-send", "dispatching2", "cache-me"]
dispatching2 = ["dptree", "cache-me"]
default = ["native-tls", "ctrlc_handler", "teloxide-core/default", "auto-send", "cache-me"]
sqlite-storage = ["sqlx"]
redis-storage = ["redis"]
cbor-serializer = ["serde_cbor"]
bincode-serializer = ["bincode"]
frunk- = ["frunk"]
macros = ["teloxide-macros"]
ctrlc_handler = ["tokio/signal"]
@ -44,7 +41,6 @@ full = [
"redis-storage",
"cbor-serializer",
"bincode-serializer",
"frunk",
"macros",
"ctrlc_handler",
"teloxide-core/full",
@ -59,29 +55,26 @@ full = [
[dependencies]
teloxide-core = { version = "0.4", default-features = false }
#teloxide-macros = { version = "0.5.1", optional = true }
teloxide-macros = { git = "https://github.com/teloxide/teloxide-macros", rev = "7e000b9", optional = true }
teloxide-macros = { git = "https://github.com/teloxide/teloxide-macros.git", rev = "9e415b1", optional = true }
serde_json = "1.0"
serde = { version = "1.0", features = ["derive"] }
dptree = { version = "0.1.0", optional = true }
dptree = { version = "0.1.0" }
tokio = { version = "1.8", features = ["fs"] }
tokio-util = "0.6"
tokio-stream = "0.1"
flurry = "0.3"
log = "0.4"
bytes = "1.0"
mime = "0.3"
derive_more = "0.99"
thiserror = "1.0"
async-trait = "0.1"
futures = "0.3.15"
pin-project = "1.0"
serde_with_macros = "1.4"
aquamarine = "0.1.11"
sqlx = { version = "0.5", optional = true, default-features = false, features = [
"runtime-tokio-native-tls",
@ -91,22 +84,18 @@ sqlx = { version = "0.5", optional = true, default-features = false, features =
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"
rand = "0.8.3"
pretty_env_logger = "0.4.0"
once_cell = "1.9.0"
lazy_static = "1.4.0"
anyhow = "1.0.52"
serde = "1"
serde_json = "1"
tokio = { version = "1.8", features = ["fs", "rt-multi-thread", "macros"] }
warp = "0.3.0"
reqwest = "0.10.4"
chrono = "0.4"
tokio-stream = "0.1"
[package.metadata.docs.rs]
all-features = true
@ -150,5 +139,5 @@ name = "admin"
required-features = ["macros"]
[[example]]
name = "dispatching2_features"
name = "dispatching_features"
required-features = ["macros"]

32
README.md
View file

@ -88,7 +88,7 @@ This bot replies with a dice throw to each received message:
([Full](examples/dices.rs))
```rust,no_run
use teloxide::prelude2::*;
use teloxide::prelude::*;
#[tokio::main]
async fn main() {
@ -97,7 +97,7 @@ async fn main() {
let bot = Bot::from_env().auto_send();
teloxide::repls2::repl(bot, |message: Message, bot: AutoSend<Bot>| async move {
teloxide::repl(bot, |message: Message, bot: AutoSend<Bot>| async move {
bot.send_dice(message.chat.id).await?;
respond(())
})
@ -125,10 +125,20 @@ Commands are strongly typed and defined declaratively, similar to how we define
([Full](examples/simple_commands.rs))
```rust,no_run
use teloxide::{prelude2::*, utils::command::BotCommands};
use teloxide::{prelude::*, utils::command::BotCommands};
use std::error::Error;
#[tokio::main]
async fn main() {
pretty_env_logger::init();
log::info!("Starting simple_commands_bot...");
let bot = Bot::from_env().auto_send();
teloxide::commands_repl(bot, answer, Command::ty()).await;
}
#[derive(BotCommands, Clone)]
#[command(rename = "lowercase", description = "These commands are supported:")]
enum Command {
@ -146,7 +156,9 @@ async fn answer(
command: Command,
) -> Result<(), Box<dyn Error + Send + Sync>> {
match command {
Command::Help => bot.send_message(message.chat.id, Command::descriptions().to_string()).await?,
Command::Help => {
bot.send_message(message.chat.id, Command::descriptions().to_string()).await?
}
Command::Username(username) => {
bot.send_message(message.chat.id, format!("Your username is @{}.", username)).await?
}
@ -161,16 +173,6 @@ async fn answer(
Ok(())
}
#[tokio::main]
async fn main() {
pretty_env_logger::init();
log::info!("Starting simple_commands_bot...");
let bot = Bot::from_env().auto_send();
teloxide::repls2::commands_repl(bot, answer, Command::ty()).await;
}
```
<div align="center">
@ -190,7 +192,7 @@ Below is a bot that asks you three questions and then sends the answers back to
([Full](examples/dialogue.rs))
```rust,ignore
use teloxide::{dispatching2::dialogue::InMemStorage, macros::DialogueState, prelude2::*};
use teloxide::{dispatching::dialogue::InMemStorage, macros::DialogueState, prelude::*};
type MyDialogue = Dialogue<State, InMemStorage<State>>;

110
examples/admin.rs
View file

@ -1,7 +1,7 @@
use std::{error::Error, str::FromStr};
use chrono::Duration;
use teloxide::{prelude2::*, types::ChatPermissions, utils::command::BotCommands};
use teloxide::{prelude::*, types::ChatPermissions, utils::command::BotCommands};
// Derive BotCommands to parse text with a command into this enumeration.
//
@ -53,17 +53,35 @@ impl FromStr for UnitOfTime {
}
}
// Calculates time of user restriction.
fn calc_restrict_time(time: u64, unit: UnitOfTime) -> Duration {
match unit {
UnitOfTime::Hours => Duration::hours(time as i64),
UnitOfTime::Minutes => Duration::minutes(time as i64),
UnitOfTime::Seconds => Duration::seconds(time as i64),
}
#[tokio::main]
async fn main() {
pretty_env_logger::init();
log::info!("Starting admin_bot...");
let bot = teloxide::Bot::from_env().auto_send();
teloxide::commands_repl(bot, action, Command::ty()).await;
}
type Bot = AutoSend<teloxide::Bot>;
async fn action(
bot: Bot,
msg: Message,
command: Command,
) -> Result<(), Box<dyn Error + Send + Sync>> {
match command {
Command::Help => {
bot.send_message(msg.chat.id, Command::descriptions().to_string()).await?;
}
Command::Kick => kick_user(bot, msg).await?,
Command::Ban { time, unit } => ban_user(bot, msg, calc_restrict_time(time, unit)).await?,
Command::Mute { time, unit } => mute_user(bot, msg, calc_restrict_time(time, unit)).await?,
};
Ok(())
}
// Kick a user with a replied message.
async fn kick_user(bot: Bot, msg: Message) -> Result<(), Box<dyn Error + Send + Sync>> {
match msg.reply_to_message() {
@ -78,6 +96,29 @@ async fn kick_user(bot: Bot, msg: Message) -> Result<(), Box<dyn Error + Send +
Ok(())
}
// Ban a user with replied message.
async fn ban_user(
bot: Bot,
msg: Message,
time: Duration,
) -> Result<(), Box<dyn Error + Send + Sync>> {
match msg.reply_to_message() {
Some(replied) => {
bot.kick_chat_member(
msg.chat.id,
replied.from().expect("Must be MessageKind::Common").id,
)
.until_date(msg.date + time)
.await?;
}
None => {
bot.send_message(msg.chat.id, "Use this command in a reply to another message!")
.await?;
}
}
Ok(())
}
// Mute a user with a replied message.
async fn mute_user(
bot: Bot,
@ -102,52 +143,11 @@ async fn mute_user(
Ok(())
}
// Ban a user with replied message.
async fn ban_user(
bot: Bot,
msg: Message,
time: Duration,
) -> Result<(), Box<dyn Error + Send + Sync>> {
match msg.reply_to_message() {
Some(replied) => {
bot.kick_chat_member(
msg.chat.id,
replied.from().expect("Must be MessageKind::Common").id,
)
.until_date(msg.date + time)
.await?;
}
None => {
bot.send_message(msg.chat.id, "Use this command in a reply to another message!")
.await?;
}
// Calculates time of user restriction.
fn calc_restrict_time(time: u64, unit: UnitOfTime) -> Duration {
match unit {
UnitOfTime::Hours => Duration::hours(time as i64),
UnitOfTime::Minutes => Duration::minutes(time as i64),
UnitOfTime::Seconds => Duration::seconds(time as i64),
}
Ok(())
}
async fn action(
bot: Bot,
msg: Message,
command: Command,
) -> Result<(), Box<dyn Error + Send + Sync>> {
match command {
Command::Help => {
bot.send_message(msg.chat.id, Command::descriptions().to_string()).await?;
}
Command::Kick => kick_user(bot, msg).await?,
Command::Ban { time, unit } => ban_user(bot, msg, calc_restrict_time(time, unit)).await?,
Command::Mute { time, unit } => mute_user(bot, msg, calc_restrict_time(time, unit)).await?,
};
Ok(())
}
#[tokio::main]
async fn main() {
pretty_env_logger::init();
log::info!("Starting admin_bot...");
let bot = teloxide::Bot::from_env().auto_send();
teloxide::repls2::commands_repl(bot, action, Command::ty()).await;
}

37
examples/buttons.rs
View file

@ -1,7 +1,7 @@
use std::error::Error;
use teloxide::{
payloads::SendMessageSetters,
prelude2::*,
prelude::*,
types::{
InlineKeyboardButton, InlineKeyboardMarkup, InlineQueryResultArticle, InputMessageContent,
InputMessageContentText,
@ -18,6 +18,22 @@ enum Command {
Start,
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
pretty_env_logger::init();
log::info!("Starting bot...");
let bot = Bot::from_env().auto_send();
let handler = dptree::entry()
.branch(Update::filter_message().endpoint(message_handler))
.branch(Update::filter_callback_query().endpoint(callback_handler))
.branch(Update::filter_inline_query().endpoint(inline_query_handler));
Dispatcher::builder(bot, handler).build().setup_ctrlc_handler().dispatch().await;
Ok(())
}
/// Creates a keyboard made by buttons in a big column.
fn make_keyboard() -> InlineKeyboardMarkup {
let mut keyboard: Vec<Vec<InlineKeyboardButton>> = vec![];
@ -111,22 +127,3 @@ async fn callback_handler(
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
pretty_env_logger::init();
log::info!("Starting bot...");
let bot = Bot::from_env().auto_send();
let handler = dptree::entry()
.branch(Update::filter_message().endpoint(message_handler))
.branch(Update::filter_callback_query().endpoint(callback_handler))
.branch(Update::filter_inline_query().endpoint(inline_query_handler));
Dispatcher::builder(bot, handler).build().setup_ctrlc_handler().dispatch().await;
log::info!("Closing bot... Goodbye!");
Ok(())
}

4
examples/db_remember.rs
View file

@ -2,12 +2,12 @@
// Otherwise, the default is Sqlite.
use teloxide::{
dispatching2::dialogue::{
dispatching::dialogue::{
serializer::{Bincode, Json},
ErasedStorage, RedisStorage, SqliteStorage, Storage,
},
macros::DialogueState,
prelude2::*,
prelude::*,
types::Me,
utils::command::BotCommands,
};

17
examples/dialogue.rs
View file

@ -13,12 +13,13 @@
// Age: 223
// Location: Middle-earth
// ```
use teloxide::{dispatching2::dialogue::InMemStorage, macros::DialogueState, prelude2::*};
use teloxide::{dispatching::dialogue::InMemStorage, macros::DialogueState, prelude::*};
type MyDialogue = Dialogue<State, InMemStorage<State>>;
type HandlerResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
#[derive(DialogueState, Clone)]
#[handler_out(anyhow::Result<()>)]
#[handler_out(HandlerResult)]
pub enum State {
#[handler(handle_start)]
Start,
@ -59,11 +60,7 @@ async fn main() {
.await;
}
async fn handle_start(
bot: AutoSend<Bot>,
msg: Message,
dialogue: MyDialogue,
) -> anyhow::Result<()> {
async fn handle_start(bot: AutoSend<Bot>, msg: Message, dialogue: MyDialogue) -> HandlerResult {
bot.send_message(msg.chat.id, "Let's start! What's your full name?").await?;
dialogue.update(State::ReceiveFullName).await?;
Ok(())
@ -73,7 +70,7 @@ async fn handle_receive_full_name(
bot: AutoSend<Bot>,
msg: Message,
dialogue: MyDialogue,
) -> anyhow::Result<()> {
) -> HandlerResult {
match msg.text() {
Some(text) => {
bot.send_message(msg.chat.id, "How old are you?").await?;
@ -92,7 +89,7 @@ async fn handle_receive_age(
msg: Message,
dialogue: MyDialogue,
(full_name,): (String,), // Available from `State::ReceiveAge`.
) -> anyhow::Result<()> {
) -> HandlerResult {
match msg.text().map(|text| text.parse::<u8>()) {
Some(Ok(age)) => {
bot.send_message(msg.chat.id, "What's your location?").await?;
@ -111,7 +108,7 @@ async fn handle_receive_location(
msg: Message,
dialogue: MyDialogue,
(full_name, age): (String, u8), // Available from `State::ReceiveLocation`.
) -> anyhow::Result<()> {
) -> HandlerResult {
match msg.text() {
Some(location) => {
let message = format!("Full name: {}\nAge: {}\nLocation: {}", full_name, age, location);

4
examples/dices.rs
View file

@ -1,6 +1,6 @@
// This bot throws a dice on each incoming message.
use teloxide::prelude2::*;
use teloxide::prelude::*;
#[tokio::main]
async fn main() {
@ -9,7 +9,7 @@ async fn main() {
let bot = Bot::from_env().auto_send();
teloxide::repls2::repl(bot, |message: Message, bot: AutoSend<Bot>| async move {
teloxide::repl(bot, |message: Message, bot: AutoSend<Bot>| async move {
bot.send_dice(message.chat.id).await?;
respond(())
})

12
examples/dispatching2_features.rs → examples/dispatching_features.rs
View file

@ -1,20 +1,18 @@
// This example provide a quick overview of the new features in the
// `dispatching2` module.
// `dispatching` module.
use rand::Rng;
// You need to import `prelude2` because `prelude` contains items from the old
// dispatching system, which will be deprecated in the future.
use teloxide::{
prelude2::*,
types::{Dice, Me, Update},
prelude::*,
types::{Dice, Update},
utils::command::BotCommands,
};
#[tokio::main]
async fn main() {
pretty_env_logger::init();
log::info!("Starting dispatching2_features_bot...");
log::info!("Starting dispatching_features_bot...");
let bot = Bot::from_env().auto_send();
@ -124,7 +122,7 @@ async fn simple_commands_handler(
bot: AutoSend<Bot>,
cmd: SimpleCommand,
cfg: ConfigParameters,
me: Me,
me: teloxide::types::Me,
) -> Result<(), teloxide::RequestError> {
let text = match cmd {
SimpleCommand::Help => {

4
examples/heroku_ping_pong.rs
View file

@ -23,7 +23,7 @@ use teloxide::{
stop_token::AsyncStopToken,
update_listeners::{self, StatefulListener},
},
prelude2::*,
prelude::*,
types::Update,
};
@ -41,7 +41,7 @@ async fn main() {
let bot = Bot::from_env().auto_send();
teloxide::repls2::repl_with_listener(
teloxide::repl_with_listener(
bot.clone(),
|msg: Message, bot: AutoSend<Bot>| async move {
bot.send_message(msg.chat.id, "pong").await?;

2
examples/inline.rs
View file

@ -1,5 +1,5 @@
use teloxide::{
prelude2::*,
prelude::*,
types::{
InlineQueryResult, InlineQueryResultArticle, InputMessageContent, InputMessageContentText,
},

4
examples/ngrok_ping_pong.rs
View file

@ -6,7 +6,7 @@ use teloxide::{
stop_token::AsyncStopToken,
update_listeners::{self, StatefulListener},
},
prelude2::*,
prelude::*,
types::Update,
};
@ -24,7 +24,7 @@ async fn main() {
let bot = Bot::from_env().auto_send();
teloxide::repls2::repl_with_listener(
teloxide::repl_with_listener(
bot.clone(),
|msg: Message, bot: AutoSend<Bot>| async move {
bot.send_message(msg.chat.id, "pong").await?;

2
examples/shared_state.rs
View file

@ -3,7 +3,7 @@
use std::sync::atomic::{AtomicU64, Ordering};
use once_cell::sync::Lazy;
use teloxide::prelude2::*;
use teloxide::prelude::*;
static MESSAGES_TOTAL: Lazy<AtomicU64> = Lazy::new(AtomicU64::default);

22
examples/simple_commands.rs
View file

@ -1,7 +1,17 @@
use teloxide::{prelude2::*, utils::command::BotCommands};
use teloxide::{prelude::*, utils::command::BotCommands};
use std::error::Error;
#[tokio::main]
async fn main() {
pretty_env_logger::init();
log::info!("Starting simple_commands_bot...");
let bot = Bot::from_env().auto_send();
teloxide::commands_repl(bot, answer, Command::ty()).await;
}
#[derive(BotCommands, Clone)]
#[command(rename = "lowercase", description = "These commands are supported:")]
enum Command {
@ -36,13 +46,3 @@ async fn answer(
Ok(())
}
#[tokio::main]
async fn main() {
pretty_env_logger::init();
log::info!("Starting simple_commands_bot...");
let bot = Bot::from_env().auto_send();
teloxide::repls2::commands_repl(bot, answer, Command::ty()).await;
}

291
src/dispatching/dialogue/dialogue_dispatcher.rs
View file

@ -1,291 +0,0 @@
use crate::dispatching::{
dialogue::{
DialogueDispatcherHandler, DialogueStage, DialogueWithCx, GetChatId, InMemStorage, Storage,
},
DispatcherHandler, UpdateWithCx,
};
use std::{fmt::Debug, marker::PhantomData};
use futures::{future::BoxFuture, FutureExt, StreamExt};
use tokio::sync::mpsc;
use crate::dispatching::dialogue::InMemStorageError;
use flurry::HashMap;
use std::sync::{Arc, Mutex};
use teloxide_core::requests::Requester;
use tokio_stream::wrappers::UnboundedReceiverStream;
/// A dispatcher of dialogues.
///
/// Note that it implements [`DispatcherHandler`], so you can just put an
/// instance of this dispatcher into the [`Dispatcher`]'s methods.
///
/// Note that when the storage methods [`Storage::remove_dialogue`] and
/// [`Storage::update_dialogue`] are failed, the errors are logged, but a result
/// from [`Storage::get_dialogue`] is provided to a user handler as-is so you
/// can respond to a concrete user with an error description.
///
/// See the [module-level documentation](crate::dispatching::dialogue) for the
/// design overview.
///
/// [`Dispatcher`]: crate::dispatching::Dispatcher
/// [`DispatcherHandler`]: crate::dispatching::DispatcherHandler
#[deprecated(note = "Use dispatching2 instead")]
pub struct DialogueDispatcher<R, D, S, H, Upd> {
storage: Arc<S>,
handler: Arc<H>,
_phantom: PhantomData<Mutex<D>>,
/// A lock-free map to handle updates from the same chat sequentially, but
/// concurrently from different chats.
///
/// A value is the TX part of an unbounded asynchronous MPSC channel. A
/// handler that executes updates from the same chat ID sequentially
/// handles the RX part.
senders: Arc<HashMap<i64, mpsc::UnboundedSender<UpdateWithCx<R, Upd>>>>,
}
impl<R, D, H, Upd> DialogueDispatcher<R, D, InMemStorage<D>, H, Upd>
where
H: DialogueDispatcherHandler<R, Upd, D, InMemStorageError> + Send + Sync + 'static,
Upd: GetChatId + Send + 'static,
D: Default + Send + 'static,
{
/// Creates a dispatcher with the specified `handler` and [`InMemStorage`]
/// (a default storage).
///
/// [`InMemStorage`]: crate::dispatching::dialogue::InMemStorage
#[must_use]
pub fn new(handler: H) -> Self {
Self {
storage: InMemStorage::new(),
handler: Arc::new(handler),
senders: Arc::new(HashMap::new()),
_phantom: PhantomData,
}
}
}
impl<R, D, S, H, Upd> DialogueDispatcher<R, D, S, H, Upd>
where
H: DialogueDispatcherHandler<R, Upd, D, S::Error> + Send + Sync + 'static,
Upd: GetChatId + Send + 'static,
D: Default + Send + 'static,
S: Storage<D> + Send + Sync + 'static,
S::Error: Debug + Send + 'static,
{
/// Creates a dispatcher with the specified `handler` and `storage`.
#[must_use]
pub fn with_storage(handler: H, storage: Arc<S>) -> Self {
Self {
storage,
handler: Arc::new(handler),
senders: Arc::new(HashMap::new()),
_phantom: PhantomData,
}
}
#[must_use]
fn new_tx(&self) -> mpsc::UnboundedSender<UpdateWithCx<R, Upd>>
where
R: Requester + Send + 'static,
{
let (tx, rx) = mpsc::unbounded_channel();
let storage = Arc::clone(&self.storage);
let handler = Arc::clone(&self.handler);
let senders = Arc::clone(&self.senders);
tokio::spawn(UnboundedReceiverStream::new(rx).for_each(move |cx: UpdateWithCx<R, Upd>| {
let storage = Arc::clone(&storage);
let handler = Arc::clone(&handler);
let senders = Arc::clone(&senders);
async move {
let chat_id = cx.update.chat_id();
let dialogue =
Arc::clone(&storage).get_dialogue(chat_id).await.map(Option::unwrap_or_default);
match handler.handle(DialogueWithCx { cx, dialogue }).await {
DialogueStage::Next(new_dialogue) => {
if let Err(e) = storage.update_dialogue(chat_id, new_dialogue).await {
log::error!("Storage::update_dialogue failed: {:?}", e);
}
}
DialogueStage::Exit => {
// On the next .poll() call, the spawned future will
// return Poll::Ready, because we are dropping the
// sender right here:
senders.pin().remove(&chat_id);
if let Err(e) = storage.remove_dialogue(chat_id).await {
log::error!("Storage::remove_dialogue failed: {:?}", e);
}
}
}
}
}));
tx
}
}
impl<R, D, S, H, Upd> DispatcherHandler<R, Upd> for DialogueDispatcher<R, D, S, H, Upd>
where
H: DialogueDispatcherHandler<R, Upd, D, S::Error> + Send + Sync + 'static,
Upd: GetChatId + Send + 'static,
D: Default + Send + 'static,
S: Storage<D> + Send + Sync + 'static,
S::Error: Debug + Send + 'static,
R: Requester + Send,
{
fn handle(
self,
updates: mpsc::UnboundedReceiver<UpdateWithCx<R, Upd>>,
) -> BoxFuture<'static, ()>
where
UpdateWithCx<R, Upd>: 'static,
{
let this = Arc::new(self);
UnboundedReceiverStream::new(updates)
.for_each(move |cx| {
let this = Arc::clone(&this);
let chat_id = cx.update.chat_id();
match this.senders.pin().get(&chat_id) {
// An old dialogue
Some(tx) => {
assert!(
tx.send(cx).is_ok(),
"We are not dropping a receiver or call .close() on it"
);
}
None => {
let tx = this.new_tx();
assert!(
tx.send(cx).is_ok(),
"We are not dropping a receiver or call .close() on it"
);
this.senders.pin().insert(chat_id, tx);
}
}
async {}
})
.boxed()
}
}
#[cfg(test)]
mod tests {
use super::*;
use futures::{stream, StreamExt};
use lazy_static::lazy_static;
use teloxide_core::Bot;
use tokio::{
sync::{mpsc, Mutex},
time::Duration,
};
#[tokio::test]
#[allow(deprecated)]
async fn updates_from_same_chat_executed_sequentially() {
#[derive(Debug)]
struct MyUpdate {
chat_id: i64,
unique_number: u32,
}
impl MyUpdate {
fn new(chat_id: i64, unique_number: u32) -> Self {
Self { chat_id, unique_number }
}
}
impl GetChatId for MyUpdate {
fn chat_id(&self) -> i64 {
self.chat_id
}
}
lazy_static! {
static ref SEQ1: Mutex<Vec<u32>> = Mutex::new(Vec::new());
static ref SEQ2: Mutex<Vec<u32>> = Mutex::new(Vec::new());
static ref SEQ3: Mutex<Vec<u32>> = Mutex::new(Vec::new());
}
let dispatcher = DialogueDispatcher::new(
|cx: DialogueWithCx<Bot, MyUpdate, (), InMemStorageError>| async move {
tokio::time::sleep(Duration::from_millis(300)).await;
match cx.cx.update {
MyUpdate { chat_id: 1, unique_number } => {
SEQ1.lock().await.push(unique_number);
}
MyUpdate { chat_id: 2, unique_number } => {
SEQ2.lock().await.push(unique_number);
}
MyUpdate { chat_id: 3, unique_number } => {
SEQ3.lock().await.push(unique_number);
}
_ => unreachable!(),
}
DialogueStage::Next(())
},
);
let updates = stream::iter(
vec![
MyUpdate::new(1, 174),
MyUpdate::new(1, 125),
MyUpdate::new(2, 411),
MyUpdate::new(1, 2),
MyUpdate::new(2, 515),
MyUpdate::new(2, 623),
MyUpdate::new(1, 193),
MyUpdate::new(1, 104),
MyUpdate::new(2, 2222),
MyUpdate::new(2, 737),
MyUpdate::new(3, 72782),
MyUpdate::new(3, 2737),
MyUpdate::new(1, 7),
MyUpdate::new(1, 7778),
MyUpdate::new(3, 5475),
MyUpdate::new(3, 1096),
MyUpdate::new(3, 872),
MyUpdate::new(2, 10),
MyUpdate::new(2, 55456),
MyUpdate::new(3, 5665),
MyUpdate::new(3, 1611),
]
.into_iter()
.map(|update| UpdateWithCx { update, requester: Bot::new("Doesn't matter here") })
.collect::<Vec<UpdateWithCx<Bot, MyUpdate>>>(),
);
let (tx, rx) = mpsc::unbounded_channel();
updates
.for_each(move |update| {
let tx = tx.clone();
async move {
assert!(tx.send(update).is_ok(), "tx.send(update) failed");
}
})
.await;
dispatcher.handle(rx).await;
// Wait until our futures to be finished.
tokio::time::sleep(Duration::from_millis(3000)).await;
assert_eq!(*SEQ1.lock().await, vec![174, 125, 2, 193, 104, 7, 7778]);
assert_eq!(*SEQ2.lock().await, vec![411, 515, 623, 2222, 737, 10, 55456]);
assert_eq!(*SEQ3.lock().await, vec![72782, 2737, 5475, 1096, 872, 5665, 1611]);
}
}

41
src/dispatching/dialogue/dialogue_dispatcher_handler.rs
View file

@ -1,41 +0,0 @@
use crate::dispatching::dialogue::{DialogueStage, DialogueWithCx};
use futures::future::BoxFuture;
use std::{future::Future, sync::Arc};
/// An asynchronous handler of an update used in [`DialogueDispatcher`].
///
/// See [the module-level documentation for the design
/// overview](crate::dispatching::dialogue).
///
/// [`DialogueDispatcher`]: crate::dispatching::dialogue::DialogueDispatcher
#[deprecated(note = "Use dispatching2 instead")]
pub trait DialogueDispatcherHandler<R, Upd, D, E> {
#[must_use]
fn handle(
self: Arc<Self>,
cx: DialogueWithCx<R, Upd, D, E>,
) -> BoxFuture<'static, DialogueStage<D>>
where
DialogueWithCx<R, Upd, D, E>: Send + 'static,
R: Send,
Upd: Send,
D: Send,
E: Send;
}
impl<R, Upd, D, E, F, Fut> DialogueDispatcherHandler<R, Upd, D, E> for F
where
F: Fn(DialogueWithCx<R, Upd, D, E>) -> Fut + Send + Sync + 'static,
Fut: Future<Output = DialogueStage<D>> + Send + 'static,
{
fn handle(self: Arc<Self>, cx: DialogueWithCx<R, Upd, D, E>) -> BoxFuture<'static, Fut::Output>
where
DialogueWithCx<R, Upd, D, E>: Send + 'static,
R: Send,
Upd: Send,
D: Send,
E: Send,
{
Box::pin(async move { self(cx).await })
}
}

40
src/dispatching/dialogue/dialogue_stage.rs
View file

@ -1,40 +0,0 @@
use crate::dispatching::dialogue::TransitionOut;
/// Continue or terminate a dialogue.
///
/// See [the module-level documentation for the design
/// overview](crate::dispatching::dialogue).
#[derive(Debug, Copy, Clone, Eq, Hash, PartialEq)]
#[deprecated(note = "Use dispatching2 instead")]
pub enum DialogueStage<D> {
Next(D),
Exit,
}
/// Returns a new dialogue state.
///
/// Note the `Dialogue: From<State>` constraint. It means that you don't need to
/// pass `Dialogue` -- you can just pass one of it's states. [`From`] can be
/// conveniently derived by [derive-more].
///
/// See [the module-level documentation for the design
/// overview](crate::dispatching::dialogue).
///
/// [`From`]: std::convert::From
/// [derive-more]: https://crates.io/crates/derive_more
#[deprecated(note = "Use dispatching2 instead")]
pub fn next<Dialogue, State, E>(new_state: State) -> TransitionOut<Dialogue, E>
where
Dialogue: From<State>,
{
Ok(DialogueStage::Next(Dialogue::from(new_state)))
}
/// Exits a dialogue.
///
/// See [the module-level documentation for the design
/// overview](crate::dispatching::dialogue).
#[deprecated(note = "Use dispatching2 instead")]
pub fn exit<D, E>() -> TransitionOut<D, E> {
Ok(DialogueStage::Exit)
}

33
src/dispatching/dialogue/dialogue_with_cx.rs
View file

@ -1,33 +0,0 @@
use crate::dispatching::{dialogue::GetChatId, UpdateWithCx};
use std::fmt::Debug;
use teloxide_core::requests::Requester;
/// A context of a [`DialogueDispatcher`]'s message handler.
///
/// See [the module-level documentation for the design
/// overview](crate::dispatching::dialogue).
///
/// [`DialogueDispatcher`]: crate::dispatching::dialogue::DialogueDispatcher
#[derive(Debug)]
#[deprecated(note = "Use dispatching2 instead")]
pub struct DialogueWithCx<R, Upd, D, E> {
pub cx: UpdateWithCx<R, Upd>,
pub dialogue: Result<D, E>,
}
impl<Upd, R, D, E> DialogueWithCx<R, Upd, D, E> {
/// Creates a new instance with the provided fields.
pub fn new(cx: UpdateWithCx<R, Upd>, dialogue: D) -> Self {
Self { cx, dialogue: Ok(dialogue) }
}
}
impl<Upd, R, D, E> GetChatId for DialogueWithCx<R, Upd, D, E>
where
Upd: GetChatId,
R: Requester,
{
fn chat_id(&self) -> i64 {
self.cx.update.chat_id()
}
}

16
src/dispatching/dialogue/get_chat_id.rs
View file

@ -1,14 +1,20 @@
use crate::types::CallbackQuery;
use teloxide_core::types::Message;
/// Something that has a chat ID.
#[deprecated(note = "Use dispatching2 instead")]
/// Something that may has a chat ID.
pub trait GetChatId {
#[must_use]
fn chat_id(&self) -> i64;
fn chat_id(&self) -> Option<i64>;
}
impl GetChatId for Message {
fn chat_id(&self) -> i64 {
self.chat.id
fn chat_id(&self) -> Option<i64> {
Some(self.chat.id)
}
}
impl GetChatId for CallbackQuery {
fn chat_id(&self) -> Option<i64> {
self.message.as_ref().map(|mes| mes.chat.id)
}
}

311
src/dispatching/dialogue/mod.rs
View file

@ -1,172 +1,181 @@
//! Dealing with dialogues.
//! Support for user dialogues.
//!
//! There are three main components:
//! The main type is (surprise!) [`Dialogue`]. Under the hood, it is just a
//! wrapper over [`Storage`] and a chat ID. All it does is provides convenient
//! method for manipulating the dialogue state. [`Storage`] is where all
//! dialogue states are stored; it can be either [`InMemStorage`], which is a
//! simple hash map, or database wrappers such as [`SqliteStorage`]. In the
//! latter case, your dialogues are _persistent_, meaning that you can safely
//! restart your bot and all dialogues will remain in the database -- this is a
//! preferred method for production bots.
//!
//! 1. Your type `D` (typically an enumeration), implementing [`Transition`].
//! It is essentially a [FSM]: its variants are possible dialogue states and
//! [`Transition::react`] is a transition function.
//! [`examples/dialogue.rs`] clearly demonstrates the typical usage of
//! dialogues. Your dialogue state can be represented as an enumeration:
//!
//! 2. State types, forming `D`. They implement [`Subtransition`].
//! ```ignore
//! #[derive(DialogueState, Clone)]
//! #[handler_out(anyhow::Result<()>)]
//! pub enum State {
//! #[handler(handle_start)]
//! Start,
//!
//! 2. [`Storage<D>`], which encapsulates all the dialogues.
//! #[handler(handle_receive_full_name)]
//! ReceiveFullName,
//!
//! 3. [`DialogueDispatcher`], which encapsulates your handler, [`Storage<D>`],
//! and implements [`DispatcherHandler`].
//! #[handler(handle_receive_age)]
//! ReceiveAge { full_name: String },
//!
//! You pass [`DialogueDispatcher`] into [`Dispatcher`]. Every time
//! [`Dispatcher`] sees an incoming input, it is transferred into
//! [`DialogueDispatcher`], and the following steps are executed:
//!
//! 1. If a storage doesn't contain a dialogue from this chat, supply
//! `D::default()` into you handler, otherwise, supply the saved dialogue
//! from this chat.
//! 2. If a handler has returned [`DialogueStage::Exit`], remove the dialogue
//! from the storage, otherwise ([`DialogueStage::Next`]) force the storage to
//! update the dialogue.
//!
//! To avoid boilerplate, teloxide exports these convenient things: the [`next`]
//! and [`exit`] functions, and `#[derive(BotDialogue)]` with
//! `#[teloxide(subtransition)]`. Here's how your dialogues management code
//! skeleton should look like:
//!
//! ```no_run
//! # #[cfg(feature = "macros")] {
//! use std::convert::Infallible;
//!
//! use teloxide::{
//! dispatching::dialogue::{InMemStorageError, Transition},
//! prelude::*,
//! teloxide, RequestError,
//! };
//!
//! #[derive(Clone)]
//! struct _1State;
//! #[derive(Clone)]
//! struct _2State;
//! #[derive(Clone)]
//! struct _3State;
//!
//! type Out = TransitionOut<D, RequestError>;
//!
//! #[teloxide(subtransition)]
//! async fn _1_transition(_state: _1State, _cx: TransitionIn<AutoSend<Bot>>) -> Out {
//! todo!()
//! #[handler(handle_receive_location)]
//! ReceiveLocation { full_name: String, age: u8 },
//! }
//!
//! #[teloxide(subtransition)]
//! async fn _2_transition(_state: _2State, _cx: TransitionIn<AutoSend<Bot>>) -> Out {
//! todo!()
//! }
//!
//! #[teloxide(subtransition)]
//! async fn _3_transition(_state: _3State, _cx: TransitionIn<AutoSend<Bot>>) -> Out {
//! todo!()
//! }
//!
//! #[derive(Clone, Transition)]
//! enum D {
//! _1(_1State),
//! _2(_2State),
//! _3(_3State),
//! }
//!
//! impl Default for D {
//! fn default() -> Self {
//! Self::_1(_1State)
//! }
//! }
//!
//! type In = DialogueWithCx<AutoSend<Bot>, Message, D, InMemStorageError>;
//!
//! #[tokio::main]
//! async fn main() {
//! pretty_env_logger::init();
//! log::info!("Starting dialogue_bot!");
//!
//! let bot = Bot::from_env().auto_send();
//!
//! Dispatcher::new(bot)
//! .messages_handler(DialogueDispatcher::new(
//! |DialogueWithCx { cx, dialogue }: In| async move {
//! // No panic because of std::convert::Infallible.
//! let dialogue = dialogue.unwrap();
//! dialogue
//! // Instead of () you can pass an arbitrary value, see below.
//! .react(cx, ())
//! .await
//! .expect("Something wrong with the bot!")
//! },
//! ))
//! .dispatch()
//! .await;
//! }
//! # }
//! ```
//!
//! - `#[teloxide(subtransition)]` implements [`Subtransition`] for the first
//! argument of a function.
//! - `#[derive(Transition)]` implements [`Transition`] for `D`, if all the
//! variants implement [`Subtransition`].
//! Each state is associated with its respective handler: e.g., when a dialogue
//! state is `ReceiveAge`, `handle_receive_age` is invoked:
//!
//! `()` in `.react(cx, ())` is an arbitrary value, which you can pass into
//! Subtransitions. Just append `ans: T` to the parameters of the
//! Subtransitions to pass a differen type.
//! ```ignore
//! async fn handle_receive_age(
//! bot: AutoSend<Bot>,
//! msg: Message,
//! dialogue: MyDialogue,
//! (full_name,): (String,), // Available from `State::ReceiveAge`.
//! ) -> anyhow::Result<()> {
//! match msg.text().map(|text| text.parse::<u8>()) {
//! Some(Ok(age)) => {
//! bot.send_message(msg.chat.id, "What's your location?").await?;
//! dialogue.update(State::ReceiveLocation { full_name, age }).await?;
//! }
//! _ => {
//! bot.send_message(msg.chat.id, "Send me a number.").await?;
//! }
//! }
//!
//! See [examples/dialogue_bot] as a real example.
//! Ok(())
//! }
//! ```
//!
//! [`Transition`]: crate::dispatching::dialogue::Transition
//! [`Subtransition`]: crate::dispatching::dialogue::Subtransition
//! [`Transition::react`]:
//! crate::dispatching::dialogue::Transition::react
//! [FSM]: https://en.wikipedia.org/wiki/Finite-state_machine
//! Variant's fields are passed to state handlers as tuples: `(full_name,):
//! (String,)`. Using [`Dialogue::update`], you can update the dialogue with a
//! new state, in our case -- `State::ReceiveLocation { full_name, age }`. To
//! exit the dialogue, just call [`Dialogue::exit`] and it will be removed from
//! the inner storage:
//!
//! [`Storage<D>`]: crate::dispatching::dialogue::Storage
//! ```ignore
//! async fn handle_receive_location(
//! bot: AutoSend<Bot>,
//! msg: Message,
//! dialogue: MyDialogue,
//! (full_name, age): (String, u8), // Available from `State::ReceiveLocation`.
//! ) -> anyhow::Result<()> {
//! match msg.text() {
//! Some(location) => {
//! let message =
//! format!("Full name: {}\nAge: {}\nLocation: {}", full_name, age, location);
//! bot.send_message(msg.chat.id, message).await?;
//! dialogue.exit().await?;
//! }
//! None => {
//! bot.send_message(msg.chat.id, "Send me a text message.").await?;
//! }
//! }
//!
//! [`DialogueStage<D>`]: crate::dispatching::dialogue::DialogueStage
//! [`DialogueDispatcher`]: crate::dispatching::dialogue::DialogueDispatcher
//! Ok(())
//! }
//! ```
//!
//! [`DialogueStage::Exit`]:
//! crate::dispatching::dialogue::DialogueStage::Exit
//! [`DialogueStage::Next`]: crate::dispatching::dialogue::DialogueStage::Next
//!
//! [`up!`]: crate::up
//! [`next`]: crate::dispatching::dialogue::next
//! [`exit`]: crate::dispatching::dialogue::exit
//!
//! [`DispatcherHandler`]: crate::dispatching::DispatcherHandler
//! [`Dispatcher`]: crate::dispatching::Dispatcher
//! [`Dispatcher::messages_handler`]:
//! crate::dispatching::Dispatcher::messages_handler
//! [`UpdateKind::Message(message)`]: crate::types::UpdateKind::Message
//!
//! [examples/dialogue_bot]: https://github.com/teloxide/teloxide/tree/master/examples/dialogue_bot
#![allow(clippy::type_complexity)]
mod dialogue_dispatcher;
mod dialogue_dispatcher_handler;
mod dialogue_stage;
mod dialogue_with_cx;
mod get_chat_id;
mod storage;
mod transition;
pub use dialogue_dispatcher::DialogueDispatcher;
pub use dialogue_dispatcher_handler::DialogueDispatcherHandler;
pub use dialogue_stage::{exit, next, DialogueStage};
pub use dialogue_with_cx::DialogueWithCx;
pub use get_chat_id::GetChatId;
pub use transition::{
Subtransition, SubtransitionOutputType, Transition, TransitionIn, TransitionOut,
};
#[cfg(feature = "macros")]
pub use teloxide_macros::Transition;
//! [`examples/dialogue.rs`]: https://github.com/teloxide/teloxide/blob/master/examples/dialogue.rs
#[cfg(feature = "redis-storage")]
pub use storage::{RedisStorage, RedisStorageError};
pub use crate::dispatching::dialogue::{RedisStorage, RedisStorageError};
#[cfg(feature = "sqlite-storage")]
pub use storage::{SqliteStorage, SqliteStorageError};
pub use crate::dispatching::dialogue::{SqliteStorage, SqliteStorageError};
pub use get_chat_id::GetChatId;
pub use storage::*;
use std::{marker::PhantomData, sync::Arc};
mod get_chat_id;
mod storage;
/// A handle for controlling dialogue state.
#[derive(Debug)]
pub struct Dialogue<D, S>
where
S: ?Sized,
{
storage: Arc<S>,
chat_id: i64,
_phantom: PhantomData<D>,
}
// `#[derive]` requires generics to implement `Clone`, but `S` is wrapped around
// `Arc`, and `D` is wrapped around PhantomData.
impl<D, S> Clone for Dialogue<D, S>
where
S: ?Sized,
{
fn clone(&self) -> Self {
Dialogue { storage: self.storage.clone(), chat_id: self.chat_id, _phantom: PhantomData }
}
}
impl<D, S> Dialogue<D, S>
where
D: Send + 'static,
S: Storage<D> + ?Sized,
{
/// Constructs a new dialogue with `storage` (where dialogues are stored)
/// and `chat_id` of a current dialogue.
pub fn new(storage: Arc<S>, chat_id: i64) -> Self {
Self { storage, chat_id, _phantom: PhantomData }
}
/// Retrieves the current state of the dialogue or `None` if there is no
/// dialogue.
pub async fn get(&self) -> Result<Option<D>, S::Error> {
self.storage.clone().get_dialogue(self.chat_id).await
}
/// Like [`Dialogue::get`] but returns a default value if there is no
/// dialogue.
pub async fn get_or_default(&self) -> Result<D, S::Error>
where
D: Default,
{
match self.get().await? {
Some(d) => Ok(d),
None => {
self.storage.clone().update_dialogue(self.chat_id, D::default()).await?;
Ok(D::default())
}
}
}
/// Updates the dialogue state.
///
/// The dialogue type `D` must implement `From<State>` to allow implicit
/// conversion from `State` to `D`.
pub async fn update<State>(&self, state: State) -> Result<(), S::Error>
where
D: From<State>,
{
let new_dialogue = state.into();
self.storage.clone().update_dialogue(self.chat_id, new_dialogue).await?;
Ok(())
}
/// Updates the dialogue with a default value.
pub async fn reset(&self) -> Result<(), S::Error>
where
D: Default,
{
self.update(D::default()).await
}
/// Removes the dialogue from the storage provided to [`Dialogue::new`].
pub async fn exit(&self) -> Result<(), S::Error> {
self.storage.clone().remove_dialogue(self.chat_id).await
}
}

67
src/dispatching/dialogue/transition.rs
View file

@ -1,67 +0,0 @@
use crate::dispatching::{dialogue::DialogueStage, UpdateWithCx};
use futures::future::BoxFuture;
use teloxide_core::types::Message;
/// Represents a transition function of a dialogue FSM.
#[deprecated(note = "Use dispatching2 instead")]
pub trait Transition: Sized {
type Aux;
type Error;
type Requester;
/// Turns itself into another state, depending on the input message.
///
/// `aux` will be passed to each subtransition function.
fn react(
self,
cx: TransitionIn<Self::Requester>,
aux: Self::Aux,
) -> BoxFuture<'static, TransitionOut<Self, Self::Error>>;
}
/// Like [`Transition`], but from `StateN` -> `Dialogue`.
///
/// [`Transition`]: crate::dispatching::dialogue::Transition
#[deprecated(note = "Use dispatching2 instead")]
pub trait Subtransition
where
Self::Dialogue: Transition<Aux = Self::Aux>,
{
type Aux;
type Dialogue;
type Error;
type Requester;
/// Turns itself into another state, depending on the input message.
///
/// `aux` is something that is provided by the call side, for example,
/// message's text.
fn react(
self,
cx: TransitionIn<Self::Requester>,
aux: Self::Aux,
) -> BoxFuture<'static, TransitionOut<Self::Dialogue, Self::Error>>;
}
/// A type returned from a FSM subtransition function.
///
/// Now it is used only inside `#[teloxide(subtransition)]` for type inference.
#[doc(hidden)]
#[deprecated(note = "Use dispatching2 instead")]
pub trait SubtransitionOutputType {
type Output;
type Error;
}
impl<D, E> SubtransitionOutputType for TransitionOut<D, E> {
type Output = D;
type Error = E;
}
/// An input passed into a FSM (sub)transition function.
#[deprecated(note = "Use dispatching2 instead")]
pub type TransitionIn<R> = UpdateWithCx<R, Message>;
/// A type returned from a FSM (sub)transition function.
#[deprecated(note = "Use dispatching2 instead")]
pub type TransitionOut<D, E = crate::RequestError> = Result<DialogueStage<D>, E>;

642
src/dispatching/dispatcher.rs
View file

@ -1,263 +1,158 @@
use std::{fmt::Debug, sync::Arc};
use crate::{
adaptors::CacheMe,
dispatching::{
stop_token::StopToken,
update_listeners::{self, UpdateListener},
DispatcherHandler, UpdateWithCx,
stop_token::StopToken, update_listeners, update_listeners::UpdateListener, ShutdownToken,
},
error_handlers::{ErrorHandler, LoggingErrorHandler},
requests::Requester,
types::{AllowedUpdate, Update},
utils::shutdown_token::shutdown_check_timeout_for,
};
use futures::{stream::FuturesUnordered, StreamExt};
use dptree::di::{DependencyMap, DependencySupplier};
use futures::{future::BoxFuture, StreamExt};
use std::{collections::HashSet, fmt::Debug, ops::ControlFlow, sync::Arc};
use teloxide_core::{
requests::Requester,
types::{
AllowedUpdate, CallbackQuery, ChatJoinRequest, ChatMemberUpdated, ChosenInlineResult,
InlineQuery, Message, Poll, PollAnswer, PreCheckoutQuery, ShippingQuery, Update,
UpdateKind,
},
requests::{Request, RequesterExt},
types::UpdateKind,
};
use tokio::{sync::mpsc, task::JoinHandle, time::timeout};
use tokio::time::timeout;
use crate::utils::shutdown_token::ShutdownToken;
use std::future::Future;
type Tx<Upd, R> = Option<mpsc::UnboundedSender<UpdateWithCx<Upd, R>>>;
/// The builder for [`Dispatcher`].
pub struct DispatcherBuilder<R, Err> {
bot: R,
dependencies: DependencyMap,
handler: UpdateHandler<Err>,
default_handler: DefaultHandler,
error_handler: Arc<dyn ErrorHandler<Err> + Send + Sync>,
}
/// One dispatcher to rule them all.
///
/// See the [module-level documentation](crate::dispatching) for the design
/// overview.
#[deprecated(note = "Use dispatching2 instead")]
pub struct Dispatcher<R> {
requester: R,
impl<R, Err> DispatcherBuilder<R, Err>
where
R: Clone + Requester + Clone + Send + Sync + 'static,
Err: Debug + Send + Sync + 'static,
{
/// Specifies a handler that will be called for an unhandled update.
///
/// By default, it is a mere [`log::warn`].
#[must_use]
pub fn default_handler<H, Fut>(self, handler: H) -> Self
where
H: Fn(Arc<Update>) -> Fut + Send + Sync + 'static,
Fut: Future<Output = ()> + Send + 'static,
{
let handler = Arc::new(handler);
messages_queue: Tx<R, Message>,
edited_messages_queue: Tx<R, Message>,
channel_posts_queue: Tx<R, Message>,
edited_channel_posts_queue: Tx<R, Message>,
inline_queries_queue: Tx<R, InlineQuery>,
chosen_inline_results_queue: Tx<R, ChosenInlineResult>,
callback_queries_queue: Tx<R, CallbackQuery>,
shipping_queries_queue: Tx<R, ShippingQuery>,
pre_checkout_queries_queue: Tx<R, PreCheckoutQuery>,
polls_queue: Tx<R, Poll>,
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>,
Self {
default_handler: Box::new(move |upd| {
let handler = Arc::clone(&handler);
Box::pin(handler(upd))
}),
..self
}
}
running_handlers: FuturesUnordered<JoinHandle<()>>,
/// Specifies a handler that will be called on a handler error.
///
/// By default, it is [`LoggingErrorHandler`].
#[must_use]
pub fn error_handler(self, handler: Arc<dyn ErrorHandler<Err> + Send + Sync>) -> Self {
Self { error_handler: handler, ..self }
}
/// Specifies dependencies that can be used inside of handlers.
///
/// By default, there is no dependencies.
#[must_use]
pub fn dependencies(self, dependencies: DependencyMap) -> Self {
Self { dependencies, ..self }
}
/// Constructs [`Dispatcher`].
#[must_use]
pub fn build(self) -> Dispatcher<R, Err> {
Dispatcher {
bot: self.bot.clone(),
cache_me_bot: self.bot.cache_me(),
dependencies: self.dependencies,
handler: self.handler,
default_handler: self.default_handler,
error_handler: self.error_handler,
allowed_updates: Default::default(),
state: ShutdownToken::new(),
}
}
}
/// The base for update dispatching.
pub struct Dispatcher<R, Err> {
bot: R,
cache_me_bot: CacheMe<R>,
dependencies: DependencyMap,
handler: UpdateHandler<Err>,
default_handler: DefaultHandler,
error_handler: Arc<dyn ErrorHandler<Err> + Send + Sync>,
// TODO: respect allowed_udpates
allowed_updates: HashSet<AllowedUpdate>,
state: ShutdownToken,
}
impl<R> Dispatcher<R>
// TODO: it is allowed to return message as response on telegram request in
// webhooks, so we can allow this too. See more there: https://core.telegram.org/bots/api#making-requests-when-getting-updates
/// A handler that processes updates from Telegram.
pub type UpdateHandler<Err> = dptree::Handler<'static, DependencyMap, Result<(), Err>>;
type DefaultHandler = Box<dyn Fn(Arc<Update>) -> BoxFuture<'static, ()> + Send + Sync>;
impl<R, Err> Dispatcher<R, Err>
where
R: Send + 'static,
R: Requester + Clone + Send + Sync + 'static,
Err: Send + Sync + 'static,
{
/// Constructs a new dispatcher with the specified `requester`.
/// Constructs a new [`DispatcherBuilder`] with `bot` and `handler`.
#[must_use]
pub fn new(requester: R) -> Self {
Self {
requester,
messages_queue: None,
edited_messages_queue: None,
channel_posts_queue: None,
edited_channel_posts_queue: None,
inline_queries_queue: None,
chosen_inline_results_queue: None,
callback_queries_queue: None,
shipping_queries_queue: None,
pre_checkout_queries_queue: None,
polls_queue: None,
poll_answers_queue: None,
my_chat_members_queue: None,
chat_members_queue: None,
chat_join_requests_queue: None,
running_handlers: FuturesUnordered::new(),
state: ShutdownToken::new(),
pub fn builder(bot: R, handler: UpdateHandler<Err>) -> DispatcherBuilder<R, Err>
where
Err: Debug,
{
DispatcherBuilder {
bot,
dependencies: DependencyMap::new(),
handler,
default_handler: Box::new(|upd| {
log::warn!("Unhandled update: {:?}", upd);
Box::pin(async {})
}),
error_handler: LoggingErrorHandler::new(),
}
}
#[must_use]
fn new_tx<H, Upd>(&mut self, h: H) -> Tx<R, Upd>
where
H: DispatcherHandler<R, Upd> + Send + 'static,
Upd: Send + 'static,
R: Send + 'static,
{
let (tx, rx) = mpsc::unbounded_channel();
let join_handle = tokio::spawn(h.handle(rx));
self.running_handlers.push(join_handle);
Some(tx)
}
/// Setup the `^C` handler which [`shutdown`]s dispatching.
///
/// [`shutdown`]: ShutdownToken::shutdown
#[cfg(feature = "ctrlc_handler")]
#[must_use]
pub fn setup_ctrlc_handler(self) -> Self {
let token = self.state.clone();
tokio::spawn(async move {
loop {
tokio::signal::ctrl_c().await.expect("Failed to listen for ^C");
match token.shutdown() {
Ok(f) => {
log::info!("^C received, trying to shutdown the dispatcher...");
f.await;
log::info!("dispatcher is shutdown...");
}
Err(_) => {
log::info!("^C received, the dispatcher isn't running, ignoring the signal")
}
}
}
});
self
}
#[must_use]
pub fn messages_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, Message> + 'static + Send,
{
self.messages_queue = self.new_tx(h);
self
}
#[must_use]
pub fn edited_messages_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, Message> + 'static + Send,
{
self.edited_messages_queue = self.new_tx(h);
self
}
#[must_use]
pub fn channel_posts_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, Message> + 'static + Send,
{
self.channel_posts_queue = self.new_tx(h);
self
}
#[must_use]
pub fn edited_channel_posts_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, Message> + 'static + Send,
{
self.edited_channel_posts_queue = self.new_tx(h);
self
}
#[must_use]
pub fn inline_queries_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, InlineQuery> + 'static + Send,
{
self.inline_queries_queue = self.new_tx(h);
self
}
#[must_use]
pub fn chosen_inline_results_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, ChosenInlineResult> + 'static + Send,
{
self.chosen_inline_results_queue = self.new_tx(h);
self
}
#[must_use]
pub fn callback_queries_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, CallbackQuery> + 'static + Send,
{
self.callback_queries_queue = self.new_tx(h);
self
}
#[must_use]
pub fn shipping_queries_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, ShippingQuery> + 'static + Send,
{
self.shipping_queries_queue = self.new_tx(h);
self
}
#[must_use]
pub fn pre_checkout_queries_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, PreCheckoutQuery> + 'static + Send,
{
self.pre_checkout_queries_queue = self.new_tx(h);
self
}
#[must_use]
pub fn polls_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, Poll> + 'static + Send,
{
self.polls_queue = self.new_tx(h);
self
}
#[must_use]
pub fn poll_answers_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, PollAnswer> + 'static + Send,
{
self.poll_answers_queue = self.new_tx(h);
self
}
#[must_use]
pub fn my_chat_members_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, ChatMemberUpdated> + 'static + Send,
{
self.my_chat_members_queue = self.new_tx(h);
self
}
#[must_use]
pub fn chat_members_handler<H>(mut self, h: H) -> Self
where
H: DispatcherHandler<R, ChatMemberUpdated> + 'static + Send,
{
self.chat_members_queue = self.new_tx(h);
self
}
/// Starts your bot with the default parameters.
///
/// The default parameters are a long polling update listener and log all
/// errors produced by this listener).
/// errors produced by this listener.
///
/// Please note that after shutting down (either because of [`shutdown`],
/// [a ctrlc signal], or [`UpdateListener`] returning `None`) all handlers
/// will be gone. As such, to restart listening you need to re-add
/// handlers.
/// Each time a handler is invoked, [`Dispatcher`] adds the following
/// dependencies (in addition to those passed to
/// [`DispatcherBuilder::dependencies`]):
///
/// - Your bot passed to [`Dispatcher::builder`];
/// - An update from Telegram;
/// - [`crate::types::Me`] (can be used in [`HandlerExt::filter_command`]).
///
/// [`shutdown`]: ShutdownToken::shutdown
/// [a ctrlc signal]: Dispatcher::setup_ctrlc_handler
/// [`HandlerExt::filter_command`]: crate::dispatching::HandlerExt::filter_command
pub async fn dispatch(&mut self)
where
R: Requester + Clone,
<R as Requester>::GetUpdates: Send,
{
let listener = update_listeners::polling_default(self.requester.clone()).await;
let listener = update_listeners::polling_default(self.bot.clone()).await;
let error_handler =
LoggingErrorHandler::with_custom_text("An error from the update listener");
@ -267,10 +162,7 @@ where
/// Starts your bot with custom `update_listener` and
/// `update_listener_error_handler`.
///
/// Please note that after shutting down (either because of [`shutdown`],
/// [a ctrlc signal], or [`UpdateListener`] returning `None`) all handlers
/// will be gone. As such, to restart listening you need to re-add
/// handlers.
/// This method adds the same dependencies as [`Dispatcher::dispatch`].
///
/// [`shutdown`]: ShutdownToken::shutdown
/// [a ctrlc signal]: Dispatcher::setup_ctrlc_handler
@ -282,9 +174,8 @@ where
UListener: UpdateListener<ListenerE> + 'a,
Eh: ErrorHandler<ListenerE> + 'a,
ListenerE: Debug,
R: Requester + Clone,
{
self.hint_allowed_updates(&mut update_listener);
update_listener.hint_allowed_updates(&mut self.allowed_updates.clone().into_iter());
let shutdown_check_timeout = shutdown_check_timeout_for(&update_listener);
let mut stop_token = Some(update_listener.stop_token());
@ -309,222 +200,111 @@ where
if let Some(token) = stop_token.take() {
log::debug!("Start shutting down dispatching...");
token.stop();
break;
}
}
}
}
self.wait_for_handlers().await;
// TODO: wait for executing handlers?
self.state.done();
}
async fn process_update<LErr, LErrHandler>(
&self,
update: Result<Update, LErr>,
err_handler: &Arc<LErrHandler>,
) where
LErrHandler: ErrorHandler<LErr>,
{
match update {
Ok(upd) => {
if let UpdateKind::Error(err) = upd.kind {
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/issues.",
err,
);
return;
}
let mut deps = self.dependencies.clone();
deps.insert(upd);
deps.insert(self.bot.clone());
deps.insert(
self.cache_me_bot.get_me().send().await.expect("Failed to retrieve 'me'"),
);
match self.handler.dispatch(deps).await {
ControlFlow::Break(Ok(())) => {}
ControlFlow::Break(Err(err)) => {
self.error_handler.clone().handle_error(err).await
}
ControlFlow::Continue(deps) => {
let upd = deps.get();
(self.default_handler)(upd).await;
}
}
}
Err(err) => err_handler.clone().handle_error(err).await,
}
}
/// Setups the `^C` handler that [`shutdown`]s dispatching.
///
/// [`shutdown`]: ShutdownToken::shutdown
#[cfg(feature = "ctrlc_handler")]
pub fn setup_ctrlc_handler(&mut self) -> &mut Self {
let token = self.state.clone();
tokio::spawn(async move {
loop {
tokio::signal::ctrl_c().await.expect("Failed to listen for ^C");
match token.shutdown() {
Ok(f) => {
log::info!("^C received, trying to shutdown the dispatcher...");
f.await;
log::info!("dispatcher is shutdown...");
}
Err(_) => {
log::info!("^C received, the dispatcher isn't running, ignoring the signal")
}
}
}
});
self
}
/// Returns a shutdown token, which can later be used to shutdown
/// dispatching.
pub fn shutdown_token(&self) -> ShutdownToken {
self.state.clone()
}
}
async fn process_update<ListenerE, Eh>(
&self,
update: Result<Update, ListenerE>,
update_listener_error_handler: &Arc<Eh>,
) where
R: Requester + Clone,
Eh: ErrorHandler<ListenerE>,
ListenerE: Debug,
{
{
log::trace!("Dispatcher received an update: {:?}", update);
#[cfg(test)]
mod tests {
use std::convert::Infallible;
let update = match update {
Ok(update) => update,
Err(error) => {
Arc::clone(update_listener_error_handler).handle_error(error).await;
return;
}
};
use teloxide_core::Bot;
match update.kind {
UpdateKind::Message(message) => {
send(&self.requester, &self.messages_queue, message, "UpdateKind::Message")
}
UpdateKind::EditedMessage(message) => send(
&self.requester,
&self.edited_messages_queue,
message,
"UpdateKind::EditedMessage",
),
UpdateKind::ChannelPost(post) => send(
&self.requester,
&self.channel_posts_queue,
post,
"UpdateKind::ChannelPost",
),
UpdateKind::EditedChannelPost(post) => send(
&self.requester,
&self.edited_channel_posts_queue,
post,
"UpdateKind::EditedChannelPost",
),
UpdateKind::InlineQuery(query) => send(
&self.requester,
&self.inline_queries_queue,
query,
"UpdateKind::InlineQuery",
),
UpdateKind::ChosenInlineResult(result) => send(
&self.requester,
&self.chosen_inline_results_queue,
result,
"UpdateKind::ChosenInlineResult",
),
UpdateKind::CallbackQuery(query) => send(
&self.requester,
&self.callback_queries_queue,
query,
"UpdateKind::CallbackQuer",
),
UpdateKind::ShippingQuery(query) => send(
&self.requester,
&self.shipping_queries_queue,
query,
"UpdateKind::ShippingQuery",
),
UpdateKind::PreCheckoutQuery(query) => send(
&self.requester,
&self.pre_checkout_queries_queue,
query,
"UpdateKind::PreCheckoutQuery",
),
UpdateKind::Poll(poll) => {
send(&self.requester, &self.polls_queue, poll, "UpdateKind::Poll")
}
UpdateKind::PollAnswer(answer) => send(
&self.requester,
&self.poll_answers_queue,
answer,
"UpdateKind::PollAnswer",
),
UpdateKind::MyChatMember(chat_member_updated) => send(
&self.requester,
&self.my_chat_members_queue,
chat_member_updated,
"UpdateKind::MyChatMember",
),
UpdateKind::ChatMember(chat_member_updated) => send(
&self.requester,
&self.chat_members_queue,
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/issues.",
err,
);
}
use super::*;
#[tokio::test]
async fn test_tokio_spawn() {
tokio::spawn(async {
// Just check that this code compiles.
if false {
Dispatcher::<_, Infallible>::builder(Bot::new(""), dptree::entry())
.build()
.dispatch()
.await;
}
}
}
fn hint_allowed_updates<E>(&self, listener: &mut impl UpdateListener<E>) {
fn hint_handler_allowed_update<T>(
queue: &Option<T>,
kind: AllowedUpdate,
) -> std::option::IntoIter<AllowedUpdate> {
queue.as_ref().map(|_| kind).into_iter()
}
let mut allowed = hint_handler_allowed_update(&self.messages_queue, AllowedUpdate::Message)
.chain(hint_handler_allowed_update(
&self.edited_messages_queue,
AllowedUpdate::EditedMessage,
))
.chain(hint_handler_allowed_update(
&self.channel_posts_queue,
AllowedUpdate::ChannelPost,
))
.chain(hint_handler_allowed_update(
&self.edited_channel_posts_queue,
AllowedUpdate::EditedChannelPost,
))
.chain(hint_handler_allowed_update(
&self.inline_queries_queue,
AllowedUpdate::InlineQuery,
))
.chain(hint_handler_allowed_update(
&self.chosen_inline_results_queue,
AllowedUpdate::ChosenInlineResult,
))
.chain(hint_handler_allowed_update(
&self.callback_queries_queue,
AllowedUpdate::CallbackQuery,
))
.chain(hint_handler_allowed_update(
&self.shipping_queries_queue,
AllowedUpdate::ShippingQuery,
))
.chain(hint_handler_allowed_update(
&self.pre_checkout_queries_queue,
AllowedUpdate::PreCheckoutQuery,
))
.chain(hint_handler_allowed_update(&self.polls_queue, AllowedUpdate::Poll))
.chain(hint_handler_allowed_update(&self.poll_answers_queue, AllowedUpdate::PollAnswer))
.chain(hint_handler_allowed_update(
&self.my_chat_members_queue,
AllowedUpdate::MyChatMember,
))
.chain(hint_handler_allowed_update(
&self.chat_members_queue,
AllowedUpdate::ChatMember,
));
listener.hint_allowed_updates(&mut allowed);
}
async fn wait_for_handlers(&mut self) {
log::debug!("Waiting for handlers to finish");
// Drop all senders, so handlers can stop
self.messages_queue.take();
self.edited_messages_queue.take();
self.channel_posts_queue.take();
self.edited_channel_posts_queue.take();
self.inline_queries_queue.take();
self.chosen_inline_results_queue.take();
self.callback_queries_queue.take();
self.shipping_queries_queue.take();
self.pre_checkout_queries_queue.take();
self.polls_queue.take();
self.poll_answers_queue.take();
self.my_chat_members_queue.take();
self.chat_members_queue.take();
// Wait untill all handlers finish
self.running_handlers.by_ref().for_each(|_| async {}).await;
}
}
fn send<'a, R, Upd>(requester: &'a R, tx: &'a Tx<R, Upd>, update: Upd, variant: &'static str)
where
Upd: Debug,
R: Requester + Clone,
{
if let Some(tx) = tx {
if let Err(error) = tx.send(UpdateWithCx { requester: requester.clone(), update }) {
log::error!(
"The RX part of the {} channel is closed, but an update is received.\nError:{}\n",
variant,
error
);
}
})
.await
.unwrap();
}
}

31
src/dispatching/dispatcher_handler.rs
View file

@ -1,31 +0,0 @@
use std::future::Future;
use crate::dispatching::{DispatcherHandlerRx, UpdateWithCx};
use futures::future::BoxFuture;
/// An asynchronous handler of a stream of updates used in [`Dispatcher`].
///
/// See the [module-level documentation](crate::dispatching) for the design
/// overview.
///
/// [`Dispatcher`]: crate::dispatching::Dispatcher
#[deprecated(note = "Use dispatching2 instead")]
pub trait DispatcherHandler<R, Upd> {
#[must_use]
fn handle(self, updates: DispatcherHandlerRx<R, Upd>) -> BoxFuture<'static, ()>
where
UpdateWithCx<R, Upd>: Send + 'static;
}
impl<R, Upd, F, Fut> DispatcherHandler<R, Upd> for F
where
F: FnOnce(DispatcherHandlerRx<R, Upd>) -> Fut + Send + 'static,
Fut: Future<Output = ()> + Send + 'static,
{
fn handle(self, updates: DispatcherHandlerRx<R, Upd>) -> BoxFuture<'static, ()>
where
UpdateWithCx<R, Upd>: Send + 'static,
{
Box::pin(async move { self(updates).await })
}
}

62
src/dispatching/dispatcher_handler_rx_ext.rs
View file

@ -1,62 +0,0 @@
use crate::{dispatching::UpdateWithCx, utils::command::BotCommands};
use futures::{stream::BoxStream, Stream, StreamExt};
use teloxide_core::types::Message;
/// An extension trait to be used with [`DispatcherHandlerRx`].
///
/// See the [module-level documentation](crate::dispatching) for the design
/// overview.
///
/// [`DispatcherHandlerRx`]: crate::dispatching::DispatcherHandlerRx
#[deprecated(note = "Use dispatching2 instead")]
pub trait DispatcherHandlerRxExt<R> {
/// Extracts only text messages from this stream of arbitrary messages.
fn text_messages(self) -> BoxStream<'static, (UpdateWithCx<R, Message>, String)>
where
Self: Stream<Item = UpdateWithCx<R, Message>>,
R: Send + 'static;
/// Extracts only commands with their arguments from this stream of
/// arbitrary messages.
fn commands<C, N>(self, bot_name: N) -> BoxStream<'static, (UpdateWithCx<R, Message>, C)>
where
Self: Stream<Item = UpdateWithCx<R, Message>>,
C: BotCommands,
N: Into<String> + Send,
R: Send + 'static;
}
impl<R, T> DispatcherHandlerRxExt<R> for T
where
T: Send + 'static,
{
fn text_messages(self) -> BoxStream<'static, (UpdateWithCx<R, Message>, String)>
where
Self: Stream<Item = UpdateWithCx<R, Message>>,
R: Send + 'static,
{
self.filter_map(|cx| async move {
let text = cx.update.text().map(ToOwned::to_owned);
text.map(move |text| (cx, text))
})
.boxed()
}
fn commands<C, N>(self, bot_name: N) -> BoxStream<'static, (UpdateWithCx<R, Message>, C)>
where
Self: Stream<Item = UpdateWithCx<R, Message>>,
C: BotCommands,
N: Into<String> + Send,
R: Send + 'static,
{
let bot_name = bot_name.into();
self.text_messages()
.filter_map(move |(cx, text)| {
let bot_name = bot_name.clone();
async move { C::parse(&text, &bot_name).map(|command| (cx, command)).ok() }
})
.boxed()
}
}

0
src/dispatching2/filter_ext.rs → src/dispatching/filter_ext.rs
View file

2
src/dispatching2/handler_ext.rs → src/dispatching/handler_ext.rs
View file

@ -1,7 +1,7 @@
use std::sync::Arc;
use crate::{
dispatching2::{
dispatching::{
dialogue::{Dialogue, GetChatId, Storage},
HandlerFactory,
},

0
src/dispatching2/handler_factory.rs → src/dispatching/handler_factory.rs
View file

151
src/dispatching/mod.rs
View file

@ -1,74 +1,113 @@
//! Old updates dispatching (**DEPRECATED**: use [`crate::dispatching2`]
//! instead).
//! An update dispatching model based on [`dptree`].
//!
//! The key type here is [`Dispatcher`]. It encapsulates [`Bot`] and handlers
//! for [all the update kinds].
//! In teloxide, updates are dispatched by a pipleine. The central type is
//! [`dptree::Handler`] -- it represents a handler of an update; since the API
//! is highly declarative, you can combine handlers with each other via such
//! methods as [`dptree::Handler::chain`] and [`dptree::Handler::branch`]. The
//! former method pipes one handler to another one, whilst the latter creates a
//! new node, as communicated by the name. For more information, please refer to
//! the documentation of [`dptree`].
//!
//! Every handler accept [`tokio::sync::mpsc::UnboundedReceiver`] (the RX halve
//! of an asynchronous channel). Inside a body of your handler, you typically
//! asynchronously concurrently iterate through updates like this:
//! The pattern itself is called [chain of responsibility], a well-known design
//! technique across OOP developers. But unlike typical object-oriented design,
//! we employ declarative FP-style functions like [`dptree::filter`],
//! [`dptree::filter_map`], and [`dptree::endpoint`]; these functions create
//! special forms of [`dptree::Handler`]; for more information, please refer to
//! their respective documentation. Each of these higher-order functions accept
//! a closure that is made into a handler -- this closure can take any
//! additional parameters, which must be supplied while creating [`Dispatcher`]
//! (see [`DispatcherBuilder::dependencies`]).
//!
//! ```
//! The [`Dispatcher`] type puts all these things together: it only provides
//! [`Dispatcher::dispatch`] and a handful of other methods. Once you call
//! `.dispatch()`, it will retrieve updates from the Telegram server and pass
//! them to your handler, which is a parameter of [`Dispatcher::builder`].
//!
//! Let us look at a simple example:
//!
//!
//! ([Full](https://github.com/teloxide/teloxide/blob/master/examples/shared_state.rs))
//!
//! ```no_run
//! use std::sync::atomic::{AtomicU64, Ordering};
//!
//! use once_cell::sync::Lazy;
//! use teloxide::prelude::*;
//! use tokio_stream::wrappers::UnboundedReceiverStream;
//!
//! async fn handle_messages(rx: DispatcherHandlerRx<AutoSend<Bot>, Message>) {
//! UnboundedReceiverStream::new(rx)
//! .for_each_concurrent(None, |message| async move {
//! dbg!(message.update);
//! })
//! .await;
//! }
//! static MESSAGES_TOTAL: Lazy<AtomicU64> = Lazy::new(AtomicU64::default);
//!
//! # #[tokio::main]
//! # async fn main() {
//! pretty_env_logger::init();
//! log::info!("Starting shared_state_bot...");
//!
//! let bot = Bot::from_env().auto_send();
//!
//! let handler = Update::filter_message().branch(dptree::endpoint(
//! |msg: Message, bot: AutoSend<Bot>| async move {
//! let previous = MESSAGES_TOTAL.fetch_add(1, Ordering::Relaxed);
//! bot.send_message(msg.chat.id, format!("I received {} messages in total.", previous))
//! .await?;
//! respond(())
//! },
//! ));
//!
//! Dispatcher::builder(bot, handler).build().setup_ctrlc_handler().dispatch().await;
//! # }
//! ```
//!
//! When [`Update`] is received from Telegram, [`Dispatcher`] pushes it into an
//! appropriate handler, depending on its kind. That's simple!
//! 1. First, we create the bot: `let bot = Bot::from_env().auto_send()`.
//! 2. Then we construct an update handler. While it is possible to handle all
//! kinds of [`crate::types::Update`], here we are only interested in
//! [`crate::types::Message`]: [`UpdateFilterExt::filter_message`] create a
//! handler object which filters all messages out of a generic update.
//! 3. By doing `.branch(dptree::endpoint(...))`, we set up a custom handling
//! closure that receives `msg: Message` and `bot: AutoSend<Bot>`. There are
//! called dependencies: `msg` is supplied by
//! [`UpdateFilterExt::filter_message`], while `bot` is supplied by
//! [`Dispatcher`].
//!
//! **Note** that handlers must implement [`DispatcherHandler`], which means
//! that:
//! - You are able to supply [`DialogueDispatcher`] as a handler.
//! - You are able to supply functions that accept
//! [`tokio::sync::mpsc::UnboundedReceiver`] and return `Future<Output = ()>`
//! as a handler.
//! That being said, if we receive a message, the dispatcher will call our
//! handler, but if we receive something other than a message (e.g., a channel
//! post), you will see an unhandled update notice in your terminal.
//!
//! Since they implement [`DispatcherHandler`] too.
//! This is a very limited example of update pipelining facilities. In more
//! involved scenarios, there are multiple branches and chains; if one element
//! of a chain fails to handle an update, the update will be passed forwards; if
//! no handler succeeds at handling the update, [`Dispatcher`] will invoke a
//! default handler set up via [`DispatcherBuilder::default_handler`].
//!
//! [See the examples](https://github.com/teloxide/teloxide/tree/master/examples).
//! Update pipelining provides several advantages over the typical `match
//! (update.kind) { ... }` approach:
//!
//! [`Dispatcher`]: crate::dispatching::Dispatcher
//! [all the update kinds]: crate::types::UpdateKind
//! [`Update`]: crate::types::Update
//! [`ErrorHandler`]: crate::dispatching::ErrorHandler
//! [`DispatcherHandler`]: crate::dispatching::DispatcherHandler
//! [`DialogueDispatcher`]: crate::dispatching::dialogue::DialogueDispatcher
//! [`DispatcherHandlerResult`]: crate::dispatching::DispatcherHandlerResult
//! [`Bot`]: crate::Bot
//! [`tokio::sync::mpsc::UnboundedReceiver`]: https://docs.rs/tokio/0.2.11/tokio/sync/mpsc/struct.UnboundedReceiver.html
//! [examples/dialogue_bot]: https://github.com/teloxide/teloxide/tree/master/examples/dialogue_bot
//! 1. It supports _extension_: e.g., you
//! can define extension filters or some other handlers and then combine them in
//! a single place, thus facilitating loose coupling.
//! 2. Pipelining exhibits a natural syntax for expressing message processing.
//! 3. Lastly, it provides a primitive form of [dependency injection (DI)],
//! which allows you to deal with such objects as a bot and various update types
//! easily.
//!
//! For a more involved example, see [`examples/dispatching_features.rs`](https://github.com/teloxide/teloxide/blob/master/examples/dispatching_features.rs).
//!
//! TODO: explain a more involved example with multiple branches.
//!
//! [chain of responsibility]: https://en.wikipedia.org/wiki/Chain-of-responsibility_pattern
//! [dependency injection (DI)]: https://en.wikipedia.org/wiki/Dependency_injection
#![allow(deprecated)]
#[cfg(all(feature = "ctrlc_handler"))]
pub mod repls;
pub mod dialogue;
mod dispatcher;
mod filter_ext;
mod handler_ext;
mod handler_factory;
pub mod stop_token;
pub mod update_listeners;
#[cfg(feature = "ctrlc_handler")]
pub(crate) mod repls;
mod dispatcher;
mod dispatcher_handler;
mod dispatcher_handler_rx_ext;
mod update_with_cx;
pub use crate::utils::shutdown_token::{IdleShutdownError, ShutdownToken};
pub use dispatcher::Dispatcher;
pub use dispatcher_handler::DispatcherHandler;
pub use dispatcher_handler_rx_ext::DispatcherHandlerRxExt;
use tokio::sync::mpsc::UnboundedReceiver;
pub use update_with_cx::{UpdateWithCx, UpdateWithCxRequesterType};
/// A type of a stream, consumed by [`Dispatcher`]'s handlers.
///
/// [`Dispatcher`]: crate::dispatching::Dispatcher
#[deprecated(note = "Use dispatching2 instead")]
pub type DispatcherHandlerRx<R, Upd> = UnboundedReceiver<UpdateWithCx<R, Upd>>;
pub use dispatcher::{Dispatcher, DispatcherBuilder, UpdateHandler};
pub use filter_ext::{MessageFilterExt, UpdateFilterExt};
pub use handler_ext::HandlerExt;
pub use handler_factory::HandlerFactory;

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

@ -1,21 +1,20 @@
use crate::{
dispatching::{
update_listeners, update_listeners::UpdateListener, Dispatcher, DispatcherHandlerRx,
DispatcherHandlerRxExt, UpdateWithCx,
update_listeners, update_listeners::UpdateListener, HandlerExt, UpdateFilterExt,
},
error_handlers::{LoggingErrorHandler, OnError},
error_handlers::LoggingErrorHandler,
types::Update,
utils::command::BotCommands,
};
use futures::StreamExt;
use std::{fmt::Debug, future::Future, sync::Arc};
use teloxide_core::{requests::Requester, types::Message};
use tokio_stream::wrappers::UnboundedReceiverStream;
use dptree::di::{DependencyMap, Injectable};
use std::{fmt::Debug, marker::PhantomData};
use teloxide_core::requests::Requester;
/// A [REPL] for commands.
///
/// All errors from an update listener and handler will be logged.
///
/// # Caution
/// ## Caution
/// **DO NOT** use this function together with [`Dispatcher`] and other REPLs,
/// because Telegram disallow multiple requests at the same time from the same
/// bot.
@ -23,24 +22,21 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
/// [REPL]: https://en.wikipedia.org/wiki/Read-eval-print_loop
/// [`Dispatcher`]: crate::dispatching::Dispatcher
#[cfg(feature = "ctrlc_handler")]
pub async fn commands_repl<R, Cmd, H, Fut, HandlerE, N>(requester: R, bot_name: N, handler: H)
pub async fn commands_repl<'a, R, Cmd, H, E, Args>(bot: R, handler: H, cmd: PhantomData<Cmd>)
where
Cmd: BotCommands + Send + 'static,
H: Fn(UpdateWithCx<R, Message>, Cmd) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<(), HandlerE>> + Send + 'static,
Result<(), HandlerE>: OnError<HandlerE>,
HandlerE: Debug + Send,
N: Into<String> + Send + 'static,
R: Requester + Send + Clone + 'static,
Cmd: BotCommands + Send + Sync + 'static,
H: Injectable<DependencyMap, Result<(), E>, Args> + Send + Sync + 'static,
R: Requester + Clone + Send + Sync + 'static,
<R as Requester>::GetUpdates: Send,
E: Debug + Send + Sync + 'static,
{
let cloned_requester = requester.clone();
let cloned_bot = bot.clone();
commands_repl_with_listener(
requester,
bot_name,
bot,
handler,
update_listeners::polling_default(cloned_requester).await,
update_listeners::polling_default(cloned_bot).await,
cmd,
)
.await;
}
@ -49,7 +45,7 @@ where
///
/// All errors from an update listener and handler will be logged.
///
/// # Caution
/// ## Caution
/// **DO NOT** use this function together with [`Dispatcher`] and other REPLs,
/// because Telegram disallow multiple requests at the same time from the same
/// bot.
@ -58,41 +54,35 @@ 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, Fut, L, ListenerE, HandlerE, N>(
requester: R,
bot_name: N,
pub async fn commands_repl_with_listener<'a, R, Cmd, H, L, ListenerE, E, Args>(
bot: R,
handler: H,
listener: L,
_cmd: PhantomData<Cmd>,
) where
Cmd: BotCommands + Send + 'static,
H: Fn(UpdateWithCx<R, Message>, Cmd) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<(), HandlerE>> + Send + 'static,
Cmd: BotCommands + Send + Sync + 'static,
H: Injectable<DependencyMap, Result<(), E>, Args> + Send + Sync + 'static,
L: UpdateListener<ListenerE> + Send + 'a,
ListenerE: Debug + Send + 'a,
Result<(), HandlerE>: OnError<HandlerE>,
HandlerE: Debug + Send,
N: Into<String> + Send + 'static,
R: Requester + Clone + Send + 'static,
R: Requester + Clone + Send + Sync + 'static,
E: Debug + Send + Sync + 'static,
{
let handler = Arc::new(handler);
use crate::dispatching::Dispatcher;
Dispatcher::<R>::new(requester)
.messages_handler(move |rx: DispatcherHandlerRx<R, Message>| {
UnboundedReceiverStream::new(rx).commands::<Cmd, N>(bot_name).for_each_concurrent(
None,
move |(cx, cmd)| {
let handler = Arc::clone(&handler);
// Other update types are of no interest to use since this REPL is only for
// commands. See <https://github.com/teloxide/teloxide/issues/557>.
let ignore_update = |_upd| Box::pin(async {});
async move {
handler(cx, cmd).await.log_on_error().await;
}
},
)
})
.setup_ctrlc_handler()
.dispatch_with_listener(
listener,
LoggingErrorHandler::with_custom_text("An error from the update listener"),
)
.await
Dispatcher::builder(
bot,
Update::filter_message().filter_command::<Cmd>().branch(dptree::endpoint(handler)),
)
.default_handler(ignore_update)
.build()
.setup_ctrlc_handler()
.dispatch_with_listener(
listener,
LoggingErrorHandler::with_custom_text("An error from the update listener"),
)
.await;
}

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

@ -1,96 +0,0 @@
use crate::{
dispatching::{
dialogue::{DialogueDispatcher, DialogueStage, DialogueWithCx, InMemStorageError},
update_listeners,
update_listeners::UpdateListener,
Dispatcher, UpdateWithCx,
},
error_handlers::LoggingErrorHandler,
};
use std::{fmt::Debug, future::Future, sync::Arc};
use teloxide_core::{requests::Requester, types::Message};
/// A [REPL] for dialogues.
///
/// All errors from an update listener and handler will be logged. This function
/// uses [`InMemStorage`].
///
/// # Caution
/// **DO NOT** use this function together with [`Dispatcher`] and other REPLs,
/// because Telegram disallow multiple requests at the same time from the same
/// bot.
///
/// [REPL]: https://en.wikipedia.org/wiki/Read-eval-print_loop
/// [`Dispatcher`]: crate::dispatching::Dispatcher
/// [`InMemStorage`]: crate::dispatching::dialogue::InMemStorage
#[cfg(feature = "ctrlc_handler")]
pub async fn dialogues_repl<'a, R, H, D, Fut>(requester: R, handler: H)
where
H: Fn(UpdateWithCx<R, Message>, D) -> Fut + Send + Sync + 'static,
D: Clone + Default + Send + 'static,
Fut: Future<Output = DialogueStage<D>> + Send + 'static,
R: Requester + Send + Clone + 'static,
<R as Requester>::GetUpdates: Send,
{
let cloned_requester = requester.clone();
dialogues_repl_with_listener(
requester,
handler,
update_listeners::polling_default(cloned_requester).await,
)
.await;
}
/// Like [`dialogues_repl`], but with a custom [`UpdateListener`].
///
/// All errors from an update listener and handler will be logged. This function
/// uses [`InMemStorage`].
///
/// # Caution
/// **DO NOT** use this function together with [`Dispatcher`] and other REPLs,
/// because Telegram disallow multiple requests at the same time from the same
/// bot.
///
/// [`Dispatcher`]: crate::dispatching::Dispatcher
/// [`dialogues_repl`]: crate::dispatching::repls::dialogues_repl()
/// [`UpdateListener`]: crate::dispatching::update_listeners::UpdateListener
/// [`InMemStorage`]: crate::dispatching::dialogue::InMemStorage
#[cfg(feature = "ctrlc_handler")]
pub async fn dialogues_repl_with_listener<'a, R, H, D, Fut, L, ListenerE>(
requester: R,
handler: H,
listener: L,
) where
H: Fn(UpdateWithCx<R, Message>, D) -> Fut + Send + Sync + 'static,
D: Clone + Default + Send + 'static,
Fut: Future<Output = DialogueStage<D>> + Send + 'static,
L: UpdateListener<ListenerE> + Send + 'a,
ListenerE: Debug + Send + 'a,
R: Requester + Send + Clone + 'static,
{
let handler = Arc::new(handler);
Dispatcher::new(requester)
.messages_handler(DialogueDispatcher::new(
move |DialogueWithCx { cx, dialogue }: DialogueWithCx<
R,
Message,
D,
InMemStorageError,
>| {
let handler = Arc::clone(&handler);
async move {
let dialogue = dialogue.expect("std::convert::Infallible");
handler(cx, dialogue).await
}
},
))
.setup_ctrlc_handler()
.dispatch_with_listener(
listener,
LoggingErrorHandler::with_custom_text("An error from the update listener"),
)
.await;
}

4
src/dispatching/repls/mod.rs
View file

@ -1,7 +1,7 @@
//! REPLs for dispatching updates.
mod commands_repl;
mod dialogues_repl;
mod repl;
pub use commands_repl::{commands_repl, commands_repl_with_listener};
pub use dialogues_repl::{dialogues_repl, dialogues_repl_with_listener};
pub use repl::{repl, repl_with_listener};

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

@ -1,14 +1,11 @@
use crate::{
dispatching::{
update_listeners, update_listeners::UpdateListener, Dispatcher, DispatcherHandlerRx,
UpdateWithCx,
},
dispatching::{update_listeners, update_listeners::UpdateListener, UpdateFilterExt},
error_handlers::{LoggingErrorHandler, OnError},
types::Update,
};
use futures::StreamExt;
use std::{fmt::Debug, future::Future, sync::Arc};
use teloxide_core::{requests::Requester, types::Message};
use tokio_stream::wrappers::UnboundedReceiverStream;
use dptree::di::{DependencyMap, Injectable};
use std::fmt::Debug;
use teloxide_core::requests::Requester;
/// A [REPL] for messages.
///
@ -22,22 +19,16 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
/// [REPL]: https://en.wikipedia.org/wiki/Read-eval-print_loop
/// [`Dispatcher`]: crate::dispatching::Dispatcher
#[cfg(feature = "ctrlc_handler")]
pub async fn repl<R, H, Fut, E>(requester: R, handler: H)
pub async fn repl<R, H, E, Args>(bot: R, handler: H)
where
H: Fn(UpdateWithCx<R, Message>) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<(), E>> + Send + 'static,
H: Injectable<DependencyMap, Result<(), E>, Args> + Send + Sync + 'static,
Result<(), E>: OnError<E>,
E: Debug + Send,
R: Requester + Send + Clone + 'static,
E: Debug + Send + Sync + 'static,
R: Requester + Send + Sync + Clone + 'static,
<R as Requester>::GetUpdates: Send,
{
let cloned_requester = requester.clone();
repl_with_listener(
requester,
handler,
update_listeners::polling_default(cloned_requester).await,
)
.await;
let cloned_bot = bot.clone();
repl_with_listener(bot, handler, update_listeners::polling_default(cloned_bot).await).await;
}
/// Like [`repl`], but with a custom [`UpdateListener`].
@ -53,31 +44,24 @@ 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, Fut, E, L, ListenerE>(
requester: R,
handler: H,
listener: L,
) where
H: Fn(UpdateWithCx<R, Message>) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<(), E>> + Send + 'static,
pub async fn repl_with_listener<'a, R, H, E, L, ListenerE, Args>(bot: R, handler: H, listener: L)
where
H: Injectable<DependencyMap, Result<(), E>, Args> + Send + Sync + 'static,
L: UpdateListener<ListenerE> + Send + 'a,
ListenerE: Debug,
Result<(), E>: OnError<E>,
E: Debug + Send,
R: Requester + Clone + Send + 'static,
E: Debug + Send + Sync + 'static,
R: Requester + Clone + Send + Sync + 'static,
{
let handler = Arc::new(handler);
use crate::dispatching::Dispatcher;
Dispatcher::new(requester)
.messages_handler(|rx: DispatcherHandlerRx<R, Message>| {
UnboundedReceiverStream::new(rx).for_each_concurrent(None, move |message| {
let handler = Arc::clone(&handler);
// Other update types are of no interest to use since this REPL is only for
// messages. See <https://github.com/teloxide/teloxide/issues/557>.
let ignore_update = |_upd| Box::pin(async {});
async move {
handler(message).await.log_on_error().await;
}
})
})
Dispatcher::builder(bot, Update::filter_message().branch(dptree::endpoint(handler)))
.default_handler(ignore_update)
.build()
.setup_ctrlc_handler()
.dispatch_with_listener(
listener,
@ -89,7 +73,7 @@ pub async fn repl_with_listener<'a, R, H, Fut, E, L, ListenerE>(
#[test]
fn repl_is_send() {
let bot = crate::Bot::new("");
let repl = crate::repl(bot, |_| async { crate::respond(()) });
let repl = crate::repl(bot, || async { crate::respond(()) });
assert_send(&repl);
fn assert_send(_: &impl Send) {}

248
src/dispatching/update_with_cx.rs
View file

@ -1,248 +0,0 @@
use crate::dispatching::dialogue::GetChatId;
use teloxide_core::{
payloads::{
SendAnimationSetters, SendAudioSetters, SendContactSetters, SendDocumentSetters,
SendLocationSetters, SendMediaGroupSetters, SendMessageSetters, SendPhotoSetters,
SendStickerSetters, SendVenueSetters, SendVideoNoteSetters, SendVideoSetters,
SendVoiceSetters,
},
requests::{Request, Requester},
types::{ChatId, InputFile, InputMedia, Message},
};
/// A [`Dispatcher`]'s handler's context of a bot and an update.
///
/// See the [module-level documentation](crate::dispatching) for the design
/// overview.
///
/// [`Dispatcher`]: crate::dispatching::Dispatcher
#[derive(Debug)]
#[deprecated(note = "Use dispatching2 instead")]
pub struct UpdateWithCx<R, Upd> {
pub requester: R,
pub update: Upd,
}
impl<Upd, R> GetChatId for UpdateWithCx<R, Upd>
where
Upd: GetChatId,
{
fn chat_id(&self) -> i64 {
self.update.chat_id()
}
}
#[doc(hidden)]
// Now it is used only inside `#[teloxide(subtransition)]` for type inference.
pub trait UpdateWithCxRequesterType {
type Requester;
}
impl<R, Upd> UpdateWithCxRequesterType for UpdateWithCx<R, Upd> {
type Requester = R;
}
impl<R> UpdateWithCx<R, Message>
where
R: Requester,
{
/// A shortcut for `.answer(text).send().await`.
#[deprecated(note = "Use .answer(text).await instead")]
pub async fn answer_str<T>(&self, text: T) -> Result<Message, R::Err>
where
T: Into<String>,
R::SendMessage: std::future::Future,
{
self.answer(text).send().await
}
pub fn answer<T>(&self, text: T) -> R::SendMessage
where
T: Into<String>,
{
self.requester.send_message(self.chat_id(), text)
}
pub fn reply_to<T>(&self, text: T) -> R::SendMessage
where
T: Into<String>,
{
self.requester.send_message(self.chat_id(), text).reply_to_message_id(self.update.id)
}
pub fn reply_audio(&self, audio: InputFile) -> R::SendAudio {
self.requester.send_audio(self.update.chat.id, audio).reply_to_message_id(self.update.id)
}
pub fn reply_animation(&self, animation: InputFile) -> R::SendAnimation {
self.requester
.send_animation(self.update.chat.id, animation)
.reply_to_message_id(self.update.id)
}
pub fn reply_document(&self, document: InputFile) -> R::SendDocument {
self.requester
.send_document(self.update.chat.id, document)
.reply_to_message_id(self.update.id)
}
pub fn reply_photo(&self, photo: InputFile) -> R::SendPhoto {
self.requester.send_photo(self.update.chat.id, photo).reply_to_message_id(self.update.id)
}
pub fn reply_video(&self, video: InputFile) -> R::SendVideo {
self.requester.send_video(self.update.chat.id, video).reply_to_message_id(self.update.id)
}
pub fn reply_voice(&self, voice: InputFile) -> R::SendVoice {
self.requester.send_voice(self.update.chat.id, voice).reply_to_message_id(self.update.id)
}
pub fn reply_media_group<T>(&self, media_group: T) -> R::SendMediaGroup
where
T: IntoIterator<Item = InputMedia>,
{
self.requester
.send_media_group(self.update.chat.id, media_group)
.reply_to_message_id(self.update.id)
}
pub fn reply_location(&self, latitude: f64, longitude: f64) -> R::SendLocation {
self.requester
.send_location(self.update.chat.id, latitude, longitude)
.reply_to_message_id(self.update.id)
}
pub fn reply_venue<T, U>(
&self,
latitude: f64,
longitude: f64,
title: T,
address: U,
) -> R::SendVenue
where
T: Into<String>,
U: Into<String>,
{
self.requester
.send_venue(self.update.chat.id, latitude, longitude, title, address)
.reply_to_message_id(self.update.id)
}
pub fn reply_video_note(&self, video_note: InputFile) -> R::SendVideoNote {
self.requester
.send_video_note(self.update.chat.id, video_note)
.reply_to_message_id(self.update.id)
}
pub fn reply_contact<T, U>(&self, phone_number: T, first_name: U) -> R::SendContact
where
T: Into<String>,
U: Into<String>,
{
self.requester
.send_contact(self.update.chat.id, phone_number, first_name)
.reply_to_message_id(self.update.id)
}
pub fn reply_sticker(&self, sticker: InputFile) -> R::SendSticker {
self.requester
.send_sticker(self.update.chat.id, sticker)
.reply_to_message_id(self.update.id)
}
pub fn answer_photo(&self, photo: InputFile) -> R::SendPhoto {
self.requester.send_photo(self.update.chat.id, photo)
}
pub fn answer_audio(&self, audio: InputFile) -> R::SendAudio {
self.requester.send_audio(self.update.chat.id, audio)
}
pub fn answer_animation(&self, animation: InputFile) -> R::SendAnimation {
self.requester.send_animation(self.update.chat.id, animation)
}
pub fn answer_document(&self, document: InputFile) -> R::SendDocument {
self.requester.send_document(self.update.chat.id, document)
}
pub fn answer_video(&self, video: InputFile) -> R::SendVideo {
self.requester.send_video(self.update.chat.id, video)
}
pub fn answer_voice(&self, voice: InputFile) -> R::SendVoice {
self.requester.send_voice(self.update.chat.id, voice)
}
pub fn answer_media_group<T>(&self, media_group: T) -> R::SendMediaGroup
where
T: IntoIterator<Item = InputMedia>,
{
self.requester.send_media_group(self.update.chat.id, media_group)
}
pub fn answer_location(&self, latitude: f64, longitude: f64) -> R::SendLocation {
self.requester.send_location(self.update.chat.id, latitude, longitude)
}
pub fn answer_venue<T, U>(
&self,
latitude: f64,
longitude: f64,
title: T,
address: U,
) -> R::SendVenue
where
T: Into<String>,
U: Into<String>,
{
self.requester.send_venue(self.update.chat.id, latitude, longitude, title, address)
}
pub fn answer_video_note(&self, video_note: InputFile) -> R::SendVideoNote {
self.requester.send_video_note(self.update.chat.id, video_note)
}
pub fn answer_contact<T, U>(&self, phone_number: T, first_name: U) -> R::SendContact
where
T: Into<String>,
U: Into<String>,
{
self.requester.send_contact(self.chat_id(), phone_number, first_name)
}
pub fn answer_sticker(&self, sticker: InputFile) -> R::SendSticker {
self.requester.send_sticker(self.update.chat.id, sticker)
}
pub fn forward_to<T>(&self, chat_id: T) -> R::ForwardMessage
where
T: Into<ChatId>,
{
self.requester.forward_message(chat_id, self.update.chat.id, self.update.id)
}
pub fn edit_message_text<T>(&self, text: T) -> R::EditMessageText
where
T: Into<String>,
{
self.requester.edit_message_text(self.update.chat.id, self.update.id, text)
}
pub fn edit_message_caption(&self) -> R::EditMessageCaption {
self.requester.edit_message_caption(self.update.chat.id, self.update.id)
}
pub fn delete_message(&self) -> R::DeleteMessage {
self.requester.delete_message(self.update.chat.id, self.update.id)
}
pub fn pin_message(&self) -> R::PinChatMessage {
self.requester.pin_chat_message(self.update.chat.id, self.update.id)
}
pub fn answer_dice(&self) -> R::SendDice {
self.requester.send_dice(self.update.chat.id)
}
}

20
src/dispatching2/dialogue/get_chat_id.rs
View file

@ -1,20 +0,0 @@
use crate::types::CallbackQuery;
use teloxide_core::types::Message;
/// Something that may has a chat ID.
pub trait GetChatId {
#[must_use]
fn chat_id(&self) -> Option<i64>;
}
impl GetChatId for Message {
fn chat_id(&self) -> Option<i64> {
Some(self.chat.id)
}
}
impl GetChatId for CallbackQuery {
fn chat_id(&self) -> Option<i64> {
self.message.as_ref().map(|mes| mes.chat.id)
}
}

182
src/dispatching2/dialogue/mod.rs
View file

@ -1,182 +0,0 @@
//! Support for user dialogues.
//!
//! The main type is (surprise!) [`Dialogue`]. Under the hood, it is just a
//! wrapper over [`Storage`] and a chat ID. All it does is provides convenient
//! method for manipulating the dialogue state. [`Storage`] is where all
//! dialogue states are stored; it can be either [`InMemStorage`], which is a
//! simple hash map, or database wrappers such as [`SqliteStorage`]. In the
//! latter case, your dialogues are _persistent_, meaning that you can safely
//! restart your bot and all dialogues will remain in the database -- this is a
//! preferred method for production bots.
//!
//! [`examples/dialogue.rs`] clearly demonstrates the typical usage of
//! dialogues. Your dialogue state can be represented as an enumeration:
//!
//! ```ignore
//! #[derive(DialogueState, Clone)]
//! #[handler_out(anyhow::Result<()>)]
//! pub enum State {
//! #[handler(handle_start)]
//! Start,
//!
//! #[handler(handle_receive_full_name)]
//! ReceiveFullName,
//!
//! #[handler(handle_receive_age)]
//! ReceiveAge { full_name: String },
//!
//! #[handler(handle_receive_location)]
//! ReceiveLocation { full_name: String, age: u8 },
//! }
//! ```
//!
//! Each state is associated with its respective handler: e.g., when a dialogue
//! state is `ReceiveAge`, `handle_receive_age` is invoked:
//!
//! ```ignore
//! async fn handle_receive_age(
//! bot: AutoSend<Bot>,
//! msg: Message,
//! dialogue: MyDialogue,
//! (full_name,): (String,), // Available from `State::ReceiveAge`.
//! ) -> anyhow::Result<()> {
//! match msg.text().map(|text| text.parse::<u8>()) {
//! Some(Ok(age)) => {
//! bot.send_message(msg.chat.id, "What's your location?").await?;
//! dialogue.update(State::ReceiveLocation { full_name, age }).await?;
//! }
//! _ => {
//! bot.send_message(msg.chat.id, "Send me a number.").await?;
//! }
//! }
//!
//! Ok(())
//! }
//! ```
//!
//! Variant's fields are passed to state handlers as tuples: `(full_name,):
//! (String,)`. Using [`Dialogue::update`], you can update the dialogue with a
//! new state, in our case -- `State::ReceiveLocation { full_name, age }`. To
//! exit the dialogue, just call [`Dialogue::exit`] and it will be removed from
//! the inner storage:
//!
//! ```ignore
//! async fn handle_receive_location(
//! bot: AutoSend<Bot>,
//! msg: Message,
//! dialogue: MyDialogue,
//! (full_name, age): (String, u8), // Available from `State::ReceiveLocation`.
//! ) -> anyhow::Result<()> {
//! match msg.text() {
//! Some(location) => {
//! let message =
//! format!("Full name: {}\nAge: {}\nLocation: {}", full_name, age, location);
//! bot.send_message(msg.chat.id, message).await?;
//! dialogue.exit().await?;
//! }
//! None => {
//! bot.send_message(msg.chat.id, "Send me a text message.").await?;
//! }
//! }
//!
//! Ok(())
//! }
//! ```
//!
//! [`examples/dialogue.rs`]: https://github.com/teloxide/teloxide/blob/master/examples/dialogue.rs
#[cfg(feature = "redis-storage")]
pub use crate::dispatching::dialogue::{RedisStorage, RedisStorageError};
#[cfg(feature = "sqlite-storage")]
pub use crate::dispatching::dialogue::{SqliteStorage, SqliteStorageError};
pub use crate::dispatching::dialogue::{
serializer, ErasedStorage, InMemStorage, InMemStorageError, Serializer, Storage, TraceStorage,
};
pub use get_chat_id::GetChatId;
use std::{marker::PhantomData, sync::Arc};
mod get_chat_id;
/// A handle for controlling dialogue state.
#[derive(Debug)]
pub struct Dialogue<D, S>
where
S: ?Sized,
{
storage: Arc<S>,
chat_id: i64,
_phantom: PhantomData<D>,
}
// `#[derive]` requires generics to implement `Clone`, but `S` is wrapped around
// `Arc`, and `D` is wrapped around PhantomData.
impl<D, S> Clone for Dialogue<D, S>
where
S: ?Sized,
{
fn clone(&self) -> Self {
Dialogue { storage: self.storage.clone(), chat_id: self.chat_id, _phantom: PhantomData }
}
}
impl<D, S> Dialogue<D, S>
where
D: Send + 'static,
S: Storage<D> + ?Sized,
{
/// Constructs a new dialogue with `storage` (where dialogues are stored)
/// and `chat_id` of a current dialogue.
pub fn new(storage: Arc<S>, chat_id: i64) -> Self {
Self { storage, chat_id, _phantom: PhantomData }
}
/// Retrieves the current state of the dialogue or `None` if there is no
/// dialogue.
pub async fn get(&self) -> Result<Option<D>, S::Error> {
self.storage.clone().get_dialogue(self.chat_id).await
}
/// Like [`Dialogue::get`] but returns a default value if there is no
/// dialogue.
pub async fn get_or_default(&self) -> Result<D, S::Error>
where
D: Default,
{
match self.get().await? {
Some(d) => Ok(d),
None => {
self.storage.clone().update_dialogue(self.chat_id, D::default()).await?;
Ok(D::default())
}
}
}
/// Updates the dialogue state.
///
/// The dialogue type `D` must implement `From<State>` to allow implicit
/// conversion from `State` to `D`.
pub async fn update<State>(&self, state: State) -> Result<(), S::Error>
where
D: From<State>,
{
let new_dialogue = state.into();
self.storage.clone().update_dialogue(self.chat_id, new_dialogue).await?;
Ok(())
}
/// Updates the dialogue with a default value.
pub async fn reset(&self) -> Result<(), S::Error>
where
D: Default,
{
self.update(D::default()).await
}
/// Removes the dialogue from the storage provided to [`Dialogue::new`].
pub async fn exit(&self) -> Result<(), S::Error> {
self.storage.clone().remove_dialogue(self.chat_id).await
}
}

310
src/dispatching2/dispatcher.rs
View file

@ -1,310 +0,0 @@
use crate::{
adaptors::CacheMe,
dispatching::{
stop_token::StopToken, update_listeners, update_listeners::UpdateListener, ShutdownToken,
},
error_handlers::{ErrorHandler, LoggingErrorHandler},
requests::Requester,
types::{AllowedUpdate, Update},
utils::shutdown_token::shutdown_check_timeout_for,
};
use dptree::di::{DependencyMap, DependencySupplier};
use futures::{future::BoxFuture, StreamExt};
use std::{collections::HashSet, fmt::Debug, ops::ControlFlow, sync::Arc};
use teloxide_core::{
requests::{Request, RequesterExt},
types::UpdateKind,
};
use tokio::time::timeout;
use std::future::Future;
/// The builder for [`Dispatcher`].
pub struct DispatcherBuilder<R, Err> {
bot: R,
dependencies: DependencyMap,
handler: UpdateHandler<Err>,
default_handler: DefaultHandler,
error_handler: Arc<dyn ErrorHandler<Err> + Send + Sync>,
}
impl<R, Err> DispatcherBuilder<R, Err>
where
R: Clone + Requester + Clone + Send + Sync + 'static,
Err: Debug + Send + Sync + 'static,
{
/// Specifies a handler that will be called for an unhandled update.
///
/// By default, it is a mere [`log::warn`].
#[must_use]
pub fn default_handler<H, Fut>(self, handler: H) -> Self
where
H: Fn(Arc<Update>) -> Fut + Send + Sync + 'static,
Fut: Future<Output = ()> + Send + 'static,
{
let handler = Arc::new(handler);
Self {
default_handler: Box::new(move |upd| {
let handler = Arc::clone(&handler);
Box::pin(handler(upd))
}),
..self
}
}
/// Specifies a handler that will be called on a handler error.
///
/// By default, it is [`LoggingErrorHandler`].
#[must_use]
pub fn error_handler(self, handler: Arc<dyn ErrorHandler<Err> + Send + Sync>) -> Self {
Self { error_handler: handler, ..self }
}
/// Specifies dependencies that can be used inside of handlers.
///
/// By default, there is no dependencies.
#[must_use]
pub fn dependencies(self, dependencies: DependencyMap) -> Self {
Self { dependencies, ..self }
}
/// Constructs [`Dispatcher`].
#[must_use]
pub fn build(self) -> Dispatcher<R, Err> {
Dispatcher {
bot: self.bot.clone(),
cache_me_bot: self.bot.cache_me(),
dependencies: self.dependencies,
handler: self.handler,
default_handler: self.default_handler,
error_handler: self.error_handler,
allowed_updates: Default::default(),
state: ShutdownToken::new(),
}
}
}
/// The base for update dispatching.
pub struct Dispatcher<R, Err> {
bot: R,
cache_me_bot: CacheMe<R>,
dependencies: DependencyMap,
handler: UpdateHandler<Err>,
default_handler: DefaultHandler,
error_handler: Arc<dyn ErrorHandler<Err> + Send + Sync>,
// TODO: respect allowed_udpates
allowed_updates: HashSet<AllowedUpdate>,
state: ShutdownToken,
}
// TODO: it is allowed to return message as response on telegram request in
// webhooks, so we can allow this too. See more there: https://core.telegram.org/bots/api#making-requests-when-getting-updates
/// A handler that processes updates from Telegram.
pub type UpdateHandler<Err> = dptree::Handler<'static, DependencyMap, Result<(), Err>>;
type DefaultHandler = Box<dyn Fn(Arc<Update>) -> BoxFuture<'static, ()> + Send + Sync>;
impl<R, Err> Dispatcher<R, Err>
where
R: Requester + Clone + Send + Sync + 'static,
Err: Send + Sync + 'static,
{
/// Constructs a new [`DispatcherBuilder`] with `bot` and `handler`.
#[must_use]
pub fn builder(bot: R, handler: UpdateHandler<Err>) -> DispatcherBuilder<R, Err>
where
Err: Debug,
{
DispatcherBuilder {
bot,
dependencies: DependencyMap::new(),
handler,
default_handler: Box::new(|upd| {
log::warn!("Unhandled update: {:?}", upd);
Box::pin(async {})
}),
error_handler: LoggingErrorHandler::new(),
}
}
/// Starts your bot with the default parameters.
///
/// The default parameters are a long polling update listener and log all
/// errors produced by this listener.
///
/// Each time a handler is invoked, [`Dispatcher`] adds the following
/// dependencies (in addition to those passed to
/// [`DispatcherBuilder::dependencies`]):
///
/// - Your bot passed to [`Dispatcher::builder`];
/// - An update from Telegram;
/// - [`crate::types::Me`] (can be used in [`HandlerExt::filter_command`]).
///
/// [`shutdown`]: ShutdownToken::shutdown
/// [a ctrlc signal]: Dispatcher::setup_ctrlc_handler
/// [`HandlerExt::filter_command`]: crate::dispatching2::HandlerExt::filter_command
pub async fn dispatch(&mut self)
where
R: Requester + Clone,
<R as Requester>::GetUpdates: Send,
{
let listener = update_listeners::polling_default(self.bot.clone()).await;
let error_handler =
LoggingErrorHandler::with_custom_text("An error from the update listener");
self.dispatch_with_listener(listener, error_handler).await;
}
/// Starts your bot with custom `update_listener` and
/// `update_listener_error_handler`.
///
/// This method adds the same dependencies as [`Dispatcher::dispatch`].
///
/// [`shutdown`]: ShutdownToken::shutdown
/// [a ctrlc signal]: Dispatcher::setup_ctrlc_handler
pub async fn dispatch_with_listener<'a, UListener, ListenerE, 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,
{
update_listener.hint_allowed_updates(&mut self.allowed_updates.clone().into_iter());
let shutdown_check_timeout = shutdown_check_timeout_for(&update_listener);
let mut stop_token = Some(update_listener.stop_token());
self.state.start_dispatching();
{
let stream = update_listener.as_stream();
tokio::pin!(stream);
loop {
// False positive
#[allow(clippy::collapsible_match)]
if let Ok(upd) = timeout(shutdown_check_timeout, stream.next()).await {
match upd {
None => break,
Some(upd) => self.process_update(upd, &update_listener_error_handler).await,
}
}
if self.state.is_shutting_down() {
if let Some(token) = stop_token.take() {
log::debug!("Start shutting down dispatching...");
token.stop();
break;
}
}
}
}
// TODO: wait for executing handlers?
self.state.done();
}
async fn process_update<LErr, LErrHandler>(
&self,
update: Result<Update, LErr>,
err_handler: &Arc<LErrHandler>,
) where
LErrHandler: ErrorHandler<LErr>,
{
match update {
Ok(upd) => {
if let UpdateKind::Error(err) = upd.kind {
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/issues.",
err,
);
return;
}
let mut deps = self.dependencies.clone();
deps.insert(upd);
deps.insert(self.bot.clone());
deps.insert(
self.cache_me_bot.get_me().send().await.expect("Failed to retrieve 'me'"),
);
match self.handler.dispatch(deps).await {
ControlFlow::Break(Ok(())) => {}
ControlFlow::Break(Err(err)) => {
self.error_handler.clone().handle_error(err).await
}
ControlFlow::Continue(deps) => {
let upd = deps.get();
(self.default_handler)(upd).await;
}
}
}
Err(err) => err_handler.clone().handle_error(err).await,
}
}
/// Setups the `^C` handler that [`shutdown`]s dispatching.
///
/// [`shutdown`]: ShutdownToken::shutdown
#[cfg(feature = "ctrlc_handler")]
pub fn setup_ctrlc_handler(&mut self) -> &mut Self {
let token = self.state.clone();
tokio::spawn(async move {
loop {
tokio::signal::ctrl_c().await.expect("Failed to listen for ^C");
match token.shutdown() {
Ok(f) => {
log::info!("^C received, trying to shutdown the dispatcher...");
f.await;
log::info!("dispatcher is shutdown...");
}
Err(_) => {
log::info!("^C received, the dispatcher isn't running, ignoring the signal")
}
}
}
});
self
}
/// Returns a shutdown token, which can later be used to shutdown
/// dispatching.
pub fn shutdown_token(&self) -> ShutdownToken {
self.state.clone()
}
}
#[cfg(test)]
mod tests {
use std::convert::Infallible;
use teloxide_core::Bot;
use super::*;
#[tokio::test]
async fn test_tokio_spawn() {
tokio::spawn(async {
// Just check that this code compiles.
if false {
Dispatcher::<_, Infallible>::builder(Bot::new(""), dptree::entry())
.build()
.dispatch()
.await;
}
})
.await
.unwrap();
}
}

110
src/dispatching2/mod.rs
View file

@ -1,110 +0,0 @@
//! A new dispatching model based on [`dptree`].
//!
//! In teloxide, updates are dispatched by a pipleine. The central type is
//! [`dptree::Handler`] -- it represents a handler of an update; since the API
//! is highly declarative, you can combine handlers with each other via such
//! methods as [`dptree::Handler::chain`] and [`dptree::Handler::branch`]. The
//! former method pipes one handler to another one, whilst the latter creates a
//! new node, as communicated by the name. For more information, please refer to
//! the documentation of [`dptree`].
//!
//! The pattern itself is called [chain of responsibility], a well-known design
//! technique across OOP developers. But unlike typical object-oriented design,
//! we employ declarative FP-style functions like [`dptree::filter`],
//! [`dptree::filter_map`], and [`dptree::endpoint`]; these functions create
//! special forms of [`dptree::Handler`]; for more information, please refer to
//! their respective documentation. Each of these higher-order functions accept
//! a closure that is made into a handler -- this closure can take any
//! additional parameters, which must be supplied while creating [`Dispatcher`]
//! (see [`DispatcherBuilder::dependencies`]).
//!
//! The [`Dispatcher`] type puts all these things together: it only provides
//! [`Dispatcher::dispatch`] and a handful of other methods. Once you call
//! `.dispatch()`, it will retrieve updates from the Telegram server and pass
//! them to your handler, which is a parameter of [`Dispatcher::builder`].
//!
//! Let us look at a simple example:
//!
//!
//! ([Full](https://github.com/teloxide/teloxide/blob/master/examples/shared_state.rs))
//!
//! ```no_run
//! use std::sync::atomic::{AtomicU64, Ordering};
//!
//! use once_cell::sync::Lazy;
//! use teloxide::prelude2::*;
//!
//! static MESSAGES_TOTAL: Lazy<AtomicU64> = Lazy::new(AtomicU64::default);
//!
//! # #[tokio::main]
//! # async fn main() {
//! pretty_env_logger::init();
//! log::info!("Starting shared_state_bot...");
//!
//! let bot = Bot::from_env().auto_send();
//!
//! let handler = Update::filter_message().branch(dptree::endpoint(
//! |msg: Message, bot: AutoSend<Bot>| async move {
//! let previous = MESSAGES_TOTAL.fetch_add(1, Ordering::Relaxed);
//! bot.send_message(msg.chat.id, format!("I received {} messages in total.", previous))
//! .await?;
//! respond(())
//! },
//! ));
//!
//! Dispatcher::builder(bot, handler).build().setup_ctrlc_handler().dispatch().await;
//! # }
//! ```
//!
//! 1. First, we create the bot: `let bot = Bot::from_env().auto_send()`.
//! 2. Then we construct an update handler. While it is possible to handle all
//! kinds of [`crate::types::Update`], here we are only interested in
//! [`crate::types::Message`]: [`UpdateFilterExt::filter_message`] create a
//! handler object which filters all messages out of a generic update.
//! 3. By doing `.branch(dptree::endpoint(...))`, we set up a custom handling
//! closure that receives `msg: Message` and `bot: AutoSend<Bot>`. There are
//! called dependencies: `msg` is supplied by
//! [`UpdateFilterExt::filter_message`], while `bot` is supplied by
//! [`Dispatcher`].
//!
//! That being said, if we receive a message, the dispatcher will call our
//! handler, but if we receive something other than a message (e.g., a channel
//! post), you will see an unhandled update notice in your terminal.
//!
//! This is a very limited example of update pipelining facilities. In more
//! involved scenarios, there are multiple branches and chains; if one element
//! of a chain fails to handle an update, the update will be passed forwards; if
//! no handler succeeds at handling the update, [`Dispatcher`] will invoke a
//! default handler set up via [`DispatcherBuilder::default_handler`].
//!
//! Update pipelining provides several advantages over the typical `match
//! (update.kind) { ... }` approach:
//!
//! 1. It supports _extension_: e.g., you
//! can define extension filters or some other handlers and then combine them in
//! a single place, thus facilitating loose coupling.
//! 2. Pipelining exhibits a natural syntax for expressing message processing.
//! 3. Lastly, it provides a primitive form of [dependency injection (DI)],
//! which allows you to deal with such objects as a bot and various update types
//! easily.
//!
//! For a more involved example, see [`examples/dispatching2_features.rs`](https://github.com/teloxide/teloxide/blob/master/examples/dispatching2_features.rs).
//!
//! TODO: explain a more involved example with multiple branches.
//!
//! [chain of responsibility]: https://en.wikipedia.org/wiki/Chain-of-responsibility_pattern
//! [dependency injection (DI)]: https://en.wikipedia.org/wiki/Dependency_injection
#[cfg(all(feature = "dispatching2", feature = "ctrlc_handler"))]
pub mod repls;
pub mod dialogue;
mod dispatcher;
mod filter_ext;
mod handler_ext;
mod handler_factory;
pub use dispatcher::{Dispatcher, DispatcherBuilder, UpdateHandler};
pub use filter_ext::{MessageFilterExt, UpdateFilterExt};
pub use handler_ext::HandlerExt;
pub use handler_factory::HandlerFactory;

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

@ -1,102 +0,0 @@
use crate::{
dispatching::{update_listeners, update_listeners::UpdateListener},
dispatching2::{HandlerExt, UpdateFilterExt},
error_handlers::LoggingErrorHandler,
types::Update,
utils::command::BotCommands,
};
use dptree::di::{DependencyMap, Injectable};
use std::{fmt::Debug, marker::PhantomData};
use teloxide_core::requests::Requester;
/// A [REPL] for commands.
///
/// All errors from an update listener and handler will be logged.
///
/// ## Caution
/// **DO NOT** use this function together with [`Dispatcher`] and other REPLs,
/// because Telegram disallow multiple requests at the same time from the same
/// bot.
///
/// ## Dependency requirements
///
/// - Those of [`HandlerExt::filter_command`].
///
/// [REPL]: https://en.wikipedia.org/wiki/Read-eval-print_loop
/// [`Dispatcher`]: crate::dispatching::Dispatcher
#[cfg(feature = "ctrlc_handler")]
pub async fn commands_repl<'a, R, Cmd, H, E, Args>(bot: R, handler: H, cmd: PhantomData<Cmd>)
where
Cmd: BotCommands + Send + Sync + 'static,
H: Injectable<DependencyMap, Result<(), E>, Args> + Send + Sync + 'static,
R: Requester + Clone + Send + Sync + 'static,
<R as Requester>::GetUpdates: Send,
E: Debug + Send + Sync + 'static,
{
let cloned_bot = bot.clone();
commands_repl_with_listener(
bot,
handler,
update_listeners::polling_default(cloned_bot).await,
cmd,
)
.await;
}
/// Like [`commands_repl`], but with a custom [`UpdateListener`].
///
/// All errors from an update listener and handler will be logged.
///
/// ## Caution
/// **DO NOT** use this function together with [`Dispatcher`] and other REPLs,
/// because Telegram disallow multiple requests at the same time from the same
/// bot.
///
/// ## Dependency requirements
///
/// - Those of [`HandlerExt::filter_command`].
///
/// [`Dispatcher`]: crate::dispatching::Dispatcher
/// [`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>(
bot: R,
handler: H,
listener: L,
_cmd: PhantomData<Cmd>,
) where
Cmd: BotCommands + Send + Sync + 'static,
H: Injectable<DependencyMap, Result<(), E>, Args> + Send + Sync + 'static,
L: UpdateListener<ListenerE> + Send + 'a,
ListenerE: Debug + Send + 'a,
R: Requester + Clone + Send + Sync + 'static,
E: Debug + Send + Sync + 'static,
{
use crate::dispatching2::Dispatcher;
// Other update types are of no interest to use since this REPL is only for
// commands. See <https://github.com/teloxide/teloxide/issues/557>.
let ignore_update = |_upd| Box::pin(async {});
let mut dispatcher = Dispatcher::builder(
bot,
Update::filter_message().filter_command::<Cmd>().branch(dptree::endpoint(handler)),
)
.default_handler(ignore_update)
.build();
#[cfg(feature = "ctrlc_handler")]
dispatcher.setup_ctrlc_handler();
// To make mutable var from immutable.
let mut dispatcher = dispatcher;
dispatcher
.dispatch_with_listener(
listener,
LoggingErrorHandler::with_custom_text("An error from the update listener"),
)
.await;
}

96
src/dispatching2/repls/dialogues_repl.rs
View file

@ -1,96 +0,0 @@
use crate::{
dispatching::{
dialogue::{DialogueDispatcher, DialogueStage, DialogueWithCx, InMemStorageError},
update_listeners,
update_listeners::UpdateListener,
Dispatcher, UpdateWithCx,
},
error_handlers::LoggingErrorHandler,
};
use std::{fmt::Debug, future::Future, sync::Arc};
use teloxide_core::{requests::Requester, types::Message};
/// A [REPL] for dialogues.
///
/// All errors from an update listener and handler will be logged. This function
/// uses [`InMemStorage`].
///
/// # Caution
/// **DO NOT** use this function together with [`Dispatcher`] and other REPLs,
/// because Telegram disallow multiple requests at the same time from the same
/// bot.
///
/// [REPL]: https://en.wikipedia.org/wiki/Read-eval-print_loop
/// [`Dispatcher`]: crate::dispatching::Dispatcher
/// [`InMemStorage`]: crate::dispatching::dialogue::InMemStorage
#[cfg(feature = "ctrlc_handler")]
pub async fn dialogues_repl<'a, R, H, D, Fut>(requester: R, handler: H)
where
H: Fn(UpdateWithCx<R, Message>, D) -> Fut + Send + Sync + 'static,
D: Clone + Default + Send + 'static,
Fut: Future<Output = DialogueStage<D>> + Send + 'static,
R: Requester + Send + Clone + 'static,
<R as Requester>::GetUpdatesFaultTolerant: Send,
{
let cloned_requester = requester.clone();
dialogues_repl_with_listener(
requester,
handler,
update_listeners::polling_default(cloned_requester).await,
)
.await;
}
/// Like [`dialogues_repl`], but with a custom [`UpdateListener`].
///
/// All errors from an update listener and handler will be logged. This function
/// uses [`InMemStorage`].
///
/// # Caution
/// **DO NOT** use this function together with [`Dispatcher`] and other REPLs,
/// because Telegram disallow multiple requests at the same time from the same
/// bot.
///
/// [`Dispatcher`]: crate::dispatching::Dispatcher
/// [`dialogues_repl`]: crate::dispatching::repls::dialogues_repl()
/// [`UpdateListener`]: crate::dispatching::update_listeners::UpdateListener
/// [`InMemStorage`]: crate::dispatching::dialogue::InMemStorage
#[cfg(feature = "ctrlc_handler")]
pub async fn dialogues_repl_with_listener<'a, R, H, D, Fut, L, ListenerE>(
requester: R,
handler: H,
listener: L,
) where
H: Fn(UpdateWithCx<R, Message>, D) -> Fut + Send + Sync + 'static,
D: Clone + Default + Send + 'static,
Fut: Future<Output = DialogueStage<D>> + Send + 'static,
L: UpdateListener<ListenerE> + Send + 'a,
ListenerE: Debug + Send + 'a,
R: Requester + Send + Clone + 'static,
{
let handler = Arc::new(handler);
Dispatcher::new(requester)
.messages_handler(DialogueDispatcher::new(
move |DialogueWithCx { cx, dialogue }: DialogueWithCx<
R,
Message,
D,
InMemStorageError,
>| {
let handler = Arc::clone(&handler);
async move {
let dialogue = dialogue.expect("std::convert::Infallible");
handler(cx, dialogue).await
}
},
))
.setup_ctrlc_handler()
.dispatch_with_listener(
listener,
LoggingErrorHandler::with_custom_text("An error from the update listener"),
)
.await;
}

8
src/dispatching2/repls/mod.rs
View file

@ -1,8 +0,0 @@
//! REPLs for dispatching updates.
mod commands_repl;
mod repl;
pub use commands_repl::{commands_repl, commands_repl_with_listener};
//pub use dialogues_repl::{dialogues_repl, dialogues_repl_with_listener};
pub use repl::{repl, repl_with_listener};

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

@ -1,78 +0,0 @@
use crate::{
dispatching::{update_listeners, update_listeners::UpdateListener},
dispatching2::UpdateFilterExt,
error_handlers::{LoggingErrorHandler, OnError},
types::Update,
};
use dptree::di::{DependencyMap, Injectable};
use std::fmt::Debug;
use teloxide_core::requests::Requester;
/// A [REPL] for messages.
///
/// All errors from an update listener and a handler will be logged.
///
/// # Caution
/// **DO NOT** use this function together with [`Dispatcher`] and other REPLs,
/// because Telegram disallow multiple requests at the same time from the same
/// bot.
///
/// [REPL]: https://en.wikipedia.org/wiki/Read-eval-print_loop
/// [`Dispatcher`]: crate::dispatching::Dispatcher
#[cfg(feature = "ctrlc_handler")]
pub async fn repl<R, H, E, Args>(bot: R, handler: H)
where
H: Injectable<DependencyMap, Result<(), E>, Args> + Send + Sync + 'static,
Result<(), E>: OnError<E>,
E: Debug + Send + Sync + 'static,
R: Requester + Send + Sync + Clone + 'static,
<R as Requester>::GetUpdates: Send,
{
let cloned_bot = bot.clone();
repl_with_listener(bot, handler, update_listeners::polling_default(cloned_bot).await).await;
}
/// Like [`repl`], but with a custom [`UpdateListener`].
///
/// All errors from an update listener and handler will be logged.
///
/// # Caution
/// **DO NOT** use this function together with [`Dispatcher`] and other REPLs,
/// because Telegram disallow multiple requests at the same time from the same
/// bot.
///
/// [`Dispatcher`]: crate::dispatching::Dispatcher
/// [`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)
where
H: Injectable<DependencyMap, Result<(), E>, Args> + Send + Sync + 'static,
L: UpdateListener<ListenerE> + Send + 'a,
ListenerE: Debug,
Result<(), E>: OnError<E>,
E: Debug + Send + Sync + 'static,
R: Requester + Clone + Send + Sync + 'static,
{
use crate::dispatching2::Dispatcher;
// Other update types are of no interest to use since this REPL is only for
// messages. See <https://github.com/teloxide/teloxide/issues/557>.
let ignore_update = |_upd| Box::pin(async {});
#[allow(unused_mut)]
let mut dispatcher =
Dispatcher::builder(bot, Update::filter_message().branch(dptree::endpoint(handler)))
.default_handler(ignore_update)
.build();
#[cfg(feature = "ctrlc_handler")]
dispatcher.setup_ctrlc_handler();
dispatcher
.dispatch_with_listener(
listener,
LoggingErrorHandler::with_custom_text("An error from the update listener"),
)
.await;
}

18
src/lib.rs
View file

@ -6,7 +6,7 @@
//!
//! ([Full](https://github.com/teloxide/teloxide/blob/master/examples/dices.rs))
//! ```no_run
//! use teloxide::prelude2::*;
//! use teloxide::prelude::*;
//!
//! # #[tokio::main]
//! # async fn main() {
@ -15,7 +15,7 @@
//!
//! let bot = Bot::from_env().auto_send();
//!
//! teloxide::repls2::repl(bot, |message: Message, bot: AutoSend<Bot>| async move {
//! teloxide::repl(bot, |message: Message, bot: AutoSend<Bot>| async move {
//! bot.send_dice(message.chat.id).await?;
//! respond(())
//! })
@ -63,23 +63,14 @@
#[cfg(feature = "ctrlc_handler")]
pub use dispatching::repls::{
commands_repl, commands_repl_with_listener, dialogues_repl, dialogues_repl_with_listener, repl,
repl_with_listener,
commands_repl, commands_repl_with_listener, repl, repl_with_listener,
};
#[cfg(all(feature = "dispatching2", feature = "ctrlc_handler"))]
pub use dispatching2::repls as repls2;
mod logging;
// Things from this module is also used for the dispatching2 module.
pub mod dispatching;
#[cfg(feature = "dispatching2")]
pub mod dispatching2;
pub mod error_handlers;
pub mod prelude;
#[cfg(feature = "dispatching2")]
pub mod prelude2;
pub mod utils;
#[doc(inline)]
@ -88,10 +79,7 @@ pub use teloxide_core::*;
#[cfg(feature = "macros")]
pub use teloxide_macros as macros;
#[cfg(feature = "dispatching2")]
pub use dptree;
#[cfg(feature = "macros")]
pub use teloxide_macros::teloxide;
#[cfg(all(feature = "nightly", doctest))]
#[cfg_attr(feature = "nightly", cfg_attr(feature = "nightly", doc = include_str!("../README.md")))]

21
src/prelude.rs
View file

@ -1,27 +1,17 @@
//! Commonly used items.
#![deprecated(note = "Use dispatching2 instead")]
#![allow(deprecated)]
pub use crate::{
error_handlers::{LoggingErrorHandler, OnError},
respond,
};
pub use crate::dispatching::{
dialogue::{
exit, next, DialogueDispatcher, DialogueStage, DialogueWithCx, GetChatId, Transition,
TransitionIn, TransitionOut,
},
Dispatcher, DispatcherHandlerRx, DispatcherHandlerRxExt, UpdateWithCx,
dialogue::Dialogue, Dispatcher, HandlerExt as _, MessageFilterExt as _, UpdateFilterExt as _,
};
#[cfg(feature = "macros")]
pub use crate::teloxide;
pub use teloxide_core::types::{
CallbackQuery, ChatMemberUpdated, ChosenInlineResult, InlineQuery, Message, Poll, PollAnswer,
PreCheckoutQuery, ShippingQuery,
PreCheckoutQuery, ShippingQuery, Update,
};
#[cfg(feature = "auto-send")]
@ -30,9 +20,4 @@ pub use crate::adaptors::AutoSend;
#[doc(no_inline)]
pub use teloxide_core::prelude::*;
#[cfg(feature = "frunk")]
pub use crate::utils::UpState;
pub use tokio::sync::mpsc::UnboundedReceiver;
pub use futures::StreamExt;
pub use dptree::{self, prelude::*};

26
src/prelude2.rs
View file

@ -1,26 +0,0 @@
//! Commonly used items (`dispatching2`).
pub use crate::{
error_handlers::{LoggingErrorHandler, OnError},
respond,
};
pub use crate::dispatching2::{
dialogue::Dialogue, Dispatcher, HandlerExt as _, MessageFilterExt as _, UpdateFilterExt as _,
};
#[cfg(feature = "macros")]
pub use crate::teloxide;
pub use teloxide_core::types::{
CallbackQuery, ChatMemberUpdated, ChosenInlineResult, InlineQuery, Message, Poll, PollAnswer,
PreCheckoutQuery, ShippingQuery, Update,
};
#[cfg(feature = "auto-send")]
pub use crate::adaptors::AutoSend;
#[doc(no_inline)]
pub use teloxide_core::prelude::*;
pub use dptree::{self, prelude::*};

1
src/utils/mod.rs
View file

@ -4,7 +4,6 @@ pub mod command;
pub mod html;
pub mod markdown;
pub(crate) mod shutdown_token;
mod up_state;
pub use teloxide_core::net::client_from_env;

25
src/utils/up_state.rs
View file

@ -1,25 +0,0 @@
#![cfg(feature = "frunk")]
use frunk::{from_generic, generic::Generic, hlist::h_cons, into_generic, HCons, HNil};
use std::ops::Add;
/// Constructs a structure from another structure and a field.
///
/// Let `X` be a structure of `field1, ..., fieldN`, `Y` be `field1, ...,
/// fieldN, fieldN+1`. Both `X` and `Y` implement [`Generic`]. Then `Y::up(x,
/// fieldN+1)` constructs `Y` from all the fields of `x: X` plus `Y`'s
/// `fieldN+1`.
///
/// [`Generic`]: https://docs.rs/frunk/latest/frunk/generic/trait.Generic.html
pub trait UpState: Sized {
fn up<Src, F>(src: Src, field: F) -> Self
where
Src: Generic,
Self: Generic<Repr = <<Src as Generic>::Repr as Add<HCons<F, HNil>>>::Output>,
<Src as Generic>::Repr: Add<HCons<F, HNil>>,
{
from_generic(into_generic(src) + h_cons(field, HNil))
}
}
impl<Dst> UpState for Dst {}