1
0
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2025-07-02 16:38:36 +00:00
This commit is contained in:
Jonathan de Jong 2021-07-04 02:15:04 +02:00
parent a55dec9035
commit 5ec0be2b41
4 changed files with 27 additions and 11 deletions

View file

@ -340,7 +340,9 @@ impl Database {
guard.rooms.edus.presenceid_presence.clear()?; guard.rooms.edus.presenceid_presence.clear()?;
guard.admin.start_handler(Arc::clone(&db), admin_receiver); guard.admin.start_handler(Arc::clone(&db), admin_receiver);
guard.sending.start_handler(Arc::clone(&db), sending_receiver); guard
.sending
.start_handler(Arc::clone(&db), sending_receiver);
drop(guard); drop(guard);

View file

@ -35,8 +35,9 @@ impl Admin {
let guard = db.read().await; let guard = db.read().await;
let conduit_user = UserId::try_from(format!("@conduit:{}", guard.globals.server_name())) let conduit_user =
.expect("@conduit:server_name is valid"); UserId::try_from(format!("@conduit:{}", guard.globals.server_name()))
.expect("@conduit:server_name is valid");
let conduit_room = guard let conduit_room = guard
.rooms .rooms
@ -53,9 +54,7 @@ impl Admin {
drop(guard); drop(guard);
let send_message = |message: message::MessageEventContent| { let send_message = |message: message::MessageEventContent| {};
};
loop { loop {
tokio::select! { tokio::select! {

View file

@ -30,7 +30,10 @@ use ruma::{
receipt::ReceiptType, receipt::ReceiptType,
MilliSecondsSinceUnixEpoch, ServerName, UInt, UserId, MilliSecondsSinceUnixEpoch, ServerName, UInt, UserId,
}; };
use tokio::{select, sync::{Semaphore, RwLock}}; use tokio::{
select,
sync::{RwLock, Semaphore},
};
use super::abstraction::Tree; use super::abstraction::Tree;
@ -90,7 +93,11 @@ enum TransactionStatus {
} }
impl Sending { impl Sending {
pub fn start_handler(&self, db: Arc<RwLock<Database>>, mut receiver: mpsc::UnboundedReceiver<Vec<u8>>) { pub fn start_handler(
&self,
db: Arc<RwLock<Database>>,
mut receiver: mpsc::UnboundedReceiver<Vec<u8>>,
) {
tokio::spawn(async move { tokio::spawn(async move {
let mut futures = FuturesUnordered::new(); let mut futures = FuturesUnordered::new();
@ -102,7 +109,8 @@ impl Sending {
let guard = db.read().await; let guard = db.read().await;
for (key, outgoing_kind, event) in for (key, outgoing_kind, event) in
guard.sending guard
.sending
.servercurrentevents .servercurrentevents
.iter() .iter()
.filter_map(|(key, _)| { .filter_map(|(key, _)| {
@ -132,7 +140,11 @@ impl Sending {
for (outgoing_kind, events) in initial_transactions { for (outgoing_kind, events) in initial_transactions {
current_transaction_status current_transaction_status
.insert(outgoing_kind.get_prefix(), TransactionStatus::Running); .insert(outgoing_kind.get_prefix(), TransactionStatus::Running);
futures.push(Self::handle_events(outgoing_kind.clone(), events, Arc::clone(&db))); futures.push(Self::handle_events(
outgoing_kind.clone(),
events,
Arc::clone(&db),
));
} }
loop { loop {

View file

@ -12,7 +12,10 @@ mod pdu;
mod ruma_wrapper; mod ruma_wrapper;
mod utils; mod utils;
use std::{sync::{Arc, Weak}, time::{Duration, Instant}}; use std::{
sync::{Arc, Weak},
time::{Duration, Instant},
};
use database::Config; use database::Config;
pub use database::Database; pub use database::Database;