Add Redis storage & example bot using it

Cargo.toml

@@ -23,7 +23,10 @@ authors = [
 [badges]
 maintenance = { status = "actively-developed" }
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+[features]
+redis-storage = ["redis"]
+cbor-serializer = ["serde_cbor"]
+bincode-serializer = ["bincode"]
 
 [dependencies]
 serde_json = "1.0.44"
@@ -45,6 +48,10 @@ futures = "0.3.1"
 pin-project = "0.4.6"
 serde_with_macros = "1.0.1"
 
+redis = { version = "0.15.1", optional = true }
+serde_cbor = { version = "0.11.1", optional = true }
+bincode = { version = "1.2.1", optional = true }
+
 teloxide-macros = "0.2.1"
 
 [dev-dependencies]

examples/dialogue_bot/src/main.rs

@@ -41,7 +41,7 @@ enum FavouriteMusic {
 
 impl FavouriteMusic {
     fn markup() -> ReplyKeyboardMarkup {
-        ReplyKeyboardMarkup::default().append_row(vec![
+        ReplyKeyboardMarkup::default().one_time_keyboard(true).append_row(vec![
             KeyboardButton::new("Rock"),
             KeyboardButton::new("Metal"),
             KeyboardButton::new("Pop"),

examples/dialogue_bot_redis/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "dialogue_bot_redis"
+version = "0.1.0"
+authors = ["Temirkhan Myrzamadi <hirrolot@gmail.com>"]
+edition = "2018"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+log = "0.4.8"
+tokio = "0.2.9"
+pretty_env_logger = "0.4.0"
+smart-default = "0.6.0"
+parse-display = "0.1.1"
+# You can also choose "cbor-serializer" or built-in JSON serializer
+teloxide = { path = "../../", features = ["redis-storage", "bincode-serializer"] }
+serde = "1.0.104"
+
+[profile.release]
+lto = true

examples/dialogue_bot_redis/src/main.rs

@@ -0,0 +1,210 @@
+// This is a bot that asks your full name, your age, your favourite kind of
+// music and sends all the gathered information back.
+//
+// # Example
+// ```
+//  - Let's start! First, what's your full name?
+//  - Luke Skywalker
+//  - What a wonderful name! Your age?
+//  - 26
+//  - Good. Now choose your favourite music
+// *A keyboard of music kinds is displayed*
+// *You select Metal*
+//  - Metal
+//  - Fine. Your full name: Luke Skywalker, your age: 26, your favourite music: Metal
+// ```
+
+#![allow(clippy::trivial_regex)]
+
+#[macro_use]
+extern crate smart_default;
+
+use serde::{Deserialize, Serialize};
+use std::sync::Arc;
+
+use teloxide::{
+    dispatching::dialogue::{RedisStorage, Serializer, Storage},
+    prelude::*,
+    types::{KeyboardButton, ReplyKeyboardMarkup},
+};
+
+use parse_display::{Display, FromStr};
+
+// ============================================================================
+// [Favourite music kinds]
+// ============================================================================
+
+#[derive(Copy, Clone, Display, FromStr)]
+enum FavouriteMusic {
+    Rock,
+    Metal,
+    Pop,
+    Other,
+}
+
+impl FavouriteMusic {
+    fn markup() -> ReplyKeyboardMarkup {
+        ReplyKeyboardMarkup::default().one_time_keyboard(true).append_row(vec![
+            KeyboardButton::new("Rock"),
+            KeyboardButton::new("Metal"),
+            KeyboardButton::new("Pop"),
+            KeyboardButton::new("Other"),
+        ])
+    }
+}
+
+// ============================================================================
+// [A type-safe finite automaton]
+// ============================================================================
+
+#[derive(Clone, Serialize, Deserialize)]
+struct ReceiveAgeState {
+    full_name: String,
+}
+
+#[derive(Clone, Serialize, Deserialize)]
+struct ReceiveFavouriteMusicState {
+    data: ReceiveAgeState,
+    age: u8,
+}
+
+#[derive(Display)]
+#[display(
+    "Your full name: {data.data.full_name}, your age: {data.age}, your \
+     favourite music: {favourite_music}"
+)]
+struct ExitState {
+    data: ReceiveFavouriteMusicState,
+    favourite_music: FavouriteMusic,
+}
+
+#[derive(SmartDefault, Serialize, Deserialize)]
+enum Dialogue {
+    #[default]
+    Start,
+    ReceiveFullName,
+    ReceiveAge(ReceiveAgeState),
+    ReceiveFavouriteMusic(ReceiveFavouriteMusicState),
+}
+
+// ============================================================================
+// [Control a dialogue]
+// ============================================================================
+
+type Cx<State> = DialogueDispatcherHandlerCx<
+    Message,
+    State,
+    <RedisStorage as Storage<Dialogue>>::Error,
+>;
+type Res = ResponseResult<DialogueStage<Dialogue>>;
+
+async fn start(cx: Cx<()>) -> Res {
+    cx.answer("Let's start! First, what's your full name?").send().await?;
+    next(Dialogue::ReceiveFullName)
+}
+
+async fn full_name(cx: Cx<()>) -> Res {
+    match cx.update.text() {
+        None => {
+            cx.answer("Please, send me a text message!").send().await?;
+            next(Dialogue::ReceiveFullName)
+        }
+        Some(full_name) => {
+            cx.answer("What a wonderful name! Your age?").send().await?;
+            next(Dialogue::ReceiveAge(ReceiveAgeState {
+                full_name: full_name.to_owned(),
+            }))
+        }
+    }
+}
+
+async fn age(cx: Cx<ReceiveAgeState>) -> Res {
+    match cx.update.text().unwrap().parse() {
+        Ok(age) => {
+            cx.answer("Good. Now choose your favourite music:")
+                .reply_markup(FavouriteMusic::markup())
+                .send()
+                .await?;
+            next(Dialogue::ReceiveFavouriteMusic(ReceiveFavouriteMusicState {
+                data: cx.dialogue.unwrap(),
+                age,
+            }))
+        }
+        Err(_) => {
+            cx.answer("Oh, please, enter a number!").send().await?;
+            next(Dialogue::ReceiveAge(cx.dialogue.unwrap()))
+        }
+    }
+}
+
+async fn favourite_music(cx: Cx<ReceiveFavouriteMusicState>) -> Res {
+    match cx.update.text().unwrap().parse() {
+        Ok(favourite_music) => {
+            cx.answer(format!(
+                "Fine. {}",
+                ExitState {
+                    data: cx.dialogue.as_ref().unwrap().clone(),
+                    favourite_music
+                }
+            ))
+            .send()
+            .await?;
+            exit()
+        }
+        Err(_) => {
+            cx.answer("Oh, please, enter from the keyboard!").send().await?;
+            next(Dialogue::ReceiveFavouriteMusic(cx.dialogue.unwrap()))
+        }
+    }
+}
+
+async fn handle_message(cx: Cx<Dialogue>) -> Res {
+    let DialogueDispatcherHandlerCx { bot, update, dialogue } = cx;
+    match dialogue.unwrap() {
+        Dialogue::Start => {
+            start(DialogueDispatcherHandlerCx::new(bot, update, ())).await
+        }
+        Dialogue::ReceiveFullName => {
+            full_name(DialogueDispatcherHandlerCx::new(bot, update, ())).await
+        }
+        Dialogue::ReceiveAge(s) => {
+            age(DialogueDispatcherHandlerCx::new(bot, update, s)).await
+        }
+        Dialogue::ReceiveFavouriteMusic(s) => {
+            favourite_music(DialogueDispatcherHandlerCx::new(bot, update, s))
+                .await
+        }
+    }
+}
+
+// ============================================================================
+// [Run!]
+// ============================================================================
+
+#[tokio::main]
+async fn main() {
+    run().await;
+}
+
+async fn run() {
+    teloxide::enable_logging!();
+    log::info!("Starting dialogue_bot!");
+
+    let bot = Bot::from_env();
+
+    Dispatcher::new(bot)
+        .messages_handler(DialogueDispatcher::with_storage(
+            |cx| async move {
+                handle_message(cx).await.expect("Something wrong with the bot!")
+            },
+            Arc::new(
+                // You can also choose Serializer::JSON or Serializer::Bincode
+                // All serializer but JSON require enabling feature "serializer-<name>",
+                // e. g. "serializer-cbor" or "serializer-bincode"
+                RedisStorage::open("redis://127.0.0.1:6379", Serializer::CBOR)
+                    .unwrap(),
+            ),
+        ))
+        .dispatch()
+        .await;
+}

src/dispatching/dialogue/mod.rs

@@ -54,4 +54,6 @@ pub use dialogue_dispatcher_handler::DialogueDispatcherHandler;
 pub use dialogue_dispatcher_handler_cx::DialogueDispatcherHandlerCx;
 pub use dialogue_stage::{exit, next, DialogueStage};
 pub use get_chat_id::GetChatId;
-pub use storage::{InMemStorage, Storage};
+#[cfg(feature = "redis-storage")]
+pub use storage::RedisStorage;
+pub use storage::{InMemStorage, Serializer, Storage};

src/dispatching/dialogue/storage/mod.rs

@@ -1,7 +1,16 @@
+pub mod serializer;
+
 mod in_mem_storage;
 
+#[cfg(feature = "redis-storage")]
+mod redis_storage;
+
 use futures::future::BoxFuture;
+
 pub use in_mem_storage::InMemStorage;
+#[cfg(feature = "redis-storage")]
+pub use redis_storage::RedisStorage;
+pub use serializer::Serializer;
 use std::sync::Arc;
 
 /// A storage of dialogues.

src/dispatching/dialogue/storage/redis_storage.rs

@@ -0,0 +1,85 @@
+use super::{
+    serializer::{self, Serializer},
+    Storage,
+};
+use futures::future::BoxFuture;
+use redis::{AsyncCommands, FromRedisValue, IntoConnectionInfo};
+use serde::{de::DeserializeOwned, Serialize};
+use std::sync::Arc;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("{0}")]
+    SerdeError(#[from] serializer::Error),
+    #[error("error from Redis: {0}")]
+    RedisError(#[from] redis::RedisError),
+}
+
+type Result<T, E = Error> = std::result::Result<T, E>;
+
+pub struct RedisStorage {
+    client: redis::Client,
+    serializer: Serializer,
+}
+
+impl RedisStorage {
+    pub fn open(
+        url: impl IntoConnectionInfo,
+        serializer: Serializer,
+    ) -> Result<Self> {
+        Ok(Self { client: redis::Client::open(url)?, serializer })
+    }
+}
+
+impl<D> Storage<D> for RedisStorage
+where
+    D: Send + Serialize + DeserializeOwned + 'static,
+{
+    type Error = Error;
+
+    // `.del().ignore()` is much more readable than `.del()\n.ignore()`
+    #[rustfmt::skip]
+    fn remove_dialogue(
+        self: Arc<Self>,
+        chat_id: i64,
+    ) -> BoxFuture<'static, Result<Option<D>>> {
+        Box::pin(async move {
+            let mut conn = self.client.get_async_connection().await?;
+            let res = redis::pipe()
+                .atomic()
+                .get(chat_id)
+                .del(chat_id).ignore()
+                .query_async::<_, redis::Value>(&mut conn)
+                .await?;
+            // We're expecting `.pipe()` to return us an exactly one result in bulk,
+            // so all other branches should be unreachable
+            match res {
+                redis::Value::Bulk(bulk) if bulk.len() == 1 => {
+                    Ok(
+                        Option::<Vec<u8>>::from_redis_value(&bulk[0])?
+                            .map(|v| self.serializer.deserialize(&v))
+                            .transpose()?
+                    )
+                },
+                _ => unreachable!()
+            }
+        })
+    }
+
+    fn update_dialogue(
+        self: Arc<Self>,
+        chat_id: i64,
+        dialogue: D,
+    ) -> BoxFuture<'static, Result<Option<D>>> {
+        Box::pin(async move {
+            let mut conn = self.client.get_async_connection().await?;
+            let dialogue = self.serializer.serialize(&dialogue)?;
+            Ok(conn
+                .getset::<_, Vec<u8>, Option<Vec<u8>>>(chat_id, dialogue)
+                .await?
+                .map(|d| self.serializer.deserialize(&d))
+                .transpose()?)
+        })
+    }
+}

src/dispatching/dialogue/storage/serializer.rs

@@ -0,0 +1,53 @@
+use serde::{de::DeserializeOwned, ser::Serialize};
+use thiserror::Error;
+use Serializer::*;
+
+#[derive(Debug, Error)]
+pub enum Error {
+    #[error("failed parsing/serializing JSON: {0}")]
+    JSONError(#[from] serde_json::Error),
+    #[cfg(feature = "cbor-serializer")]
+    #[error("failed parsing/serializing CBOR: {0}")]
+    CBORError(#[from] serde_cbor::Error),
+    #[cfg(feature = "bincode-serializer")]
+    #[error("failed parsing/serializing Bincode: {0}")]
+    BincodeError(#[from] bincode::Error),
+}
+
+type Result<T, E = Error> = std::result::Result<T, E>;
+
+pub enum Serializer {
+    JSON,
+    #[cfg(feature = "cbor-serializer")]
+    CBOR,
+    #[cfg(feature = "bincode-serializer")]
+    Bincode,
+}
+
+impl Serializer {
+    pub fn serialize<D>(&self, val: &D) -> Result<Vec<u8>>
+    where
+        D: Serialize,
+    {
+        Ok(match self {
+            JSON => serde_json::to_vec(val)?,
+            #[cfg(feature = "cbor-serializer")]
+            CBOR => serde_cbor::to_vec(val)?,
+            #[cfg(feature = "bincode-serializer")]
+            Bincode => bincode::serialize(val)?,
+        })
+    }
+
+    pub fn deserialize<'de, D>(&self, data: &'de [u8]) -> Result<D>
+    where
+        D: DeserializeOwned,
+    {
+        Ok(match self {
+            JSON => serde_json::from_slice(data)?,
+            #[cfg(feature = "cbor-serializer")]
+            CBOR => serde_cbor::from_slice(data)?,
+            #[cfg(feature = "bincode-serializer")]
+            Bincode => bincode::deserialize(data)?,
+        })
+    }
+}