:Allow iteration over endpoints that provide pagination

4 files changed, 146 insertions, 82 deletions

diff --git a/src/bin/main.rs b/src/bin/main.rs
index 81f07ff..982dc9d 100644
--- a/src/bin/main.rs
+++ b/src/bin/main.rs
@@ -5,6 +5,7 @@ extern crate tokio;
 extern crate twitch_api;
 
 use futures::future::Future;
+use futures::Stream;
 use std::env;
 use twitch_api::HelixClient;
 
@@ -35,6 +36,18 @@ fn main() {
             ()
         });
 
+    let videos = authed_client
+        .videos()
+        .by_user("67955580")
+        .take(10)
+        .for_each(|collection| {
+            println!("{:?}", collection);
+            Ok(())
+        })
+        .map(|_| ())
+        .map_err(|err| {println!("{:?}", err); ()});
+
+
     /*
     let clip2 = client.kraken
         .clip(&"EnergeticApatheticTarsierThisIsSparta")
@@ -51,6 +64,7 @@ fn main() {
      */
     //std::mem::drop(authed_client);
     tokio::run(
+        /*
         clip.join(clip2)
             .and_then(|(c1, c2)| {
                 println!("{:?} {:?}", c1, c2);
@@ -67,5 +81,7 @@ fn main() {
             })
             .map(|_| ())
             .map_err(|_| ())
+            */
+        videos
     );
 }
diff --git a/src/helix/mod.rs b/src/helix/mod.rs
index 5daa498..15f2008 100644
--- a/src/helix/mod.rs
+++ b/src/helix/mod.rs
@@ -247,7 +247,7 @@ impl Client {
      * as other requests.
      *
      * Clients created with 'new' are bottom clients and calls
-     * to authenticate stack a authed client on top
+     * to authenticate stack an authed client on top
      */
     fn get_bottom_client(&self) -> Client {
         match self.inner.as_ref() {
@@ -355,6 +355,29 @@ struct RequestRef {
     method: Method,
 }
 
+impl RequestRef {
+    pub fn new(url: String,
+               params: BTreeMap<&str, &str>,
+               client: Client,
+               method: Method,
+               ratelimit: Option<RatelimitKey>,
+               ) -> RequestRef 
+    {
+        let mut owned_params = BTreeMap::new();
+        for (key, value) in params {
+            owned_params.insert(key.to_owned(), value.to_owned());
+        }
+
+        RequestRef {
+            url: url,
+            params: owned_params,
+            client: client,
+            method: method,
+            ratelimit: ratelimit,
+        }
+    }
+}
+
 enum RequestState<T> {
     SetupRequest,
     SetupBarriers,
@@ -381,7 +404,6 @@ enum IterableApiRequestState<T> {
 
 pub struct IterableApiRequest<T> {
     inner: Arc<RequestRef>,
-    pagination: Option<String>,
     state: IterableApiRequestState<T>, 
 }
 
@@ -395,19 +417,8 @@ impl<T: DeserializeOwned + PaginationTrait + 'static + Send> ApiRequest<T> {
                ratelimit: Option<RatelimitKey>,
                ) -> ApiRequest<T>
     {
-        let mut owned_params = BTreeMap::new();
-        for (key, value) in params {
-            owned_params.insert(key.to_owned(), value.to_owned());
-        }
-
         ApiRequest {
-            inner: Arc::new( RequestRef {
-                url: url,
-                params: owned_params,
-                client: client,
-                method: method,
-                ratelimit: ratelimit,
-            }),
+            inner: Arc::new(RequestRef::new(url, params, client, method, ratelimit)),
             state: RequestState::SetupRequest,
             attempt: 0,
             max_attempts: 1,
@@ -416,6 +427,25 @@ impl<T: DeserializeOwned + PaginationTrait + 'static + Send> ApiRequest<T> {
     }
 }
 
+impl<T: DeserializeOwned + PaginationTrait + 'static + Send> IterableApiRequest<T> {
+    
+    pub fn new(url: String,
+               params: BTreeMap<&str, &str>,
+               client: Client,
+               method: Method,
+               ratelimit: Option<RatelimitKey>
+               ) -> IterableApiRequest<T>
+    {
+        let request_ref =
+            Arc::new(RequestRef::new(url, params, client, method, ratelimit));
+
+        IterableApiRequest {
+            inner: request_ref,
+            state: IterableApiRequestState::Start,
+        }
+    }
+}
+
 
 pub struct RatelimitWaiter {
     limit: Ratelimit,
@@ -639,33 +669,7 @@ macro_rules! retry_ready {
 
 use futures::Stream;
 
-/*
-impl<T: DeserializeOwned + 'static + Send> Future for ApiRequest<T> {
-    type Item = <Self as Stream>::Item;
-    type Error = <Self as Stream>::Error;
-
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        let stream = self as &mut Stream<Item=Self::Item, Error=Self::Error>;
-        let state = stream.poll();
-        
-        match state {
-            Err(err) => Err(err),
-            Ok(maybe_ready) => {
-                match maybe_ready {
-                    Async::NotReady => Ok(Async::NotReady),
-                    Async::Ready(item) => {
-                        match item {
-                            Some(result) => Ok(Async::Ready(result)),
-                            None => panic!("Request future returned None for first result!")
-                        }
-                    }
-                }
-            }
-        }
-    }
-}
-*/
-impl<T: DeserializeOwned + 'static + Send> Stream for IterableApiRequest<T> {
+impl<T: DeserializeOwned + PaginationTrait + 'static + Send> Stream for IterableApiRequest<T> {
     type Item = T;
     type Error = Error;
 
@@ -694,8 +698,25 @@ impl<T: DeserializeOwned + 'static + Send> Stream for IterableApiRequest<T> {
                             match state {
                                 Async::NotReady => return Ok(Async::NotReady),
                                 Async::Ready(res) => {
-                                    //TODO check pagination
-                                    self.state = IterableApiRequestState::Finished;
+                                    let cursor = 
+                                        res.cursor().as_ref()
+                                        .and_then(|cursor| cursor.cursor.clone());
+
+                                    match cursor {
+                                        Some(cursor) => {
+                                            self.state = IterableApiRequestState::PollInner(
+                                                ApiRequest {
+                                                    inner: self.inner.clone(),
+                                                    state: RequestState::SetupRequest,
+                                                    attempt: 0,
+                                                    max_attempts: 1,
+                                                    pagination: Some(cursor.clone()),
+                                                });
+                                        },
+                                        None => {
+                                            self.state = IterableApiRequestState::Finished;
+                                        }
+                                    }
                                     return Ok(Async::Ready(Some(res)));
                                 }
                             }
@@ -780,9 +801,16 @@ impl<T: DeserializeOwned + 'static + Send> Future for ApiRequest<T> {
                         mut_limits.inflight = mut_limits.inflight + 1;
                     }
 
-                    let builder = reqwest.request(self.inner.method.clone(), &self.inner.url);
-                    let builder = client.apply_standard_headers(builder);
-                    let r = builder.query(&self.inner.params);
+                    let mut builder = reqwest.request(self.inner.method.clone(), &self.inner.url);
+                    builder = client.apply_standard_headers(builder);
+                    builder = builder.query(&self.inner.params);
+                    builder = 
+                        if let Some(cursor) = &self.pagination {
+                            builder.query(&[("after", cursor)])
+                        } else {
+                            builder
+                        };
+                     
 
                     let key_err = self.inner.ratelimit.clone();
                     let key_ok = self.inner.ratelimit.clone();
@@ -790,7 +818,7 @@ impl<T: DeserializeOwned + 'static + Send> Future for ApiRequest<T> {
                     let client_ok = client.clone();
                     
                     let f = 
-                        r.send()
+                        builder.send()
                             .map_err(move |err| {
                                 if let Some(key) = key_err {
                                     if let Some(limits) = client_err.ratelimit(key) {
diff --git a/src/helix/models.rs b/src/helix/models.rs
index bdb8438..4124fd2 100644
--- a/src/helix/models.rs
+++ b/src/helix/models.rs
@@ -7,7 +7,6 @@ use super::types::{UserId, VideoId, ChannelId};
 
 pub trait PaginationTrait {
     fn cursor<'a>(&'a self) -> &'a Option<Cursor>;
-    fn set_request(&mut self);
 }
 
 #[derive(Debug, Deserialize)]
@@ -17,17 +16,14 @@ pub struct DataContainer<T> {
 
 impl<T> PaginationTrait for DataContainer<T> {
     fn cursor<'a>(&'a self) -> &'a Option<Cursor> { &None }
-    fn set_request(&mut self) {}
 }
 
 impl<T> PaginationTrait for PaginationContainer<T> {
     fn cursor<'a>(&'a self) -> &'a Option<Cursor> { &self.pagination }
-    fn set_request(&mut self) {}
 }
 
 impl PaginationTrait for Credentials {
     fn cursor<'a>(&'a self) -> &'a Option<Cursor> { &None }
-    fn set_request(&mut self) {}
 }
 
 #[derive(Debug, Deserialize)]
@@ -38,9 +34,10 @@ pub struct PaginationContainer<T> {
 
 #[derive(Debug, Deserialize)]
 pub struct Cursor {
-    cursor: String
+    pub cursor: Option<String>
 }
 
+
 #[derive(Debug, Deserialize)]
 pub struct Credentials {
     pub access_token: String,
diff --git a/src/helix/namespaces/videos.rs b/src/helix/namespaces/videos.rs
index 7b8839b..382c9ea 100644
--- a/src/helix/namespaces/videos.rs
+++ b/src/helix/namespaces/videos.rs
@@ -1,20 +1,34 @@
-use futures::future::Future;
-use super::super::models::{DataContainer, PaginationContainer, User, Video, Clip};
+use super::super::models::{PaginationContainer, Video};
 use super::super::Client; 
-use std::collections::BTreeMap;
-const API_DOMAIN: &'static str = "api.twitch.tv";
+use super::super::ClientTrait;
+use super::super::RatelimitKey;
+use super::super::IterableApiRequest;
 use super::Namespace;
 
+use std::collections::BTreeMap;
+use reqwest::Method;
+
 pub struct Videos {}
 type VideosNamespace = Namespace<Videos>;
 
 impl VideosNamespace {
-    /*
-    pub fn videos(self, video_id) -> impl Future<Item=DataContainer<User>, Error=reqwest::Error> {
-        use self::videos;
-        users(self.client, id, login)
+    pub fn by_id(self, ids: Vec<&str>)
+        -> IterableApiRequest<PaginationContainer<Video>> {
+        use self::by_id;
+        by_id(self.client, ids)
+    }
+
+    pub fn by_user(self, user_id: &str)
+        -> IterableApiRequest<PaginationContainer<Video>> {
+        use self::by_user;
+        by_user(self.client, user_id)
+    }
+
+    pub fn for_game(self, game_id: &str) 
+        -> IterableApiRequest<PaginationContainer<Video>> {
+        use self::for_game;
+        for_game(self.client, game_id)
     }
-    */
 }
 
 impl Client {
@@ -24,31 +38,40 @@ impl Client {
     }
 }
 
-/*
-pub fn videos(
-    client: Client,
-    video_id:   Option<Vec<&str>>,
-    user_id:    Option<&str>,
-    game_id:    Option<&str>,
-) -> impl Future<Item = PaginationContainer<Video>, Error = reqwest::Error> {
-    let mut url =
-        String::from("https://") + &String::from(API_DOMAIN) + &String::from("/helix/videos");
+pub fn by_id(client: Client, ids: Vec<&str>) 
+    -> IterableApiRequest<PaginationContainer<Video>> {
+    let url =
+        String::from("https://") + client.domain() + &String::from("/helix/videos");
 
     let mut params  = BTreeMap::new();
-    for user in user_id {
-        params.insert("user_id", user);
+    for id in ids {
+        params.insert("id", id);
     }
 
-    let request = client.client().get(&url);
-    let request = client.apply_standard_headers(request);
-    let request = request.query(&params);
+    IterableApiRequest::new(url, params, client,
+                            Method::GET, Some(RatelimitKey::Default))
+}
 
-    request
-        .send()
-        .map(|mut res| {
-            res.json::<PaginationContainer<Video>>()
-        })
-        .and_then(|json| json)
+pub fn by_user(client: Client, user_id: &str) 
+    -> IterableApiRequest<PaginationContainer<Video>> {
+    let url =
+        String::from("https://") + client.domain() + &String::from("/helix/videos");
+
+    let mut params  = BTreeMap::new();
+    params.insert("user_id", user_id);
+
+    IterableApiRequest::new(url, params, client,
+                            Method::GET, Some(RatelimitKey::Default))
+}
+
+pub fn for_game(client: Client, game_id: &str) 
+    -> IterableApiRequest<PaginationContainer<Video>> {
+    let url =
+        String::from("https://") + client.domain() + &String::from("/helix/videos");
+
+    let mut params  = BTreeMap::new();
+    params.insert("game_id", game_id);
 
+    IterableApiRequest::new(url, params, client,
+                            Method::GET, Some(RatelimitKey::Default))
 }
-*/