2023-07-19 13:49:41 +00:00
|
|
|
use crate::context::LemmyContext;
|
|
|
|
use activitypub_federation::config::Data;
|
|
|
|
use futures::future::BoxFuture;
|
2023-08-01 13:53:36 +00:00
|
|
|
use lemmy_db_schema::{
|
|
|
|
newtypes::DbUrl,
|
|
|
|
source::{comment::Comment, community::Community, person::Person, post::Post},
|
|
|
|
};
|
2023-07-19 13:49:41 +00:00
|
|
|
use lemmy_utils::{error::LemmyResult, SYNCHRONOUS_FEDERATION};
|
|
|
|
use once_cell::sync::{Lazy, OnceCell};
|
2023-07-20 15:36:48 +00:00
|
|
|
use tokio::{
|
|
|
|
sync::{
|
|
|
|
mpsc,
|
|
|
|
mpsc::{UnboundedReceiver, UnboundedSender, WeakUnboundedSender},
|
|
|
|
Mutex,
|
|
|
|
},
|
|
|
|
task::JoinHandle,
|
2023-07-19 13:49:41 +00:00
|
|
|
};
|
|
|
|
|
|
|
|
type MatchOutgoingActivitiesBoxed =
|
|
|
|
Box<for<'a> fn(SendActivityData, &'a Data<LemmyContext>) -> BoxFuture<'a, LemmyResult<()>>>;
|
|
|
|
|
|
|
|
/// This static is necessary so that activities can be sent out synchronously for tests.
|
|
|
|
pub static MATCH_OUTGOING_ACTIVITIES: OnceCell<MatchOutgoingActivitiesBoxed> = OnceCell::new();
|
|
|
|
|
|
|
|
#[derive(Debug)]
|
|
|
|
pub enum SendActivityData {
|
|
|
|
CreatePost(Post),
|
2023-08-01 13:53:36 +00:00
|
|
|
UpdatePost(Post),
|
2023-07-28 14:39:38 +00:00
|
|
|
CreateComment(Comment),
|
2023-08-01 13:53:36 +00:00
|
|
|
DeleteComment(Comment, Person, Community),
|
|
|
|
RemoveComment(Comment, Person, Community, Option<String>),
|
|
|
|
UpdateComment(Comment),
|
|
|
|
LikePostOrComment(DbUrl, Person, Community, i16),
|
2023-07-19 13:49:41 +00:00
|
|
|
}
|
|
|
|
|
2023-07-20 15:36:48 +00:00
|
|
|
// TODO: instead of static, move this into LemmyContext. make sure that stopping the process with
|
|
|
|
// ctrl+c still works.
|
2023-07-19 13:49:41 +00:00
|
|
|
static ACTIVITY_CHANNEL: Lazy<ActivityChannel> = Lazy::new(|| {
|
|
|
|
let (sender, receiver) = mpsc::unbounded_channel();
|
2023-07-20 15:36:48 +00:00
|
|
|
let weak_sender = sender.downgrade();
|
2023-07-19 13:49:41 +00:00
|
|
|
ActivityChannel {
|
2023-07-20 15:36:48 +00:00
|
|
|
weak_sender,
|
2023-07-19 13:49:41 +00:00
|
|
|
receiver: Mutex::new(receiver),
|
2023-07-20 15:36:48 +00:00
|
|
|
keepalive_sender: Mutex::new(Some(sender)),
|
2023-07-19 13:49:41 +00:00
|
|
|
}
|
|
|
|
});
|
|
|
|
|
|
|
|
pub struct ActivityChannel {
|
2023-07-20 15:36:48 +00:00
|
|
|
weak_sender: WeakUnboundedSender<SendActivityData>,
|
2023-07-19 13:49:41 +00:00
|
|
|
receiver: Mutex<UnboundedReceiver<SendActivityData>>,
|
2023-07-20 15:36:48 +00:00
|
|
|
keepalive_sender: Mutex<Option<UnboundedSender<SendActivityData>>>,
|
2023-07-19 13:49:41 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
impl ActivityChannel {
|
|
|
|
pub async fn retrieve_activity() -> Option<SendActivityData> {
|
|
|
|
let mut lock = ACTIVITY_CHANNEL.receiver.lock().await;
|
|
|
|
lock.recv().await
|
|
|
|
}
|
|
|
|
|
|
|
|
pub async fn submit_activity(
|
|
|
|
data: SendActivityData,
|
|
|
|
context: &Data<LemmyContext>,
|
|
|
|
) -> LemmyResult<()> {
|
|
|
|
if *SYNCHRONOUS_FEDERATION {
|
|
|
|
MATCH_OUTGOING_ACTIVITIES
|
|
|
|
.get()
|
|
|
|
.expect("retrieve function pointer")(data, context)
|
|
|
|
.await?;
|
|
|
|
}
|
2023-07-20 15:36:48 +00:00
|
|
|
// could do `ACTIVITY_CHANNEL.keepalive_sender.lock()` instead and get rid of weak_sender,
|
|
|
|
// not sure which way is more efficient
|
|
|
|
else if let Some(sender) = ACTIVITY_CHANNEL.weak_sender.upgrade() {
|
|
|
|
sender.send(data)?;
|
|
|
|
}
|
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
|
|
|
|
pub async fn close(outgoing_activities_task: JoinHandle<LemmyResult<()>>) -> LemmyResult<()> {
|
|
|
|
ACTIVITY_CHANNEL.keepalive_sender.lock().await.take();
|
|
|
|
outgoing_activities_task.await??;
|
2023-07-19 13:49:41 +00:00
|
|
|
Ok(())
|
|
|
|
}
|
|
|
|
}
|