telexide_fork/client/
stream.rs

1use futures::{Future, Stream};
2use std::{
3    cmp::max,
4    collections::VecDeque,
5    pin::Pin,
6    sync::Arc,
7    task::{Context, Poll},
8};
9
10use super::APIConnector;
11use crate::{
12    api::types::{GetUpdates, UpdateType},
13    model::Update,
14    Result,
15};
16
17type FutureUpdate = Pin<Box<dyn Future<Output = Result<Vec<Update>>>>>;
18
19/// The stream of incoming updates, created by long polling the telegram API
20/// using their getUpdates endpoint.
21///
22/// In most use-cases, this will be handled for you by the [`Client`]
23/// and the new updates then dispatched to your eventhandlers.
24///
25/// ## Example
26/// ```rust,no_run
27/// # use std::sync::Arc;
28/// use futures::StreamExt;
29/// use telexide_fork::{
30///     api::APIClient,
31///     client::UpdatesStream
32/// };
33///
34/// #[tokio::main]
35/// async fn main() {
36///     let token = String::from("test token");
37///
38///     let mut stream = UpdatesStream::new(
39///         Arc::new(
40///             Box::new(
41///                 APIClient::new_default(&token)
42///             )
43///         )
44///     );
45///
46///     while let Some(poll) = stream.next().await {
47///         match poll {
48///             Ok(update) => {
49///                 println!("ID of the update received: {}", update.update_id);
50///             },
51///             Err(err) => return,
52///         }
53///     }
54/// }
55/// ```
56///
57/// [`Client`]: struct.Client.html
58#[must_use = "streams do nothing unless polled"]
59pub struct UpdatesStream {
60    api: Arc<Box<APIConnector>>,
61    buffer: VecDeque<Update>,
62    allowed_updates: Vec<UpdateType>,
63    offset: i64,
64    limit: usize,
65    timeout: usize,
66    current_request: Option<FutureUpdate>,
67}
68
69impl Stream for UpdatesStream {
70    type Item = Result<Update>;
71
72    fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
73        let ref_mut = self.get_mut();
74
75        if let Some(u) = ref_mut.buffer.pop_front() {
76            return Poll::Ready(Some(Ok(u)));
77        }
78
79        if let Some(ref mut request) = ref_mut.current_request {
80            match request.as_mut().poll(cx) {
81                Poll::Pending => return Poll::Pending,
82                Poll::Ready(Ok(ref res)) if res.is_empty() => {
83                    ref_mut.poll_telegram();
84                    return Pin::new(ref_mut).poll_next(cx);
85                }
86                Poll::Ready(Ok(res)) => {
87                    for u in res {
88                        ref_mut.offset = max(u.update_id, ref_mut.offset);
89                        ref_mut.buffer.push_back(u);
90                    }
91                }
92                Poll::Ready(Err(err)) => {
93                    ref_mut.poll_telegram();
94                    return Poll::Ready(Some(Err(err)));
95                }
96            };
97        } else {
98            ref_mut.poll_telegram();
99            return Pin::new(ref_mut).poll_next(cx);
100        }
101
102        ref_mut.current_request = None;
103        Pin::new(ref_mut).poll_next(cx)
104    }
105}
106
107impl UpdatesStream {
108    fn poll_telegram(&mut self) {
109        let mut data = GetUpdates::new();
110        data.set_limit(self.limit)
111            .set_allowed_updates(self.allowed_updates.clone())
112            .set_offset(self.offset + 1)
113            .set_timeout(self.timeout);
114
115        let api = self.api.clone();
116        self.current_request = Some(Box::pin(async move { api.get_updates(data).await }));
117    }
118
119    /// creates a new update stream using the provided [`API`]
120    ///
121    /// [`API`]: ../api/trait.API.html
122    pub fn new(api: Arc<Box<APIConnector>>) -> Self {
123        Self {
124            api,
125            buffer: VecDeque::new(),
126            allowed_updates: Vec::new(),
127            offset: 0,
128            limit: 100,
129            timeout: 5,
130            current_request: None,
131        }
132    }
133
134    /// Sets the maximum amount of updates retrieved in one API call
135    pub fn set_limit(&mut self, limit: usize) -> &mut Self {
136        self.limit = limit;
137        self
138    }
139
140    /// Set the timeout in seconds for long polling. Defaults to 5.
141    /// Should be positive, short polling should be used for testing purposes
142    /// only.
143    pub fn set_timout(&mut self, timeout: usize) -> &mut Self {
144        self.timeout = timeout;
145        self
146    }
147
148    /// Set which update types you want to receive
149    pub fn set_allowed_updates(&mut self, allowed: Vec<UpdateType>) -> &mut Self {
150        self.allowed_updates = allowed;
151        self
152    }
153
154    /// Add an update type to the list of update types you want to receive
155    pub fn add_allowed_updates(&mut self, allowed: UpdateType) -> &mut Self {
156        self.allowed_updates.push(allowed);
157        self
158    }
159
160    /// Remove an update type from the list of update types you want to receive
161    pub fn remove_allowed_updates(&mut self, to_remove: &UpdateType) -> &mut Self {
162        self.allowed_updates.retain(|t| t != to_remove);
163        self
164    }
165}
telexide_fork/client/stream.rs

telexide_fork/client/
stream.rs
