Issue 62 validator all channel list pages (#64)

* domain - repository - RepositoryStream

* sentry - resource - channel_list - response - add `serde(rename_all = "camelCase")`

* validator - persistence - channel - api - fetch all channels in one future

* validator - persistence - channel - api - fetch all channels in one future

* fix `clippy` warning

Author: elpiel

domain/src/lib.rs

@@ -55,7 +55,7 @@ impl error::Error for DomainError {
 pub mod repository {
     use std::pin::Pin;
 
-    use futures::Future;
+    use futures::{Future, Stream};
 
     pub trait IOError: std::error::Error + Send {}
 
@@ -69,4 +69,6 @@ pub mod repository {
     }
 
     pub type RepositoryFuture<T> = Pin<Box<dyn Future<Output = Result<T, RepositoryError>> + Send>>;
+
+    pub type RepositoryStream<T> = Pin<Box<dyn Stream<Item = Result<T, RepositoryError>> + Send>>;
 }

sentry/src/application/resource/channel/channel_list/response.rs

@@ -3,7 +3,8 @@ use tower_web::Response;
 
 use domain::Channel;
 
-#[derive(Debug, Response)]
+#[derive(Debug, Response, Serialize)]
+#[serde(rename_all = "camelCase")]
 pub struct ChannelListResponse {
     pub channels: Vec<Channel>,
     pub total_pages: u64,

validator/src/infrastructure/persistence/channel/api.rs

@@ -1,5 +1,9 @@
+use std::iter::once;
+use std::sync::Arc;
+
 use futures::compat::Future01CompatExt;
-use futures::future::FutureExt;
+use futures::future::{ok, try_join_all};
+use futures::{FutureExt, TryFutureExt};
 use futures_legacy::Future as LegacyFuture;
 use reqwest::r#async::{Client, Response};
 use serde::Deserialize;
@@ -9,39 +13,74 @@ use domain::{Channel, RepositoryFuture};
 use crate::domain::channel::ChannelRepository;
 use crate::infrastructure::persistence::api::ApiPersistenceError;
 
+#[derive(Clone)]
 pub struct ApiChannelRepository {
     pub client: Client,
 }
 
-impl ChannelRepository for ApiChannelRepository {
-    fn all(&self, identity: &str) -> RepositoryFuture<Vec<Channel>> {
-        let list_page_url = |page: u32| {
-            format!(
-                "http://localhost:8005/channel/list?validator={}&page={}",
-                identity, page
+impl ApiChannelRepository {
+    fn fetch_page(&self, page: u64, identity: &str) -> RepositoryFuture<ChannelAllResponse> {
+        self.client
+            // call Sentry and fetch first page, where validator = identity
+            .get(
+                format!(
+                    "http://localhost:8005/channel/list?validator={}&page={}",
+                    identity, page
+                )
+                .as_str(),
             )
-        };
-        let first_page = self
-            .client
-            // call Sentry and fetch first page, where validator = params.identifier
-            .get(&list_page_url(1))
             .send()
-            .and_then(|mut res: Response| res.json::<AllResponse>())
-            .map(|response| response.channels)
+            .and_then(|mut res: Response| res.json::<ChannelAllResponse>())
             // @TODO: Error handling
-            .map_err(|_error| ApiPersistenceError::Reading.into());
+            .map_err(|_error| ApiPersistenceError::Reading.into())
+            .compat()
+            .boxed()
+    }
+}
 
-        // call Sentry again and concat all the Channels in 1 Stream
+impl ChannelRepository for ApiChannelRepository {
+    fn all(&self, identity: &str) -> RepositoryFuture<Vec<Channel>> {
+        let identity = Arc::new(identity.to_string());
+        let handle = self.clone();
+
+        let first_page = handle.fetch_page(1, &identity.clone());
+
+        // call Sentry again and concat all the Channels in Future
         // fetching them until no more Channels are returned
-        // @TODO: fetch the rest of the results
+        first_page
+            .and_then(move |response| {
+                let first_page_future = ok(response.channels).boxed();
+
+                if response.total_pages < 2 {
+                    first_page_future
+                } else {
+                    let identity = identity.clone();
+                    let futures = (2..=response.total_pages)
+                        .map(|page| {
+                            handle
+                                .fetch_page(page, &identity)
+                                .map(|response_result| {
+                                    response_result.and_then(|response| Ok(response.channels))
+                                })
+                                .boxed()
+                        })
+                        .chain(once(first_page_future));
 
-        first_page.compat().boxed()
+                    try_join_all(futures)
+                        .map(|result_all| {
+                            result_all
+                                .and_then(|all| Ok(all.into_iter().flatten().collect::<Vec<_>>()))
+                        })
+                        .boxed()
+                }
+            })
+            .boxed()
     }
 }
 
-#[derive(Deserialize)]
+#[derive(Deserialize, Debug)]
 #[serde(rename_all = "camelCase")]
-struct AllResponse {
+struct ChannelAllResponse {
     pub channels: Vec<Channel>,
     pub total_pages: u64,
 }

validator/src/main.rs

@@ -11,10 +11,13 @@ fn main() {
         let repo = validator::infrastructure::persistence::channel::api::ApiChannelRepository {
             client: Client::new(),
         };
-        println!(
-            "{:#?}",
-            await!(repo.all("0x2892f6C41E0718eeeDd49D98D648C789668cA67d"))
-        );
+
+        let all_channels = await!(repo.all("0x2892f6C41E0718eeeDd49D98D648C789668cA67d"));
+
+        match all_channels {
+            Ok(channel) => println!("{:#?}", channel),
+            Err(error) => eprintln!("Error occurred: {:#?}", error),
+        };
     };
 
     tokio::run(future.unit_error().boxed().compat());