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

Execute broadcast synchronously

Browse Source
This commit is contained in:
Kitaiti Makoto 2022-05-04 21:12:35 +09:00
parent 3eb7662aef
commit 35aa2374c4
1 changed files with 20 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,13 +1,12 @@
use activitypub::{Activity, Link, Object}; use activitypub::{Activity, Link, Object};
use array_tool::vec::Uniq; use array_tool::vec::Uniq;
use reqwest::{header::HeaderValue, r#async::ClientBuilder, 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::prelude::*;
use tracing::{debug, warn}; use tracing::{debug, warn};
use self::sign::Signable; use self::sign::Signable;
@ -140,8 +139,6 @@ 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");
let mut rt = tokio::runtime::current_thread::Runtime::new()
.expect("Error while initializing tokio runtime for federation");
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();
@ -162,30 +159,26 @@ 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));
rt.spawn( let _ = client
client .post(&inbox)
.post(&inbox) .headers(headers.clone())
.headers(headers.clone()) .header(
.header( "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)
.body(body) .send()
.send() .map(move |r| {
.and_then(move |r| { if r.status().is_success() {
if r.status().is_success() { debug!("Successfully sent activity to inbox ({})", &inbox);
debug!("Successfully sent activity to inbox ({})", &inbox); } else {
} else { warn!("Error while sending to inbox ({} {:?})", &inbox, &r)
warn!("Error while sending to inbox ({} {:?})", &inbox, &r) }
} debug!("Response: \"{:?}\"\n", r);
r.into_body().concat2() })
}) .map_err(|e| warn!("Error while sending to inbox ({:?})", e));
.map(move |response| debug!("Response: \"{:?}\"\n", response))
.map_err(|e| warn!("Error while sending to inbox ({:?})", e)),
);
} }
rt.run().unwrap();
} }
#[derive(Shrinkwrap, Clone, Serialize, Deserialize)] #[derive(Shrinkwrap, Clone, Serialize, Deserialize)]