Merge remote-tracking branch 'origin/fix-timeout' into ap07

plume-common/Cargo.toml

@@ -11,18 +11,19 @@ heck = "0.4.0"
 hex = "0.4"
 openssl = "0.10.22"
 rocket = "0.4.6"
-reqwest = { version = "0.9", features = ["socks"] }
+reqwest = { version = "0.11.10", features = ["blocking", "json", "socks"] }
 serde = "1.0"
 serde_derive = "1.0"
 serde_json = "1.0.80"
 shrinkwraprs = "0.3.0"
 syntect = "4.5.0"
-tokio = "0.1.22"
 regex-syntax = { version = "0.6.17", default-features = false, features = ["unicode-perl"] }
 tracing = "0.1.34"
 askama_escape = "0.10.3"
 activitystreams = "0.7.0-alpha.18"
 activitystreams-ext = "0.1.0-alpha.2"
+url = "2.2.2"
+tokio = { version = "1.18.1", features = ["full"] }
 
 [dependencies.chrono]
 features = ["serde"]

plume-common/src/activity_pub/inbox.rs

@@ -366,7 +366,7 @@ pub trait FromId<C>: Sized {
     ) -> Result<Self::Object, (Option<serde_json::Value>, Self::Error)> {
         request::get(id, Self::get_sender(), proxy)
             .map_err(|_| (None, InboxError::DerefError))
-            .and_then(|mut r| {
+            .and_then(|r| {
                 let json: serde_json::Value = r
                     .json()
                     .map_err(|_| (None, InboxError::InvalidObject(None)))?;

plume-common/src/activity_pub/mod.rs

@@ -10,14 +10,14 @@ use activitystreams::{
 };
 use activitystreams_ext::{Ext1, Ext2, UnparsedExtension};
 use array_tool::vec::Uniq;
-use reqwest::{header::HeaderValue, r#async::ClientBuilder, Url};
+use reqwest::{header::HeaderValue, ClientBuilder, Url};
 use rocket::{
     http::Status,
     request::{FromRequest, Request},
     response::{Responder, Response},
     Outcome,
 };
-use tokio::prelude::*;
+use tokio::{runtime, sync::mpsc};
 use tracing::{debug, warn};
 
 use self::sign::Signable;
@@ -151,52 +151,65 @@ where
     .connect_timeout(std::time::Duration::from_secs(5))
     .build()
     .expect("Can't build client");
-    let mut rt = tokio::runtime::current_thread::Runtime::new()
+    let rt = runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
         .expect("Error while initializing tokio runtime for federation");
-    for inbox in boxes {
-        let body = signed.to_string();
-        let mut headers = request::headers();
-        let url = Url::parse(&inbox);
-        if url.is_err() {
-            warn!("Inbox is invalid URL: {:?}", &inbox);
-            continue;
+    rt.block_on(async {
+        let (tx, mut rx) = mpsc::channel(32);
+        for inbox in boxes {
+            let body = signed.to_string();
+            let mut headers = request::headers();
+            let url = Url::parse(&inbox);
+            if url.is_err() {
+                warn!("Inbox is invalid URL: {:?}", &inbox);
+                continue;
+            }
+            let url = url.unwrap();
+            if !url.has_host() {
+                warn!("Inbox doesn't have host: {:?}", &inbox);
+                continue;
+            };
+            let host_header_value = HeaderValue::from_str(url.host_str().expect("Unreachable"));
+            if host_header_value.is_err() {
+                warn!("Header value is invalid: {:?}", url.host_str());
+                continue;
+            }
+            headers.insert("Host", host_header_value.unwrap());
+            headers.insert("Digest", request::Digest::digest(&body));
+            headers.insert(
+                "Signature",
+                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(),
+                ),
+            );
         }
-        let url = url.unwrap();
-        if !url.has_host() {
-            warn!("Inbox doesn't have host: {:?}", &inbox);
-            continue;
-        };
-        let host_header_value = HeaderValue::from_str(url.host_str().expect("Unreachable"));
-        if host_header_value.is_err() {
-            warn!("Header value is invalid: {:?}", url.host_str());
-            continue;
-        }
-        headers.insert("Host", host_header_value.unwrap());
-        headers.insert("Digest", request::Digest::digest(&body));
-        rt.spawn(
-            client
-                .post(&inbox)
-                .headers(headers.clone())
-                .header(
-                    "Signature",
-                    request::signature(sender, &headers, ("post", url.path(), url.query()))
-                        .expect("activity_pub::broadcast: request signature error"),
-                )
-                .body(body)
-                .send()
-                .and_then(move |r| {
-                    if r.status().is_success() {
-                        debug!("Successfully sent activity to inbox ({})", &inbox);
-                    } else {
-                        warn!("Error while sending to inbox ({:?})", &r)
-                    }
-                    r.into_body().concat2()
+        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(move |response| debug!("Response: \"{:?}\"\n", response))
-                .map_err(|e| warn!("Error while sending to inbox ({:?})", e)),
-        );
-    }
-    rt.run().unwrap();
+                .map_err(|e| warn!("Error while sending to inbox ({:?})", e));
+        }
+    });
 }
 
 #[derive(Shrinkwrap, Clone, Serialize, Deserialize)]

plume-common/src/activity_pub/request.rs

@@ -1,10 +1,11 @@
 use chrono::{offset::Utc, DateTime};
 use openssl::hash::{Hasher, MessageDigest};
 use reqwest::{
+    blocking::{ClientBuilder, Response},
     header::{
         HeaderMap, HeaderValue, InvalidHeaderValue, ACCEPT, CONTENT_TYPE, DATE, HOST, USER_AGENT,
     },
-    ClientBuilder, Proxy, Response, Url, UrlError,
+    Proxy, Url,
 };
 use std::ops::Deref;
 use std::time::SystemTime;
@@ -18,8 +19,8 @@ const PLUME_USER_AGENT: &str = concat!("Plume/", env!("CARGO_PKG_VERSION"));
 #[derive(Debug)]
 pub struct Error();
 
-impl From<UrlError> for Error {
-    fn from(_err: UrlError) -> Self {
+impl From<url::ParseError> for Error {
+    fn from(_err: url::ParseError) -> Self {
         Error()
     }
 }