aitzol/Plume
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Broadcast asynchronously

Browse Source
This commit is contained in:
Kitaiti Makoto 2022-05-05 04:34:04 +09:00
parent 9016995d92
commit ce4b216722
1 changed files with 56 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

94
plume-common/src/activity_pub/mod.rs
View File

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