git-mirrors/continuwuity
1
0
Fork 0
mirror of https://forgejo.ellis.link/continuwuation/continuwuity.git synced 2025-07-13 11:37:39 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

move all other services to loole channels

Browse source
This commit is contained in:
iraizo 2024-04-06 18:13:05 +02:00
parent 546d0efe21
commit ce0ac32143
No known key found for this signature in database
GPG key ID: 12A9EF3A7EA88D6C
3 changed files with 34 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
src/service/admin/mod.rs
View file

@ -91,7 +91,7 @@ pub enum AdminRoomEvent {
}
pub struct Service {
pub sender: loole::Sender<AdminRoomEvent>, //mpsc::UnboundedSender<AdminRoomEvent>,
pub sender: loole::Sender<AdminRoomEvent>,
receiver: Mutex<loole::Receiver<AdminRoomEvent>>,
}

26
src/service/presence/mod.rs
View file

@ -10,10 +10,7 @@ use ruma::{
OwnedUserId, UInt, UserId,
};
use serde::{Deserialize, Serialize};
use tokio::{
sync::{mpsc, Mutex},
time::sleep,
};
use tokio::{sync::Mutex, time::sleep};
use tracing::{debug, error};
use crate::{services, utils, Config, Error, Result};
@ -71,14 +68,14 @@ impl Presence {
pub struct Service {
pub db: &'static dyn Data,
pub timer_sender: mpsc::UnboundedSender<(OwnedUserId, Duration)>,
timer_receiver: Mutex<mpsc::UnboundedReceiver<(OwnedUserId, Duration)>>,
pub timer_sender: loole::Sender<(OwnedUserId, Duration)>,
timer_receiver: Mutex<loole::Receiver<(OwnedUserId, Duration)>>,
timeout_remote_users: bool,
}
impl Service {
pub fn build(db: &'static dyn Data, config: &Config) -> Arc<Self> {
let (timer_sender, timer_receiver) = mpsc::unbounded_channel();
let (timer_sender, timer_receiver) = loole::unbounded();
Arc::new(Self {
db,
@ -176,9 +173,18 @@ impl Service {
let mut receiver = self.timer_receiver.lock().await;
loop {
tokio::select! {
Some((user_id, timeout)) = receiver.recv() => {
debug!("Adding timer {}: {user_id} timeout:{timeout:?}", presence_timers.len());
presence_timers.push(presence_timer(user_id, timeout));
event = receiver.recv_async() => {
match event {
Ok((user_id, timeout)) => {
debug!("Adding timer {}: {user_id} timeout:{timeout:?}", presence_timers.len());
presence_timers.push(presence_timer(user_id, timeout));
}
Err(e) => {
// TODO: Handle error better? I have no idea what to do here.
error!("Failed to receive presence timer: {}", e);
}
}
}
Some(user_id) = presence_timers.next() => {

30
src/service/sending/mod.rs
View file

@ -27,7 +27,7 @@ use ruma::{
};
use tokio::{
select,
sync::{mpsc, Mutex, Semaphore},
sync::{Mutex, Semaphore},
};
use tracing::{error, warn};
@ -43,8 +43,8 @@ pub struct Service {
/// The state for a given state hash.
pub(super) maximum_requests: Arc<Semaphore>,
pub sender: mpsc::UnboundedSender<(OutgoingKind, SendingEventType, Vec<u8>)>,
receiver: Mutex<mpsc::UnboundedReceiver<(OutgoingKind, SendingEventType, Vec<u8>)>>,
pub sender: loole::Sender<(OutgoingKind, SendingEventType, Vec<u8>)>,
receiver: Mutex<loole::Receiver<(OutgoingKind, SendingEventType, Vec<u8>)>>,
startup_netburst: bool,
startup_netburst_keep: i64,
timeout: u64,
@ -73,7 +73,7 @@ enum TransactionStatus {
impl Service {
pub fn build(db: &'static dyn Data, config: &Config) -> Arc<Self> {
let (sender, receiver) = mpsc::unbounded_channel();
let (sender, receiver) = loole::unbounded();
Arc::new(Self {
db,
sender,
@ -275,7 +275,7 @@ impl Service {
#[tracing::instrument(skip(self), name = "sender")]
async fn handler(&self) -> Result<()> {
let mut receiver = self.receiver.lock().await;
let receiver = self.receiver.lock().await;
let mut futures = FuturesUnordered::new();
let mut current_transaction_status = HashMap::<OutgoingKind, TransactionStatus>::new();
@ -306,7 +306,7 @@ impl Service {
}
loop {
select! {
tokio::select! {
Some(response) = futures.next() => {
match response {
Ok(outgoing_kind) => {
@ -343,13 +343,17 @@ impl Service {
}
};
},
Some((outgoing_kind, event, key)) = receiver.recv() => {
if let Ok(Some(events)) = self.select_events(
&outgoing_kind,
vec![(event, key)],
&mut current_transaction_status,
) {
futures.push(handle_events(outgoing_kind, events));
event = receiver.recv_async() => {
// TODO: Error handling for this
if let Ok((outgoing_kind, event, key)) = event {
if let Ok(Some(events)) = self.select_events(
&outgoing_kind,
vec![(event, key)],
&mut current_transaction_status,
) {
futures.push(handle_events(outgoing_kind, events));
}
}
}
}