Update the logic for fetching top stories (#6)

* Update the logic for fetching top stories using futures and tokio instead.

* Fix compile errors and remove hardcode

* Fix clippy

Cargo.toml

@@ -9,6 +9,9 @@ edition = "2018"
 [dependencies]
 termion = "1.5.3"
 reqwest = "0.9.22"
+futures = "0.1.26"
+tokio = "0.1.16"
+lazy_static = "1.4.0"
 serde = { version = "1.0", features = ["derive"] }
 num_cpus = "1.10.1"
 mockito = "0.21.0"

src/hn.rs

@@ -2,19 +2,22 @@ extern crate num_cpus;
 extern crate reqwest;
 
 use chrono::Utc;
+use futures::future::join_all;
+use futures::Future;
+use futures::Stream;
+use reqwest::r#async::Client;
 use serde::Deserialize;
-use std::ops::DerefMut;
-use std::sync::Arc;
-use std::sync::LockResult;
-use std::sync::Mutex;
-use std::sync::MutexGuard;
-use std::thread;
-
-use self::reqwest::header::CONNECTION;
+use tokio::runtime::Runtime;
+use tokio::sync::mpsc::channel;
+
 use crate::time;
 #[cfg(test)]
 use mockito;
 
+lazy_static! {
+    pub static ref CLIENT: Client = Client::new();
+}
+
 #[derive(Deserialize, Debug)]
 pub struct Story {
     pub id: i64,
@@ -35,27 +38,19 @@ impl Story {
     }
 }
 
-fn next(cursor: &mut Arc<Mutex<usize>>) -> usize {
-    let result: LockResult<MutexGuard<usize>> = cursor.lock();
-    let mut guard: MutexGuard<usize> = result.unwrap();
-    let temp = guard.deref_mut();
-    *temp += 1;
-    *temp
-}
-
-/// Get top stories on Hacker News,
+/// Fetch top stories on Hacker News,
 /// Using /v0/topstories.json and /v0/item/{:id}.json endpoints.
 ///
 /// https://github.com/HackerNews/API
 ///
 /// # Examples
 /// ```
-/// let stories = match get_top_stories(10) {
+/// let stories = match fetch_top_stories(10) {
 ///     Ok(res) => res,
 ///     Err(e) => println!("{:#?}", e)
 /// }
 /// ```
-pub fn get_top_stories(num: usize) -> Result<Vec<Story>, Box<dyn std::error::Error>> {
+pub fn fetch_top_stories(num: usize) -> Result<Vec<Story>, reqwest::Error> {
     #[cfg(not(test))]
     let hn_url = "https://hacker-news.firebaseio.com";
     #[cfg(test)]
@@ -68,54 +63,65 @@ pub fn get_top_stories(num: usize) -> Result<Vec<Story>, Box<dyn std::error::Err
     } else {
         vec[0..num].to_vec()
     };
+    fetch_stories(vec)
+}
 
-    let mut handles: Vec<thread::JoinHandle<Vec<Story>>> = Vec::new();
-    let lock: Arc<Mutex<usize>> = Arc::new(Mutex::new(0));
-
-    for _ in 0..num_cpus::get() {
-        let mut lock2 = lock.clone();
-        let hn_url2 = hn_url.to_owned();
-        let vec2 = vec.clone();
-        handles.push(thread::spawn(move || {
-            let mut stories = Vec::new();
-            loop {
-                let cursor = next(&mut lock2);
-
-                if cursor > vec2.len() {
-                    break;
-                }
-
-                let story_url = format!("{}/v0/item/{}.json", hn_url2, vec2[cursor - 1],);
-                let client = reqwest::Client::new();
-                if let Ok(mut res) = client
-                    .get(story_url.as_str())
-                    .header(CONNECTION, "Keep-Alive")
-                    .send()
-                {
-                    if let Ok(story) = res.json() {
-                        stories.push(story)
-                    }
-                }
-            }
-            stories
-        }));
+fn fetch_stories(ids: Vec<i64>) -> Result<Vec<Story>, reqwest::Error> {
+    if ids.is_empty() {
+        return Ok(Vec::new());
     }
-
+    let mut core = Runtime::new().unwrap();
+    let (tx, rx) = channel(ids.len());
+
+    let num = ids.len();
+    let all = ids.into_iter().enumerate().map(move |(i, id)| {
+        let mut tx = tx.clone();
+        fetch_story(id)
+            .then(move |x| tx.try_send((i, x)))
+            .map(|_| ())
+            .map_err(|e| println!("{:?}", e))
+    });
+    core.spawn(join_all(all).map(|_| ()));
     let mut stories = Vec::new();
-    for handle in handles.into_iter() {
-        let mut res = handle.join().unwrap();
-        stories.append(&mut res);
-    }
+    match rx.take(num as u64).collect().wait() {
+        Ok(mut x) => {
+            x.sort_by(|a, b| {
+                let (i1, _) = a;
+                let (i2, _) = b;
+                i1.cmp(i2)
+            });
+            for s in x {
+                let (_, story) = s;
+                if let Ok(st) = story {
+                    stories.push(st)
+                }
+            }
+        }
+        Err(e) => eprintln!("{:?}", e),
+    };
     Ok(stories)
 }
 
+fn fetch_story(id: i64) -> impl Future<Item = Story, Error = reqwest::Error> {
+    #[cfg(not(test))]
+    let hn_url = "https://hacker-news.firebaseio.com";
+    #[cfg(test)]
+    let hn_url = &mockito::server_url();
+
+    let url = format!("{}/v0/item/{}.json", hn_url, id);
+    CLIENT
+        .get(url.as_str())
+        .send()
+        .and_then(move |mut res| res.json())
+}
+
 #[cfg(test)]
 mod tests {
-    use crate::hn::get_top_stories;
+    use crate::hn::fetch_top_stories;
     use mockito::mock;
 
     #[test]
-    fn test_get_get_top_stories1() {
+    fn test_fetch_top_stories1() {
         let _m1 = mock("GET", "/v0/topstories.json")
             .with_status(200)
             .with_header("content-type", "application/json")
@@ -129,27 +135,27 @@ mod tests {
             .create();
 
         assert!(
-            get_top_stories(1).is_ok(),
-            "get_top_stories should return top stories"
+            fetch_top_stories(1).is_ok(),
+            "fetch_top_stories should return top stories"
         );
-        let stories = get_top_stories(1);
+        let stories = fetch_top_stories(1);
         let story = &stories.unwrap()[0];
         assert_eq!(story.by, String::from("pg"));
         assert_eq!(story.id, 1);
     }
 
     #[test]
-    fn test_get_get_top_stories2() {
+    fn test_fetch_top_stories2() {
         let _m1 = mock("GET", "/v0/topstories.json").with_status(500).create();
 
         assert!(
-            get_top_stories(1).is_err(),
-            "get_top_stories should return an error"
+            fetch_top_stories(1).is_err(),
+            "fetch_top_stories should return an error"
         );
     }
 
     #[test]
-    fn test_get_get_top_stories3() {
+    fn test_fetch_top_stories3() {
         let _m1 = mock("GET", "/v0/topstories.json")
             .with_status(200)
             .with_header("content-type", "application/json")
@@ -165,27 +171,27 @@ mod tests {
         let _m3 = mock("GET", "/v0/item/2.json").with_status(500).create();
 
         assert!(
-            get_top_stories(5).is_ok(),
-            "get_top_stories should return stories."
+            fetch_top_stories(5).is_ok(),
+            "fetch_top_stories should return stories."
         );
-        let stories = get_top_stories(1);
+        let stories = fetch_top_stories(1);
         let story = &stories.unwrap()[0];
         assert_eq!(story.by, String::from("pg"));
         assert_eq!(story.id, 1);
     }
 
     #[test]
-    fn test_get_get_top_stories4() {
+    fn test_fetch_top_stories4() {
         let _m1 = mock("GET", "/v0/topstories.json")
             .with_status(200)
             .with_header("content-type", "application/json")
             .with_body("[]")
             .create();
         assert!(
-            get_top_stories(5).is_ok(),
-            "get_top_stories should return stories."
+            fetch_top_stories(5).is_ok(),
+            "fetch_top_stories should return stories."
         );
-        let stories = get_top_stories(1);
+        let stories = fetch_top_stories(1);
         assert_eq!(stories.unwrap().len(), 0);
     }
 }