mirror of
https://github.com/LemmyNet/lemmy.git
synced 2024-06-27 01:30:33 +00:00
ccc122100e
I noticed that stopping the Lemmy process with ctrl+c wasnt working because the activity channel isnt properly closed. This is now fixed. Later we should also move the channel from static into LemmyContext, Im not doing that now to avoid conflicts with #3670.
75 lines
2.3 KiB
Rust
75 lines
2.3 KiB
Rust
use crate::context::LemmyContext;
|
|
use activitypub_federation::config::Data;
|
|
use futures::future::BoxFuture;
|
|
use lemmy_db_schema::source::post::Post;
|
|
use lemmy_utils::{error::LemmyResult, SYNCHRONOUS_FEDERATION};
|
|
use once_cell::sync::{Lazy, OnceCell};
|
|
use tokio::{
|
|
sync::{
|
|
mpsc,
|
|
mpsc::{UnboundedReceiver, UnboundedSender, WeakUnboundedSender},
|
|
Mutex,
|
|
},
|
|
task::JoinHandle,
|
|
};
|
|
|
|
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),
|
|
}
|
|
|
|
// TODO: instead of static, move this into LemmyContext. make sure that stopping the process with
|
|
// ctrl+c still works.
|
|
static ACTIVITY_CHANNEL: Lazy<ActivityChannel> = Lazy::new(|| {
|
|
let (sender, receiver) = mpsc::unbounded_channel();
|
|
let weak_sender = sender.downgrade();
|
|
ActivityChannel {
|
|
weak_sender,
|
|
receiver: Mutex::new(receiver),
|
|
keepalive_sender: Mutex::new(Some(sender)),
|
|
}
|
|
});
|
|
|
|
pub struct ActivityChannel {
|
|
weak_sender: WeakUnboundedSender<SendActivityData>,
|
|
receiver: Mutex<UnboundedReceiver<SendActivityData>>,
|
|
keepalive_sender: Mutex<Option<UnboundedSender<SendActivityData>>>,
|
|
}
|
|
|
|
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?;
|
|
}
|
|
// 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??;
|
|
Ok(())
|
|
}
|
|
}
|