crates/api_common/src/send_activity.rs

   1 use crate::context::LemmyContext;
   2 use activitypub_federation::config::Data;
   3 use futures::future::BoxFuture;
   4 use lemmy_db_schema::source::post::Post;
   5 use lemmy_utils::{error::LemmyResult, SYNCHRONOUS_FEDERATION};
   6 use once_cell::sync::{Lazy, OnceCell};
   7 use tokio::{
   8   sync::{
   9     mpsc,
  10     mpsc::{UnboundedReceiver, UnboundedSender, WeakUnboundedSender},
  11     Mutex,
  12   },
  13   task::JoinHandle,
  14 };
  15
  16 type MatchOutgoingActivitiesBoxed =
  17   Box<for<'a> fn(SendActivityData, &'a Data<LemmyContext>) -> BoxFuture<'a, LemmyResult<()>>>;
  18
  19 /// This static is necessary so that activities can be sent out synchronously for tests.
  20 pub static MATCH_OUTGOING_ACTIVITIES: OnceCell<MatchOutgoingActivitiesBoxed> = OnceCell::new();
  21
  22 #[derive(Debug)]
  23 pub enum SendActivityData {
  24   CreatePost(Post),
  25 }
  26
  27 // TODO: instead of static, move this into LemmyContext. make sure that stopping the process with
  28 //       ctrl+c still works.
  29 static ACTIVITY_CHANNEL: Lazy<ActivityChannel> = Lazy::new(|| {
  30   let (sender, receiver) = mpsc::unbounded_channel();
  31   let weak_sender = sender.downgrade();
  32   ActivityChannel {
  33     weak_sender,
  34     receiver: Mutex::new(receiver),
  35     keepalive_sender: Mutex::new(Some(sender)),
  36   }
  37 });
  38
  39 pub struct ActivityChannel {
  40   weak_sender: WeakUnboundedSender<SendActivityData>,
  41   receiver: Mutex<UnboundedReceiver<SendActivityData>>,
  42   keepalive_sender: Mutex<Option<UnboundedSender<SendActivityData>>>,
  43 }
  44
  45 impl ActivityChannel {
  46   pub async fn retrieve_activity() -> Option<SendActivityData> {
  47     let mut lock = ACTIVITY_CHANNEL.receiver.lock().await;
  48     lock.recv().await
  49   }
  50
  51   pub async fn submit_activity(
  52     data: SendActivityData,
  53     context: &Data<LemmyContext>,
  54   ) -> LemmyResult<()> {
  55     if *SYNCHRONOUS_FEDERATION {
  56       MATCH_OUTGOING_ACTIVITIES
  57         .get()
  58         .expect("retrieve function pointer")(data, context)
  59       .await?;
  60     }
  61     // could do `ACTIVITY_CHANNEL.keepalive_sender.lock()` instead and get rid of weak_sender,
  62     // not sure which way is more efficient
  63     else if let Some(sender) = ACTIVITY_CHANNEL.weak_sender.upgrade() {
  64       sender.send(data)?;
  65     }
  66     Ok(())
  67   }
  68
  69   pub async fn close(outgoing_activities_task: JoinHandle<LemmyResult<()>>) -> LemmyResult<()> {
  70     ACTIVITY_CHANNEL.keepalive_sender.lock().await.take();
  71     outgoing_activities_task.await??;
  72     Ok(())
  73   }
  74 }