Merge branch 'fix-timeout' into ap07

plume-common/Cargo.toml

@@ -23,7 +23,9 @@ askama_escape = "0.10.3"
 activitystreams = "0.7.0-alpha.18"
 activitystreams-ext = "0.1.0-alpha.2"
 url = "2.2.2"
+flume = "0.10.12"
 tokio = { version = "1.18.1", features = ["full"] }
+futures = "0.3.21"
 
 [dependencies.chrono]
 features = ["serde"]

plume-common/src/activity_pub/mod.rs

@@ -10,7 +10,8 @@ use activitystreams::{
 };
 use activitystreams_ext::{Ext1, Ext2, UnparsedExtension};
 use array_tool::vec::Uniq;
-use reqwest::{header::HeaderValue, ClientBuilder, Url};
+use futures::future::join_all;
+use reqwest::{header::HeaderValue, ClientBuilder, RequestBuilder, Url};
 use rocket::{
     http::Status,
     request::{FromRequest, Request},
@@ -156,7 +157,29 @@ where
         .build()
         .expect("Error while initializing tokio runtime for federation");
     rt.block_on(async {
-        let (tx, mut rx) = mpsc::channel(32);
+        let capacity = 50;
+        let (tx, rx) = flume::bounded::<RequestBuilder>(capacity);
+        let mut handles = Vec::with_capacity(capacity);
+        for _ in 0..capacity {
+            let rx = rx.clone();
+            let handle = rt.spawn(async move {
+                while let Ok(request_builder) = rx.recv_async().await {
+                    let _ = request_builder
+                        .send()
+                        .await
+                        .map(move |r| {
+                            if r.status().is_success() {
+                                debug!("Successfully sent activity to inbox ({})", &r.url());
+                            } else {
+                                warn!("Error while sending to inbox ({:?})", &r)
+                            }
+                            debug!("Response: \"{:?}\"\n", r);
+                        })
+                        .map_err(|e| warn!("Error while sending to inbox ({:?})", e));
+                }
+            });
+            handles.push(handle);
+        }
         for inbox in boxes {
             let body = signed.to_string();
             let mut headers = request::headers();
@@ -182,33 +205,11 @@ where
                 request::signature(sender, &headers, ("post", url.path(), url.query()))
                     .expect("activity_pub::broadcast: request signature error"),
             );
-            let client = client.clone();
-            let tx = tx.clone();
-            let _ = tx.send(
-                rt.spawn(
-                    client
-                        .post(&inbox)
-                        .headers(headers.clone())
-                        .body(body)
-                        .send(),
-                ),
-            );
-        }
-        while let Some(request) = rx.recv().await {
-            let _ = request
-                .await
-                .map(move |r| {
-                    r.map(|r| {
-                        if r.status().is_success() {
-                            debug!("Successfully sent activity to inbox ({})", &r.url());
-                        } else {
-                            warn!("Error while sending to inbox ({} {:?})", &r.url(), &r)
-                        }
-                        debug!("Response: \"{:?}\"\n", r);
-                    })
-                })
-                .map_err(|e| warn!("Error while sending to inbox ({:?})", e));
+            let request_builder = client.post(&inbox).headers(headers.clone()).body(body);
+            tx.send_async(request_builder).await.unwrap();
         }
+        drop(tx);
+        join_all(handles).await;
     });
 }