1 use crate::{signatures::sign_and_send, traits::ActorType, APUB_JSON_CONTENT_TYPE};
2 use anyhow::{anyhow, Context, Error};
5 memory_storage::Storage,
12 use lemmy_utils::{location_info, LemmyError};
15 use serde::{Deserialize, Serialize};
16 use std::{collections::BTreeMap, env, fmt::Debug, future::Future, pin::Pin};
19 pub async fn send_activity(
21 actor: &dyn ActorType,
24 activity_queue: &QueueHandle,
25 ) -> Result<(), LemmyError> {
27 let message = SendActivityTask {
28 activity: activity.clone(),
30 actor_id: actor.actor_id(),
31 private_key: actor.private_key().context(location_info!())?,
33 if env::var("APUB_TESTING_SEND_SYNC").is_ok() {
34 do_send(message, client).await?;
36 activity_queue.queue::<SendActivityTask>(message)?;
43 #[derive(Clone, Debug, Deserialize, Serialize)]
44 struct SendActivityTask {
51 /// Signs the activity with the sending actor's key, and delivers to the given inbox. Also retries
52 /// if the delivery failed.
53 impl ActixJob for SendActivityTask {
55 type Future = Pin<Box<dyn Future<Output = Result<(), Error>>>>;
56 const NAME: &'static str = "SendActivityTask";
58 const MAX_RETRIES: MaxRetries = MaxRetries::Count(10);
59 const BACKOFF: Backoff = Backoff::Exponential(2);
61 fn run(self, state: Self::State) -> Self::Future {
62 Box::pin(async move { do_send(self, &state.client).await })
66 async fn do_send(task: SendActivityTask, client: &Client) -> Result<(), Error> {
67 let mut headers = BTreeMap::<String, String>::new();
68 headers.insert("Content-Type".into(), APUB_JSON_CONTENT_TYPE.to_string());
69 let result = sign_and_send(
73 task.activity.clone(),
75 task.private_key.to_owned(),
79 if let Err(e) = result {
82 "Failed to send activity {} to {}",
90 pub fn create_activity_queue() -> QueueHandle {
91 // Start the application server. This guards access to to the jobs store
92 let queue_handle = create_server(Storage::new());
93 let arbiter = actix_web::rt::Arbiter::new();
95 // Configure and start our workers
96 WorkerConfig::new(|| MyState {
97 client: Client::default(),
99 .register::<SendActivityTask>()
100 .start_in_arbiter(&arbiter, queue_handle.clone());