Yahoo client fetches providers simultaneously.

Author: metalwhale

chloria-backend/chloria-job/src/execution/cases/collect_news.rs

@@ -90,7 +90,7 @@ impl LocalCase for CollectNewsCase {
             let sender = sender.clone();
             let semaphore = Arc::clone(&semaphore);
             let handles = new_fetcher
-                .fetch_news(Box::new(move |article| {
+                .fetch_news(Arc::new(move |article| {
                     let news = NewsEntity::new(article.id);
                     let http_helper = Arc::clone(&http_helper);
                     let file_storage = Arc::clone(&file_storage);

chloria-backend/chloria-job/src/execution/ports/news_fetcher.rs

@@ -1,4 +1,4 @@
-use std::{future::Future, pin::Pin};
+use std::{future::Future, pin::Pin, sync::Arc};
 
 use anyhow::Result;
 use async_trait::async_trait;
@@ -18,23 +18,22 @@ pub(crate) struct FetchNewsArticle {
 }
 
 pub(crate) type FetchNewsOutput = JoinHandle<Result<()>>;
-pub(crate) type FetchNewsHandler = Box<dyn Fn(FetchNewsArticle) -> FetchNewsOutput + Send>;
+pub(crate) type FetchNewsHandler = Arc<dyn Fn(FetchNewsArticle) -> FetchNewsOutput + Send + Sync>;
 
 #[async_trait]
 pub(crate) trait NewsFetcher: Send + Sync {
-    async fn fetch_news(&self, handler: FetchNewsHandler) -> Vec<FetchNewsOutput>;
+    async fn fetch_news(self: Arc<Self>, handler: FetchNewsHandler) -> Vec<FetchNewsOutput>;
 }
 
 mock! {
     pub(in super::super) NewsFetcher {}
 
     impl NewsFetcher for NewsFetcher {
-        fn fetch_news<'life0, 'async_trait>(
-            &'life0 self,
+        fn fetch_news<'async_trait>(
+            self: Arc<Self>,
             handler: FetchNewsHandler,
         ) -> Pin<Box<dyn Future<Output = Vec<FetchNewsOutput>> + Send + 'async_trait>>
         where
-            'life0: 'async_trait,
             Self: 'async_trait;
     }
 }

chloria-backend/chloria-job/src/infrastructure/news_fetcher/newsdata.rs

@@ -1,3 +1,5 @@
+use std::sync::Arc;
+
 use anyhow::{Context, Result};
 use async_trait::async_trait;
 use chrono::NaiveDateTime;
@@ -88,7 +90,7 @@ impl NewsdataClient {
 
 #[async_trait]
 impl NewsFetcher for NewsdataClient {
-    async fn fetch_news(&self, handler: FetchNewsHandler) -> Vec<FetchNewsOutput> {
+    async fn fetch_news(self: Arc<Self>, handler: FetchNewsHandler) -> Vec<FetchNewsOutput> {
         let mut outputs = vec![];
         let mut remaining_results_num = None;
         let mut page = None;

chloria-backend/chloria-job/src/infrastructure/news_fetcher/yahoo.rs

@@ -1,10 +1,13 @@
-use anyhow::Result;
+use std::sync::Arc;
+
+use anyhow::{Error, Result};
 use async_trait::async_trait;
 use chrono::{DateTime, Local};
 use log::{error, info};
 use regex::Regex;
 use reqwest::Client;
 use serde::Deserialize;
+use tokio::sync::{mpsc, Semaphore};
 
 use crate::execution::ports::news_fetcher::{FetchNewsArticle, FetchNewsHandler, FetchNewsOutput, NewsFetcher};
 
@@ -742,7 +745,7 @@ impl YahooClient {
         }
     }
 
-    async fn fetch_provider(&self, provider: &str) -> Result<Vec<FetchNewsArticle>> {
+    async fn fetch_provider(self: Arc<Self>, provider: &str) -> Result<Vec<FetchNewsArticle>> {
         let mut articles = vec![];
         let url = format!("https://news.yahoo.co.jp/rss/media/{}/all.xml", provider);
         let response_text = Client::new().get(url).send().await?.text().await?;
@@ -790,19 +793,41 @@ impl YahooClient {
 
 #[async_trait]
 impl NewsFetcher for YahooClient {
-    async fn fetch_news(&self, handler: FetchNewsHandler) -> Vec<FetchNewsOutput> {
-        let mut outputs = vec![];
+    async fn fetch_news(self: Arc<Self>, handler: FetchNewsHandler) -> Vec<FetchNewsOutput> {
+        const CHANNEL_CAPACITY: usize = 100;
+        let (sender, mut receiver) = mpsc::channel(CHANNEL_CAPACITY);
+        let receiver_handle = tokio::spawn(async move {
+            let mut handles = vec![];
+            while let Some(handle) = receiver.recv().await {
+                handles.push(handle);
+            }
+            handles
+        });
+        const PROVIDER_PERMITS_NUM: usize = 20;
+        let semaphore = Arc::new(Semaphore::new(PROVIDER_PERMITS_NUM));
         for provider in &self.providers {
-            match self.fetch_provider(provider).await {
-                Ok(articles) => {
-                    info!("provider={}, articles.len={}", provider, articles.len());
-                    for article in articles {
-                        outputs.push(handler(article));
+            let provider = provider.to_string();
+            let self = Arc::clone(&self);
+            let sender = sender.clone();
+            let semaphore = Arc::clone(&semaphore);
+            let handler = Arc::clone(&handler);
+            // Use `spawn_local` to preserve the LocalSet, which is later used by the handler
+            // Ref: `execution::cases::collect_news::CollectNewsCase`
+            tokio::task::spawn_local(async move {
+                let _permit = semaphore.acquire().await?;
+                match self.fetch_provider(&provider).await {
+                    Ok(articles) => {
+                        info!("provider={}, articles.len={}", provider, articles.len());
+                        for article in articles {
+                            sender.send(handler(article)).await?;
+                        }
                     }
+                    Err(error) => error!("provider={}, error={}", provider, error),
                 }
-                Err(error) => error!("provider={}, error={}", provider, error),
-            }
+                Ok::<(), Error>(())
+            });
         }
-        outputs
+        drop(sender); // Drop early (before awaiting the receiver) to prevent the sender from blocking the channel from closing
+        receiver_handle.await.unwrap_or(vec![])
     }
 }