rename OutgoingKind enum to OutgoingDestination
Signed-off-by: strawberry <strawberry@puppygock.gay>
This commit is contained in:
parent
0376b58006
commit
395b466b4a
3 changed files with 52 additions and 49 deletions
|
@ -4,7 +4,7 @@ use crate::{
|
||||||
database::KeyValueDatabase,
|
database::KeyValueDatabase,
|
||||||
service::{
|
service::{
|
||||||
self,
|
self,
|
||||||
sending::{OutgoingKind, SendingEventType},
|
sending::{OutgoingDestination, SendingEventType},
|
||||||
},
|
},
|
||||||
services, utils, Error, Result,
|
services, utils, Error, Result,
|
||||||
};
|
};
|
||||||
|
@ -12,7 +12,7 @@ use crate::{
|
||||||
impl service::sending::Data for KeyValueDatabase {
|
impl service::sending::Data for KeyValueDatabase {
|
||||||
fn active_requests<'a>(
|
fn active_requests<'a>(
|
||||||
&'a self,
|
&'a self,
|
||||||
) -> Box<dyn Iterator<Item = Result<(Vec<u8>, OutgoingKind, SendingEventType)>> + 'a> {
|
) -> Box<dyn Iterator<Item = Result<(Vec<u8>, OutgoingDestination, SendingEventType)>> + 'a> {
|
||||||
Box::new(
|
Box::new(
|
||||||
self.servercurrentevent_data
|
self.servercurrentevent_data
|
||||||
.iter()
|
.iter()
|
||||||
|
@ -21,7 +21,7 @@ impl service::sending::Data for KeyValueDatabase {
|
||||||
}
|
}
|
||||||
|
|
||||||
fn active_requests_for<'a>(
|
fn active_requests_for<'a>(
|
||||||
&'a self, outgoing_kind: &OutgoingKind,
|
&'a self, outgoing_kind: &OutgoingDestination,
|
||||||
) -> Box<dyn Iterator<Item = Result<(Vec<u8>, SendingEventType)>> + 'a> {
|
) -> Box<dyn Iterator<Item = Result<(Vec<u8>, SendingEventType)>> + 'a> {
|
||||||
let prefix = outgoing_kind.get_prefix();
|
let prefix = outgoing_kind.get_prefix();
|
||||||
Box::new(
|
Box::new(
|
||||||
|
@ -33,7 +33,7 @@ impl service::sending::Data for KeyValueDatabase {
|
||||||
|
|
||||||
fn delete_active_request(&self, key: Vec<u8>) -> Result<()> { self.servercurrentevent_data.remove(&key) }
|
fn delete_active_request(&self, key: Vec<u8>) -> Result<()> { self.servercurrentevent_data.remove(&key) }
|
||||||
|
|
||||||
fn delete_all_active_requests_for(&self, outgoing_kind: &OutgoingKind) -> Result<()> {
|
fn delete_all_active_requests_for(&self, outgoing_kind: &OutgoingDestination) -> Result<()> {
|
||||||
let prefix = outgoing_kind.get_prefix();
|
let prefix = outgoing_kind.get_prefix();
|
||||||
for (key, _) in self.servercurrentevent_data.scan_prefix(prefix) {
|
for (key, _) in self.servercurrentevent_data.scan_prefix(prefix) {
|
||||||
self.servercurrentevent_data.remove(&key)?;
|
self.servercurrentevent_data.remove(&key)?;
|
||||||
|
@ -42,7 +42,7 @@ impl service::sending::Data for KeyValueDatabase {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn delete_all_requests_for(&self, outgoing_kind: &OutgoingKind) -> Result<()> {
|
fn delete_all_requests_for(&self, outgoing_kind: &OutgoingDestination) -> Result<()> {
|
||||||
let prefix = outgoing_kind.get_prefix();
|
let prefix = outgoing_kind.get_prefix();
|
||||||
for (key, _) in self.servercurrentevent_data.scan_prefix(prefix.clone()) {
|
for (key, _) in self.servercurrentevent_data.scan_prefix(prefix.clone()) {
|
||||||
self.servercurrentevent_data.remove(&key).unwrap();
|
self.servercurrentevent_data.remove(&key).unwrap();
|
||||||
|
@ -55,7 +55,7 @@ impl service::sending::Data for KeyValueDatabase {
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
fn queue_requests(&self, requests: &[(&OutgoingKind, SendingEventType)]) -> Result<Vec<Vec<u8>>> {
|
fn queue_requests(&self, requests: &[(&OutgoingDestination, SendingEventType)]) -> Result<Vec<Vec<u8>>> {
|
||||||
let mut batch = Vec::new();
|
let mut batch = Vec::new();
|
||||||
let mut keys = Vec::new();
|
let mut keys = Vec::new();
|
||||||
for (outgoing_kind, event) in requests {
|
for (outgoing_kind, event) in requests {
|
||||||
|
@ -79,7 +79,7 @@ impl service::sending::Data for KeyValueDatabase {
|
||||||
}
|
}
|
||||||
|
|
||||||
fn queued_requests<'a>(
|
fn queued_requests<'a>(
|
||||||
&'a self, outgoing_kind: &OutgoingKind,
|
&'a self, outgoing_kind: &OutgoingDestination,
|
||||||
) -> Box<dyn Iterator<Item = Result<(SendingEventType, Vec<u8>)>> + 'a> {
|
) -> Box<dyn Iterator<Item = Result<(SendingEventType, Vec<u8>)>> + 'a> {
|
||||||
let prefix = outgoing_kind.get_prefix();
|
let prefix = outgoing_kind.get_prefix();
|
||||||
return Box::new(
|
return Box::new(
|
||||||
|
@ -122,7 +122,7 @@ impl service::sending::Data for KeyValueDatabase {
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tracing::instrument(skip(key))]
|
#[tracing::instrument(skip(key))]
|
||||||
fn parse_servercurrentevent(key: &[u8], value: Vec<u8>) -> Result<(OutgoingKind, SendingEventType)> {
|
fn parse_servercurrentevent(key: &[u8], value: Vec<u8>) -> Result<(OutgoingDestination, SendingEventType)> {
|
||||||
// Appservices start with a plus
|
// Appservices start with a plus
|
||||||
Ok::<_, Error>(if key.starts_with(b"+") {
|
Ok::<_, Error>(if key.starts_with(b"+") {
|
||||||
let mut parts = key[1..].splitn(2, |&b| b == 0xFF);
|
let mut parts = key[1..].splitn(2, |&b| b == 0xFF);
|
||||||
|
@ -136,7 +136,7 @@ fn parse_servercurrentevent(key: &[u8], value: Vec<u8>) -> Result<(OutgoingKind,
|
||||||
.map_err(|_| Error::bad_database("Invalid server bytes in server_currenttransaction"))?;
|
.map_err(|_| Error::bad_database("Invalid server bytes in server_currenttransaction"))?;
|
||||||
|
|
||||||
(
|
(
|
||||||
OutgoingKind::Appservice(server),
|
OutgoingDestination::Appservice(server),
|
||||||
if value.is_empty() {
|
if value.is_empty() {
|
||||||
SendingEventType::Pdu(event.to_vec())
|
SendingEventType::Pdu(event.to_vec())
|
||||||
} else {
|
} else {
|
||||||
|
@ -163,7 +163,7 @@ fn parse_servercurrentevent(key: &[u8], value: Vec<u8>) -> Result<(OutgoingKind,
|
||||||
.ok_or_else(|| Error::bad_database("Invalid bytes in servercurrentpdus."))?;
|
.ok_or_else(|| Error::bad_database("Invalid bytes in servercurrentpdus."))?;
|
||||||
|
|
||||||
(
|
(
|
||||||
OutgoingKind::Push(user_id, pushkey_string),
|
OutgoingDestination::Push(user_id, pushkey_string),
|
||||||
if value.is_empty() {
|
if value.is_empty() {
|
||||||
SendingEventType::Pdu(event.to_vec())
|
SendingEventType::Pdu(event.to_vec())
|
||||||
} else {
|
} else {
|
||||||
|
@ -183,7 +183,7 @@ fn parse_servercurrentevent(key: &[u8], value: Vec<u8>) -> Result<(OutgoingKind,
|
||||||
.map_err(|_| Error::bad_database("Invalid server bytes in server_currenttransaction"))?;
|
.map_err(|_| Error::bad_database("Invalid server bytes in server_currenttransaction"))?;
|
||||||
|
|
||||||
(
|
(
|
||||||
OutgoingKind::Normal(
|
OutgoingDestination::Normal(
|
||||||
ServerName::parse(server)
|
ServerName::parse(server)
|
||||||
.map_err(|_| Error::bad_database("Invalid server string in server_currenttransaction"))?,
|
.map_err(|_| Error::bad_database("Invalid server string in server_currenttransaction"))?,
|
||||||
),
|
),
|
||||||
|
|
|
@ -1,20 +1,20 @@
|
||||||
use ruma::ServerName;
|
use ruma::ServerName;
|
||||||
|
|
||||||
use super::{OutgoingKind, SendingEventType};
|
use super::{OutgoingDestination, SendingEventType};
|
||||||
use crate::Result;
|
use crate::Result;
|
||||||
|
|
||||||
type OutgoingSendingIter<'a> = Box<dyn Iterator<Item = Result<(Vec<u8>, OutgoingKind, SendingEventType)>> + 'a>;
|
type OutgoingSendingIter<'a> = Box<dyn Iterator<Item = Result<(Vec<u8>, OutgoingDestination, SendingEventType)>> + 'a>;
|
||||||
type SendingEventTypeIter<'a> = Box<dyn Iterator<Item = Result<(Vec<u8>, SendingEventType)>> + 'a>;
|
type SendingEventTypeIter<'a> = Box<dyn Iterator<Item = Result<(Vec<u8>, SendingEventType)>> + 'a>;
|
||||||
|
|
||||||
pub trait Data: Send + Sync {
|
pub trait Data: Send + Sync {
|
||||||
fn active_requests(&self) -> OutgoingSendingIter<'_>;
|
fn active_requests(&self) -> OutgoingSendingIter<'_>;
|
||||||
fn active_requests_for(&self, outgoing_kind: &OutgoingKind) -> SendingEventTypeIter<'_>;
|
fn active_requests_for(&self, outgoing_kind: &OutgoingDestination) -> SendingEventTypeIter<'_>;
|
||||||
fn delete_active_request(&self, key: Vec<u8>) -> Result<()>;
|
fn delete_active_request(&self, key: Vec<u8>) -> Result<()>;
|
||||||
fn delete_all_active_requests_for(&self, outgoing_kind: &OutgoingKind) -> Result<()>;
|
fn delete_all_active_requests_for(&self, outgoing_kind: &OutgoingDestination) -> Result<()>;
|
||||||
fn delete_all_requests_for(&self, outgoing_kind: &OutgoingKind) -> Result<()>;
|
fn delete_all_requests_for(&self, outgoing_kind: &OutgoingDestination) -> Result<()>;
|
||||||
fn queue_requests(&self, requests: &[(&OutgoingKind, SendingEventType)]) -> Result<Vec<Vec<u8>>>;
|
fn queue_requests(&self, requests: &[(&OutgoingDestination, SendingEventType)]) -> Result<Vec<Vec<u8>>>;
|
||||||
fn queued_requests<'a>(
|
fn queued_requests<'a>(
|
||||||
&'a self, outgoing_kind: &OutgoingKind,
|
&'a self, outgoing_kind: &OutgoingDestination,
|
||||||
) -> Box<dyn Iterator<Item = Result<(SendingEventType, Vec<u8>)>> + 'a>;
|
) -> Box<dyn Iterator<Item = Result<(SendingEventType, Vec<u8>)>> + 'a>;
|
||||||
fn mark_as_active(&self, events: &[(SendingEventType, Vec<u8>)]) -> Result<()>;
|
fn mark_as_active(&self, events: &[(SendingEventType, Vec<u8>)]) -> Result<()>;
|
||||||
fn set_latest_educount(&self, server_name: &ServerName, educount: u64) -> Result<()>;
|
fn set_latest_educount(&self, server_name: &ServerName, educount: u64) -> Result<()>;
|
||||||
|
|
|
@ -42,15 +42,15 @@ pub struct Service {
|
||||||
|
|
||||||
/// The state for a given state hash.
|
/// The state for a given state hash.
|
||||||
pub(super) maximum_requests: Arc<Semaphore>,
|
pub(super) maximum_requests: Arc<Semaphore>,
|
||||||
pub sender: loole::Sender<(OutgoingKind, SendingEventType, Vec<u8>)>,
|
pub sender: loole::Sender<(OutgoingDestination, SendingEventType, Vec<u8>)>,
|
||||||
receiver: Mutex<loole::Receiver<(OutgoingKind, SendingEventType, Vec<u8>)>>,
|
receiver: Mutex<loole::Receiver<(OutgoingDestination, SendingEventType, Vec<u8>)>>,
|
||||||
startup_netburst: bool,
|
startup_netburst: bool,
|
||||||
startup_netburst_keep: i64,
|
startup_netburst_keep: i64,
|
||||||
timeout: u64,
|
timeout: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
|
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
|
||||||
pub enum OutgoingKind {
|
pub enum OutgoingDestination {
|
||||||
Appservice(String),
|
Appservice(String),
|
||||||
Push(OwnedUserId, String), // user and pushkey
|
Push(OwnedUserId, String), // user and pushkey
|
||||||
Normal(OwnedServerName),
|
Normal(OwnedServerName),
|
||||||
|
@ -86,7 +86,7 @@ impl Service {
|
||||||
|
|
||||||
#[tracing::instrument(skip(self, pdu_id, user, pushkey))]
|
#[tracing::instrument(skip(self, pdu_id, user, pushkey))]
|
||||||
pub fn send_pdu_push(&self, pdu_id: &[u8], user: &UserId, pushkey: String) -> Result<()> {
|
pub fn send_pdu_push(&self, pdu_id: &[u8], user: &UserId, pushkey: String) -> Result<()> {
|
||||||
let outgoing_kind = OutgoingKind::Push(user.to_owned(), pushkey);
|
let outgoing_kind = OutgoingDestination::Push(user.to_owned(), pushkey);
|
||||||
let event = SendingEventType::Pdu(pdu_id.to_owned());
|
let event = SendingEventType::Pdu(pdu_id.to_owned());
|
||||||
let _cork = services().globals.db.cork()?;
|
let _cork = services().globals.db.cork()?;
|
||||||
let keys = self.db.queue_requests(&[(&outgoing_kind, event.clone())])?;
|
let keys = self.db.queue_requests(&[(&outgoing_kind, event.clone())])?;
|
||||||
|
@ -99,7 +99,7 @@ impl Service {
|
||||||
|
|
||||||
#[tracing::instrument(skip(self))]
|
#[tracing::instrument(skip(self))]
|
||||||
pub fn send_pdu_appservice(&self, appservice_id: String, pdu_id: Vec<u8>) -> Result<()> {
|
pub fn send_pdu_appservice(&self, appservice_id: String, pdu_id: Vec<u8>) -> Result<()> {
|
||||||
let outgoing_kind = OutgoingKind::Appservice(appservice_id);
|
let outgoing_kind = OutgoingDestination::Appservice(appservice_id);
|
||||||
let event = SendingEventType::Pdu(pdu_id);
|
let event = SendingEventType::Pdu(pdu_id);
|
||||||
let _cork = services().globals.db.cork()?;
|
let _cork = services().globals.db.cork()?;
|
||||||
let keys = self.db.queue_requests(&[(&outgoing_kind, event.clone())])?;
|
let keys = self.db.queue_requests(&[(&outgoing_kind, event.clone())])?;
|
||||||
|
@ -126,7 +126,7 @@ impl Service {
|
||||||
pub fn send_pdu_servers<I: Iterator<Item = OwnedServerName>>(&self, servers: I, pdu_id: &[u8]) -> Result<()> {
|
pub fn send_pdu_servers<I: Iterator<Item = OwnedServerName>>(&self, servers: I, pdu_id: &[u8]) -> Result<()> {
|
||||||
let requests = servers
|
let requests = servers
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|server| (OutgoingKind::Normal(server), SendingEventType::Pdu(pdu_id.to_owned())))
|
.map(|server| (OutgoingDestination::Normal(server), SendingEventType::Pdu(pdu_id.to_owned())))
|
||||||
.collect::<Vec<_>>();
|
.collect::<Vec<_>>();
|
||||||
let _cork = services().globals.db.cork()?;
|
let _cork = services().globals.db.cork()?;
|
||||||
let keys = self.db.queue_requests(
|
let keys = self.db.queue_requests(
|
||||||
|
@ -146,7 +146,7 @@ impl Service {
|
||||||
|
|
||||||
#[tracing::instrument(skip(self, server, serialized))]
|
#[tracing::instrument(skip(self, server, serialized))]
|
||||||
pub fn send_edu_server(&self, server: &ServerName, serialized: Vec<u8>) -> Result<()> {
|
pub fn send_edu_server(&self, server: &ServerName, serialized: Vec<u8>) -> Result<()> {
|
||||||
let outgoing_kind = OutgoingKind::Normal(server.to_owned());
|
let outgoing_kind = OutgoingDestination::Normal(server.to_owned());
|
||||||
let event = SendingEventType::Edu(serialized);
|
let event = SendingEventType::Edu(serialized);
|
||||||
let _cork = services().globals.db.cork()?;
|
let _cork = services().globals.db.cork()?;
|
||||||
let keys = self.db.queue_requests(&[(&outgoing_kind, event.clone())])?;
|
let keys = self.db.queue_requests(&[(&outgoing_kind, event.clone())])?;
|
||||||
|
@ -173,7 +173,7 @@ impl Service {
|
||||||
pub fn send_edu_servers<I: Iterator<Item = OwnedServerName>>(&self, servers: I, serialized: Vec<u8>) -> Result<()> {
|
pub fn send_edu_servers<I: Iterator<Item = OwnedServerName>>(&self, servers: I, serialized: Vec<u8>) -> Result<()> {
|
||||||
let requests = servers
|
let requests = servers
|
||||||
.into_iter()
|
.into_iter()
|
||||||
.map(|server| (OutgoingKind::Normal(server), SendingEventType::Edu(serialized.clone())))
|
.map(|server| (OutgoingDestination::Normal(server), SendingEventType::Edu(serialized.clone())))
|
||||||
.collect::<Vec<_>>();
|
.collect::<Vec<_>>();
|
||||||
let _cork = services().globals.db.cork()?;
|
let _cork = services().globals.db.cork()?;
|
||||||
let keys = self.db.queue_requests(
|
let keys = self.db.queue_requests(
|
||||||
|
@ -205,7 +205,7 @@ impl Service {
|
||||||
|
|
||||||
#[tracing::instrument(skip(self, servers))]
|
#[tracing::instrument(skip(self, servers))]
|
||||||
pub fn flush_servers<I: Iterator<Item = OwnedServerName>>(&self, servers: I) -> Result<()> {
|
pub fn flush_servers<I: Iterator<Item = OwnedServerName>>(&self, servers: I) -> Result<()> {
|
||||||
let requests = servers.into_iter().map(OutgoingKind::Normal);
|
let requests = servers.into_iter().map(OutgoingDestination::Normal);
|
||||||
|
|
||||||
for outgoing_kind in requests {
|
for outgoing_kind in requests {
|
||||||
self.sender
|
self.sender
|
||||||
|
@ -221,7 +221,7 @@ impl Service {
|
||||||
#[tracing::instrument(skip(self))]
|
#[tracing::instrument(skip(self))]
|
||||||
pub fn cleanup_events(&self, appservice_id: String) -> Result<()> {
|
pub fn cleanup_events(&self, appservice_id: String) -> Result<()> {
|
||||||
self.db
|
self.db
|
||||||
.delete_all_requests_for(&OutgoingKind::Appservice(appservice_id))?;
|
.delete_all_requests_for(&OutgoingDestination::Appservice(appservice_id))?;
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
@ -277,11 +277,11 @@ impl Service {
|
||||||
let receiver = self.receiver.lock().await;
|
let receiver = self.receiver.lock().await;
|
||||||
|
|
||||||
let mut futures = FuturesUnordered::new();
|
let mut futures = FuturesUnordered::new();
|
||||||
let mut current_transaction_status = HashMap::<OutgoingKind, TransactionStatus>::new();
|
let mut current_transaction_status = HashMap::<OutgoingDestination, TransactionStatus>::new();
|
||||||
|
|
||||||
// Retry requests we could not finish yet
|
// Retry requests we could not finish yet
|
||||||
if self.startup_netburst {
|
if self.startup_netburst {
|
||||||
let mut initial_transactions = HashMap::<OutgoingKind, Vec<SendingEventType>>::new();
|
let mut initial_transactions = HashMap::<OutgoingDestination, Vec<SendingEventType>>::new();
|
||||||
for (key, outgoing_kind, event) in self.db.active_requests().filter_map(Result::ok) {
|
for (key, outgoing_kind, event) in self.db.active_requests().filter_map(Result::ok) {
|
||||||
let entry = initial_transactions
|
let entry = initial_transactions
|
||||||
.entry(outgoing_kind.clone())
|
.entry(outgoing_kind.clone())
|
||||||
|
@ -361,9 +361,9 @@ impl Service {
|
||||||
#[tracing::instrument(skip(self, outgoing_kind, new_events, current_transaction_status))]
|
#[tracing::instrument(skip(self, outgoing_kind, new_events, current_transaction_status))]
|
||||||
fn select_events(
|
fn select_events(
|
||||||
&self,
|
&self,
|
||||||
outgoing_kind: &OutgoingKind,
|
outgoing_kind: &OutgoingDestination,
|
||||||
new_events: Vec<(SendingEventType, Vec<u8>)>, // Events we want to send: event and full key
|
new_events: Vec<(SendingEventType, Vec<u8>)>, // Events we want to send: event and full key
|
||||||
current_transaction_status: &mut HashMap<OutgoingKind, TransactionStatus>,
|
current_transaction_status: &mut HashMap<OutgoingDestination, TransactionStatus>,
|
||||||
) -> Result<Option<Vec<SendingEventType>>> {
|
) -> Result<Option<Vec<SendingEventType>>> {
|
||||||
let (allow, retry) = self.select_events_current(outgoing_kind.clone(), current_transaction_status)?;
|
let (allow, retry) = self.select_events_current(outgoing_kind.clone(), current_transaction_status)?;
|
||||||
|
|
||||||
|
@ -395,7 +395,7 @@ impl Service {
|
||||||
}
|
}
|
||||||
|
|
||||||
// Add EDU's into the transaction
|
// Add EDU's into the transaction
|
||||||
if let OutgoingKind::Normal(server_name) = outgoing_kind {
|
if let OutgoingDestination::Normal(server_name) = outgoing_kind {
|
||||||
if let Ok((select_edus, last_count)) = self.select_edus(server_name) {
|
if let Ok((select_edus, last_count)) = self.select_edus(server_name) {
|
||||||
events.extend(select_edus.into_iter().map(SendingEventType::Edu));
|
events.extend(select_edus.into_iter().map(SendingEventType::Edu));
|
||||||
self.db.set_latest_educount(server_name, last_count)?;
|
self.db.set_latest_educount(server_name, last_count)?;
|
||||||
|
@ -407,7 +407,8 @@ impl Service {
|
||||||
|
|
||||||
#[tracing::instrument(skip(self, outgoing_kind, current_transaction_status))]
|
#[tracing::instrument(skip(self, outgoing_kind, current_transaction_status))]
|
||||||
fn select_events_current(
|
fn select_events_current(
|
||||||
&self, outgoing_kind: OutgoingKind, current_transaction_status: &mut HashMap<OutgoingKind, TransactionStatus>,
|
&self, outgoing_kind: OutgoingDestination,
|
||||||
|
current_transaction_status: &mut HashMap<OutgoingDestination, TransactionStatus>,
|
||||||
) -> Result<(bool, bool)> {
|
) -> Result<(bool, bool)> {
|
||||||
let (mut allow, mut retry) = (true, false);
|
let (mut allow, mut retry) = (true, false);
|
||||||
current_transaction_status
|
current_transaction_status
|
||||||
|
@ -597,19 +598,21 @@ pub fn select_edus_receipts(
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn handle_events(
|
async fn handle_events(
|
||||||
kind: OutgoingKind, events: Vec<SendingEventType>,
|
kind: OutgoingDestination, events: Vec<SendingEventType>,
|
||||||
) -> Result<OutgoingKind, (OutgoingKind, Error)> {
|
) -> Result<OutgoingDestination, (OutgoingDestination, Error)> {
|
||||||
match kind {
|
match kind {
|
||||||
OutgoingKind::Appservice(ref id) => handle_events_kind_appservice(&kind, id, events).await,
|
OutgoingDestination::Appservice(ref id) => handle_events_kind_appservice(&kind, id, events).await,
|
||||||
OutgoingKind::Push(ref userid, ref pushkey) => handle_events_kind_push(&kind, userid, pushkey, events).await,
|
OutgoingDestination::Push(ref userid, ref pushkey) => {
|
||||||
OutgoingKind::Normal(ref server) => handle_events_kind_normal(&kind, server, events).await,
|
handle_events_kind_push(&kind, userid, pushkey, events).await
|
||||||
|
},
|
||||||
|
OutgoingDestination::Normal(ref server) => handle_events_kind_normal(&kind, server, events).await,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[tracing::instrument(skip(kind, events))]
|
#[tracing::instrument(skip(kind, events))]
|
||||||
async fn handle_events_kind_appservice(
|
async fn handle_events_kind_appservice(
|
||||||
kind: &OutgoingKind, id: &String, events: Vec<SendingEventType>,
|
kind: &OutgoingDestination, id: &String, events: Vec<SendingEventType>,
|
||||||
) -> Result<OutgoingKind, (OutgoingKind, Error)> {
|
) -> Result<OutgoingDestination, (OutgoingDestination, Error)> {
|
||||||
let mut pdu_jsons = Vec::new();
|
let mut pdu_jsons = Vec::new();
|
||||||
|
|
||||||
for event in &events {
|
for event in &events {
|
||||||
|
@ -677,8 +680,8 @@ async fn handle_events_kind_appservice(
|
||||||
|
|
||||||
#[tracing::instrument(skip(kind, events))]
|
#[tracing::instrument(skip(kind, events))]
|
||||||
async fn handle_events_kind_push(
|
async fn handle_events_kind_push(
|
||||||
kind: &OutgoingKind, userid: &OwnedUserId, pushkey: &String, events: Vec<SendingEventType>,
|
kind: &OutgoingDestination, userid: &OwnedUserId, pushkey: &String, events: Vec<SendingEventType>,
|
||||||
) -> Result<OutgoingKind, (OutgoingKind, Error)> {
|
) -> Result<OutgoingDestination, (OutgoingDestination, Error)> {
|
||||||
let mut pdus = Vec::new();
|
let mut pdus = Vec::new();
|
||||||
|
|
||||||
for event in &events {
|
for event in &events {
|
||||||
|
@ -755,8 +758,8 @@ async fn handle_events_kind_push(
|
||||||
|
|
||||||
#[tracing::instrument(skip(kind, events), name = "")]
|
#[tracing::instrument(skip(kind, events), name = "")]
|
||||||
async fn handle_events_kind_normal(
|
async fn handle_events_kind_normal(
|
||||||
kind: &OutgoingKind, dest: &OwnedServerName, events: Vec<SendingEventType>,
|
kind: &OutgoingDestination, dest: &OwnedServerName, events: Vec<SendingEventType>,
|
||||||
) -> Result<OutgoingKind, (OutgoingKind, Error)> {
|
) -> Result<OutgoingDestination, (OutgoingDestination, Error)> {
|
||||||
let mut edu_jsons = Vec::new();
|
let mut edu_jsons = Vec::new();
|
||||||
let mut pdu_jsons = Vec::new();
|
let mut pdu_jsons = Vec::new();
|
||||||
|
|
||||||
|
@ -829,23 +832,23 @@ async fn handle_events_kind_normal(
|
||||||
response
|
response
|
||||||
}
|
}
|
||||||
|
|
||||||
impl OutgoingKind {
|
impl OutgoingDestination {
|
||||||
#[tracing::instrument(skip(self))]
|
#[tracing::instrument(skip(self))]
|
||||||
pub fn get_prefix(&self) -> Vec<u8> {
|
pub fn get_prefix(&self) -> Vec<u8> {
|
||||||
let mut prefix = match self {
|
let mut prefix = match self {
|
||||||
OutgoingKind::Appservice(server) => {
|
OutgoingDestination::Appservice(server) => {
|
||||||
let mut p = b"+".to_vec();
|
let mut p = b"+".to_vec();
|
||||||
p.extend_from_slice(server.as_bytes());
|
p.extend_from_slice(server.as_bytes());
|
||||||
p
|
p
|
||||||
},
|
},
|
||||||
OutgoingKind::Push(user, pushkey) => {
|
OutgoingDestination::Push(user, pushkey) => {
|
||||||
let mut p = b"$".to_vec();
|
let mut p = b"$".to_vec();
|
||||||
p.extend_from_slice(user.as_bytes());
|
p.extend_from_slice(user.as_bytes());
|
||||||
p.push(0xFF);
|
p.push(0xFF);
|
||||||
p.extend_from_slice(pushkey.as_bytes());
|
p.extend_from_slice(pushkey.as_bytes());
|
||||||
p
|
p
|
||||||
},
|
},
|
||||||
OutgoingKind::Normal(server) => {
|
OutgoingDestination::Normal(server) => {
|
||||||
let mut p = Vec::new();
|
let mut p = Vec::new();
|
||||||
p.extend_from_slice(server.as_bytes());
|
p.extend_from_slice(server.as_bytes());
|
||||||
p
|
p
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue