mirrors/Plume
1
0
Fork 0
mirror of https://git.joinplu.me/Plume/Plume.git synced 2024-11-29 06:51:01 +00:00
Code Issues Projects Releases Packages Wiki Activity

Broadcast asynchronously

Browse source
This commit is contained in:
Kitaiti Makoto 2022-05-05 13:03:41 +09:00
parent 1f8da7e63d
commit 97632fdbfe
1 changed files with 30 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
plume-common/src/activity_pub/mod.rs
View file

@ -1,6 +1,7 @@
use activitypub::{Activity, Link, Object}; use activitypub::{Activity, Link, Object};
use array_tool::vec::Uniq; 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::{ use rocket::{
http::Status, http::Status,
request::{FromRequest, Request}, request::{FromRequest, Request},
@ -145,6 +146,29 @@ where
.build() .build()
.expect("Error while initializing tokio runtime for federation"); .expect("Error while initializing tokio runtime for federation");
rt.block_on(async { rt.block_on(async {
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 { for inbox in boxes {
let body = signed.to_string(); let body = signed.to_string();
let mut headers = request::headers(); let mut headers = request::headers();
@ -165,7 +189,7 @@ where
} }
headers.insert("Host", host_header_value.unwrap()); headers.insert("Host", host_header_value.unwrap());
headers.insert("Digest", request::Digest::digest(&body)); headers.insert("Digest", request::Digest::digest(&body));
let _ = client let request_builder = client
.post(&inbox) .post(&inbox)
.headers(headers.clone()) .headers(headers.clone())
.header( .header(
@ -173,19 +197,11 @@ where
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) .body(body);
.send() tx.send_async(request_builder).await.unwrap();
.await
.map(move |r| {
if r.status().is_success() {
debug!("Successfully sent activity to inbox ({})", &inbox);
} else {
warn!("Error while sending to inbox ({} {:?})", &inbox, &r)
}
debug!("Response: \"{:?}\"\n", r);
})
.map_err(|e| warn!("Error while sending to inbox ({:?})", e));
} }
drop(tx);
join_all(handles).await;
}); });
} }