TudbuT/conduit
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Revert "dont use loole for sending channel code"

Browse source 
This reverts commit d0a9666a29.
This commit is contained in:
strawberry 2024-04-17 15:16:01 -04:00 • committed by June
parent a64cbd0304
commit bd25709446
1 changed files with 15 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -25,7 +25,7 @@ use ruma::{
events::{push_rules::PushRulesEvent, receipt::ReceiptType, AnySyncEphemeralRoomEvent, GlobalAccountDataEventType},
push, uint, MilliSecondsSinceUnixEpoch, OwnedServerName, OwnedUserId, RoomId, ServerName, UInt, UserId,
};
use tokio::sync::{mpsc, Mutex, Semaphore};
use tokio::sync::{Mutex, Semaphore};
use tracing::{error, warn};
use crate::{service::presence::Presence, services, utils::calculate_hash, Config, Error, PduEvent, Result};
@ -42,8 +42,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,
@ -72,7 +72,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,
@ -274,7 +274,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();
@ -342,13 +342,16 @@ 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() => {
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));
}
}
}
}