LemmyNet/lemmy
2
0
Fork 0
mirror of https://github.com/LemmyNet/lemmy.git synced 2025-01-26 20:08:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

upgrade actitypub-fed to simpler interface

Browse source
This commit is contained in:
phiresky 2023-08-22 17:08:08 +00:00
parent eb761bbdcf
commit 5b4ecfb4cb
3 changed files with 5 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -11,7 +11,7 @@ checksum = "fe438c63458706e03479442743baae6c88256498e6431708f6dfc520a26515d3"
[[package]] [[package]]
name = "activitypub_federation" name = "activitypub_federation"
version = "0.4.6" version = "0.4.6"
source = "git+https://github.com/phiresky/activitypub-federation-rust/?branch=raw-sending#7991eab37536067d312d75d4fbf2a769286d4dbe" source = "git+https://github.com/phiresky/activitypub-federation-rust/?branch=raw-sending#76519bd094603b410ac939ffe64c19d34fd6121f"
dependencies = [ dependencies = [
"activitystreams-kinds", "activitystreams-kinds",
"actix-web", "actix-web",

13
crates/federate/src/worker.rs
View file

@ -2,10 +2,7 @@ use crate::{
federation_queue_state::FederationQueueState, federation_queue_state::FederationQueueState,
util::{get_activity_cached, get_actor_cached, get_latest_activity_id, retry_sleep_duration}, util::{get_activity_cached, get_actor_cached, get_latest_activity_id, retry_sleep_duration},
}; };
use activitypub_federation::{ use activitypub_federation::{activity_sending::SendActivityTask, config::Data};
activity_queue::{prepare_raw, send_raw, sign_raw},
config::Data,
};
use anyhow::Result; use anyhow::Result;
use chrono::{DateTime, TimeZone, Utc}; use chrono::{DateTime, TimeZone, Utc};
use lemmy_api_common::context::LemmyContext; use lemmy_api_common::context::LemmyContext;
@ -16,7 +13,7 @@ use lemmy_db_schema::{
utils::DbPool, utils::DbPool,
}; };
use lemmy_db_views_actor::structs::CommunityFollowerView; use lemmy_db_views_actor::structs::CommunityFollowerView;
use lemmy_utils::{error::LemmyErrorExt2, REQWEST_TIMEOUT}; use lemmy_utils::error::LemmyErrorExt2;
use reqwest::Url; use reqwest::Url;
use std::{ use std::{
collections::{HashMap, HashSet}, collections::{HashMap, HashSet},
@ -157,14 +154,13 @@ impl InstanceWorker {
let actor = get_actor_cached(pool, activity.actor_type, actor_apub_id).await?; let actor = get_actor_cached(pool, activity.actor_type, actor_apub_id).await?;
let inbox_urls = inbox_urls.into_iter().collect(); let inbox_urls = inbox_urls.into_iter().collect();
let requests = prepare_raw(object, actor.as_ref(), inbox_urls, &self.context) let requests = SendActivityTask::prepare(object, actor.as_ref(), inbox_urls, &self.context)
.await .await
.into_anyhow()?; .into_anyhow()?;
for task in requests { for task in requests {
// usually only one due to shared inbox // usually only one due to shared inbox
let mut req = sign_raw(&task, &self.context, REQWEST_TIMEOUT).await?;
tracing::info!("sending out {}", task); tracing::info!("sending out {}", task);
while let Err(e) = send_raw(&task, &self.context, req).await { while let Err(e) = task.sign_and_send(&self.context).await {
self.state.fail_count += 1; self.state.fail_count += 1;
self.state.last_retry = Utc::now(); self.state.last_retry = Utc::now();
let retry_delay: Duration = retry_sleep_duration(self.state.fail_count); let retry_delay: Duration = retry_sleep_duration(self.state.fail_count);
@ -175,7 +171,6 @@ impl InstanceWorker {
self.state.fail_count self.state.fail_count
); );
self.save_and_send_state(pool).await?; self.save_and_send_state(pool).await?;
req = sign_raw(&task, &self.context, REQWEST_TIMEOUT).await?; // resign request
tokio::select! { tokio::select! {
() = sleep(retry_delay) => {}, () = sleep(retry_delay) => {},
() = self.stop.cancelled() => { () = self.stop.cancelled() => {

2
src/lib.rs
View file

@ -173,8 +173,6 @@ pub async fn start_lemmy_server() -> Result<(), LemmyError> {
.app_data(context.clone()) .app_data(context.clone())
.client(client.clone()) .client(client.clone())
.http_fetch_limit(FEDERATION_HTTP_FETCH_LIMIT) .http_fetch_limit(FEDERATION_HTTP_FETCH_LIMIT)
.worker_count(settings.worker_count)
.retry_count(settings.retry_count)
.debug(*SYNCHRONOUS_FEDERATION) .debug(*SYNCHRONOUS_FEDERATION)
.http_signature_compat(true) .http_signature_compat(true)
.url_verifier(Box::new(VerifyUrlData(context.inner_pool().clone()))) .url_verifier(Box::new(VerifyUrlData(context.inner_pool().clone())))