Merge pull request #27 from async-telegram-bot/file_download

File download & small fixes

Cargo.toml

@@ -14,3 +14,4 @@ apply = "0.2.2"
 derive_more = "0.15.0"
 tokio = "0.2.0-alpha.4"
 bytes = "0.4.12"
+futures-preview = "0.3.0-alpha.18"

src/bot/mod.rs

@@ -1,5 +1,7 @@
 use reqwest::r#async::Client;
 
+use crate::core::network::{download_file, download_file_stream};
+use crate::core::requests::get_file::GetFile;
 use crate::core::{
     requests::{
         edit_message_live_location::EditMessageLiveLocation,
@@ -11,6 +13,10 @@ use crate::core::{
     },
     types::{InputFile, InputMedia},
 };
+use crate::DownloadError;
+use reqwest::r#async::Chunk;
+use tokio::io::AsyncWrite;
+use tokio::stream::Stream;
 
 pub struct Bot {
     token: String,
@@ -42,6 +48,62 @@ impl Bot {
 
 /// Telegram functions
 impl Bot {
+    /// Download file from telegram into `destination`.
+    /// `path` can be obtained from [`get_file`] method.
+    ///
+    /// For downloading as Stream of Chunks see [`download_file_stream`].
+    ///
+    /// ## Examples
+    ///
+    /// ```no_run
+    /// use async_telegram_bot::{
+    ///     bot::Bot,
+    ///     core::{requests::Request, types::File as TgFile},
+    /// };
+    /// use tokio::fs::File;
+    /// # use async_telegram_bot::core::requests::RequestError;
+    ///
+    /// # async fn run() -> Result<(), Box<dyn std::error::Error>> {
+    /// let bot = Bot::new("TOKEN");
+    /// let mut file = File::create("/home/waffle/Pictures/test.png").await?;
+    ///
+    /// let TgFile { file_path, .. } = bot.get_file("*file_id*").send().await?;
+    /// bot.download_file(&file_path, &mut file).await?;
+    /// # Ok(()) }
+    /// ```
+    ///
+    /// [`get_file`]: crate::bot::Bot::get_file
+    /// [`download_file_stream`]: crate::bot::Bot::download_file_stream
+    pub async fn download_file<D>(
+        &self,
+        path: &str,
+        destination: &mut D,
+    ) -> Result<(), DownloadError>
+    where
+        D: AsyncWrite + Unpin,
+    {
+        download_file(&self.client, &self.token, path, destination).await
+    }
+
+    /// Download file from telegram.
+    ///
+    /// `path` can be obtained from [`get_file`] method.
+    ///
+    /// For downloading into [`AsyncWrite`] (e.g. [`tokio::fs::File`])
+    /// see  [`download_file`].
+    ///
+    /// [`get_file`]: crate::bot::Bot::get_file
+    /// [`AsyncWrite`]: tokio::io::AsyncWrite
+    /// [`tokio::fs::File`]: tokio::fs::File
+    /// [`download_file`]: crate::bot::Bot::download_file
+    pub async fn download_file_stream(
+        &self,
+        path: &str,
+    ) -> Result<impl Stream<Item = Result<Chunk, reqwest::Error>>, reqwest::Error>
+    {
+        download_file_stream(&self.client, &self.token, path).await
+    }
+
     pub fn get_me(&self) -> GetMe {
         GetMe::new(self.ctx())
     }
@@ -135,4 +197,11 @@ impl Bot {
     pub fn stop_message_live_location(&self) -> StopMessageLiveLocation {
         StopMessageLiveLocation::new(self.ctx())
     }
+
+    pub fn get_file<F>(&self, file_id: F) -> GetFile
+    where
+        F: Into<String>,
+    {
+        GetFile::new(self.ctx(), file_id.into())
+    }
 }

src/core/mod.rs

@@ -1,3 +1,3 @@
-mod network;
+pub(crate) mod network;
 pub mod requests;
 pub mod types;

src/core/network/mod.rs

@@ -3,12 +3,19 @@ use crate::core::{
     types::ResponseParameters,
 };
 
+use crate::DownloadError;
 use apply::Apply;
+use bytes::Buf;
+use futures::StreamExt;
+use reqwest::r#async::Chunk;
 use reqwest::{
     r#async::{multipart::Form, Client, Response},
     StatusCode,
 };
 use serde::{de::DeserializeOwned, Serialize};
+use tokio::io::AsyncWriteExt;
+use tokio::prelude::AsyncWrite;
+use tokio::stream::Stream;
 
 const TELEGRAM_API_URL: &str = "https://api.telegram.org";
 
@@ -81,18 +88,7 @@ async fn process_response<T: DeserializeOwned>(
     )
     .map_err(RequestError::InvalidJson)?;
 
-    match response {
+    response.into()
-        TelegramResponse::Ok { result, .. } => Ok(result),
-        TelegramResponse::Err {
-            description,
-            error_code,
-            response_parameters: _,
-            ..
-        } => Err(RequestError::ApiError {
-            description,
-            status_code: StatusCode::from_u16(error_code).unwrap(),
-        }),
-    }
 }
 
 #[derive(Deserialize)]
@@ -116,6 +112,64 @@ enum TelegramResponse<R> {
     },
 }
 
+pub async fn download_file<D>(
+    client: &Client,
+    token: &str,
+    path: &str,
+    destination: &mut D,
+) -> Result<(), DownloadError>
+where
+    D: AsyncWrite + Unpin,
+{
+    let mut stream = download_file_stream(client, token, path).await?;
+
+    while let Some(chunk) = stream.next().await {
+        destination.write_all(chunk?.bytes()).await?;
+    }
+
+    Ok(())
+}
+
+pub(crate) async fn download_file_stream(
+    client: &Client,
+    token: &str,
+    path: &str,
+) -> Result<impl Stream<Item = Result<Chunk, reqwest::Error>>, reqwest::Error> {
+    let url = file_url(TELEGRAM_API_URL, token, path);
+    let resp = client.get(&url).send().await?.error_for_status()?;
+    Ok(resp.into_body())
+}
+
+impl<R> Into<ResponseResult<R>> for TelegramResponse<R> {
+    fn into(self) -> Result<R, RequestError> {
+        match self {
+            TelegramResponse::Ok { result, .. } => Ok(result),
+            TelegramResponse::Err {
+                description,
+                error_code,
+                response_parameters,
+                ..
+            } => {
+                if let Some(params) = response_parameters {
+                    match params {
+                        ResponseParameters::RetryAfter(i) => {
+                            Err(RequestError::RetryAfter(i))
+                        }
+                        ResponseParameters::MigrateToChatId(to) => {
+                            Err(RequestError::MigrateToChatId(to))
+                        }
+                    }
+                } else {
+                    Err(RequestError::ApiError {
+                        description,
+                        status_code: StatusCode::from_u16(error_code).unwrap(),
+                    })
+                }
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/core/requests/get_file.rs

@@ -12,11 +12,11 @@ use crate::core::types::File;
 /// It is guaranteed that the link will be valid for at least 1 hour.
 /// When the link expires, a new one can be requested by calling getFile again.
 #[derive(Debug, Clone, Serialize)]
-struct GetFile<'a> {
+pub struct GetFile<'a> {
     #[serde(skip_serializing)]
     ctx: RequestContext<'a>,
     /// File identifier to get info about
-    file_id: String,
+    pub file_id: String,
 }
 
 impl<'a> Request<'a> for GetFile<'a> {
@@ -36,6 +36,10 @@ impl<'a> Request<'a> for GetFile<'a> {
 }
 
 impl<'a> GetFile<'a> {
+    pub(crate) fn new(ctx: RequestContext<'a>, file_id: String) -> Self {
+        Self { ctx, file_id }
+    }
+
     pub fn file_id<T>(mut self, file_id: T) -> Self
     where
         T: Into<String>,

src/core/requests/mod.rs

@@ -11,11 +11,20 @@ mod utils;
 pub enum RequestError {
     #[display(fmt = "Telegram error #{}: {}", status_code, description)]
     ApiError {
-        // TODO: add response parameters
         status_code: StatusCode,
         description: String,
     },
 
+    /// The group has been migrated to a supergroup with the specified
+    /// identifier.
+    #[display(fmt = "The group has been migrated to a supergroup with id {id}", id = _0)]
+    MigrateToChatId(i64),
+
+    /// In case of exceeding flood control, the number of seconds left to wait
+    /// before the request can be repeated
+    #[display(fmt = "Retry after {secs} seconds", secs = _0)]
+    RetryAfter(i32),
+
     #[display(fmt = "Network error: {err}", err = _0)]
     NetworkError(reqwest::Error),
 
@@ -27,6 +36,8 @@ impl std::error::Error for RequestError {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         match self {
             RequestError::ApiError { .. } => None,
+            RequestError::MigrateToChatId(_) => None,
+            RequestError::RetryAfter(_) => None,
             RequestError::NetworkError(err) => Some(err),
             RequestError::InvalidJson(err) => Some(err),
         }

src/core/requests/send_chat_action.rs

@@ -8,7 +8,7 @@ use crate::core::requests::{
 /// arrives from your bot, Telegram clients clear its typing status).
 /// Returns True on success.
 #[derive(Debug, Clone, Serialize)]
-struct SendChatAction<'a> {
+pub struct SendChatAction<'a> {
     #[serde(skip_serializing)]
     ctx: RequestContext<'a>,
     /// Unique identifier for the target chat or
@@ -24,7 +24,7 @@ struct SendChatAction<'a> {
 
 #[derive(Debug, Serialize, From, Clone)]
 #[serde(rename_all = "snake_case")]
-enum ChatAction {
+pub enum ChatAction {
     Typing,
     UploadPhoto,
     RecordVideo,

src/core/requests/send_contact.rs

@@ -7,7 +7,7 @@ use crate::core::types::{Message, ReplyMarkup};
 /// Use this method to send phone contacts.
 /// returned.
 #[derive(Debug, Clone, Serialize)]
-struct SendContact<'a> {
+pub struct SendContact<'a> {
     #[serde(skip_serializing)]
     ctx: RequestContext<'a>,
     /// Unique identifier for the target chat or

src/core/requests/send_poll.rs

@@ -7,7 +7,7 @@ use crate::core::types::{Message, ReplyMarkup};
 /// Use this method to send a native poll. A native poll can't be sent to a
 /// private chat. On success, the sent Message is returned.
 #[derive(Debug, Clone, Serialize)]
-struct SendPoll<'a> {
+pub struct SendPoll<'a> {
     #[serde(skip_serializing)]
     ctx: RequestContext<'a>,
     /// identifier for the target chat or username of the target channel (in

src/core/requests/send_venue.rs

@@ -7,7 +7,7 @@ use crate::core::types::{Message, ReplyMarkup};
 /// Use this method to send information about a venue.
 /// Message is returned.
 #[derive(Debug, Clone, Serialize)]
-struct SendVenue<'a> {
+pub struct SendVenue<'a> {
     #[serde(skip_serializing)]
     ctx: RequestContext<'a>,
     /// Unique identifier for the target chat or
@@ -62,7 +62,7 @@ impl<'a> Request<'a> for SendVenue<'a> {
 }
 
 impl<'a> SendVenue<'a> {
-    pub fn new(
+    pub(crate) fn new(
         ctx: RequestContext<'a>,
         chat_id: ChatId,
         latitude: f64,

src/core/types/inline_query_result_contact.rs

@@ -1,6 +1,4 @@
-use crate::core::types::{
+use crate::core::types::{InlineKeyboardMarkup, InputMessageContent};
-    InlineKeyboardMarkup, InputMessageContent, ParseMode,
-};
 
 #[derive(Debug, Serialize, PartialEq, Clone)]
 pub struct InlineQueryResultContact {

src/core/types/inline_query_result_game.rs

@@ -1,6 +1,4 @@
-use crate::core::types::{
+use crate::core::types::InlineKeyboardMarkup;
-    InlineKeyboardMarkup, InputMessageContent, ParseMode,
-};
 
 #[derive(Debug, Serialize, Hash, PartialEq, Eq, Clone)]
 pub struct InlineQueryResultGame {

src/core/types/inline_query_result_location.rs

@@ -1,6 +1,4 @@
-use crate::core::types::{
+use crate::core::types::{InlineKeyboardMarkup, InputMessageContent};
-    InlineKeyboardMarkup, InputMessageContent, ParseMode,
-};
 
 #[derive(Debug, Serialize, PartialEq, Clone)]
 pub struct InlineQueryResultLocation {

src/core/types/inline_query_result_photo.rs

@@ -2,6 +2,9 @@ use crate::core::types::{
     InlineKeyboardMarkup, InputMessageContent, ParseMode,
 };
 
+/// Represents a link to a photo. By default, this photo will be sent by the
+/// user with optional caption. Alternatively, you can use input_message_content
+/// to send a message with the specified content instead of the photo.
 #[derive(Debug, Serialize, PartialEq, Clone)]
 pub struct InlineQueryResultPhoto {
     pub id: String,

src/core/types/inline_query_result_venue.rs

@@ -1,6 +1,4 @@
-use crate::core::types::{
+use crate::core::types::{InlineKeyboardMarkup, InputMessageContent};
-    InlineKeyboardMarkup, InputMessageContent, ParseMode,
-};
 
 #[derive(Debug, Serialize, PartialEq, Clone)]
 pub struct InlineQueryResultVenue {

src/errors.rs

@@ -0,0 +1,17 @@
+#[derive(Debug, Display, From)]
+pub enum DownloadError {
+    #[display(fmt = "Network error: {err}", err = _0)]
+    NetworkError(reqwest::Error),
+
+    #[display(fmt = "IO Error: {err}", err = _0)]
+    Io(std::io::Error),
+}
+
+impl std::error::Error for DownloadError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            DownloadError::NetworkError(err) => Some(err),
+            DownloadError::Io(err) => Some(err),
+        }
+    }
+}

src/lib.rs

@@ -5,3 +5,6 @@ extern crate serde;
 
 pub mod bot;
 pub mod core;
+pub mod errors;
+
+pub use errors::DownloadError;