Struct Service 

pub struct Service {
    pub db: Data,
    server: Arc<Server>,
    services: Arc<OnceServices>,
    channels: Vec<(Sender<Msg>, Receiver<Msg>)>,
}

Fields

db: Data

server: Arc<Server>

services: Arc<OnceServices>

channels: Vec<(Sender<Msg>, Receiver<Msg>)>

Implementations

impl Service

pub(super) async fn sender(self: Arc<Self>, id: usize) -> Result

async fn work_loop<'a>( &'a self, id: usize, futures: &mut FuturesUnordered<BoxFuture<'a, Result<Destination, (Destination, Error)>>>, statuses: &mut HashMap<Destination, TransactionStatus>, )

async fn handle_response<'a>( &'a self, response: Result<Destination, (Destination, Error)>, futures: &mut FuturesUnordered<BoxFuture<'a, Result<Destination, (Destination, Error)>>>, statuses: &mut HashMap<Destination, TransactionStatus>, )

fn handle_response_err( dest: Destination, statuses: &mut HashMap<Destination, TransactionStatus>, e: &Error, )

async fn handle_response_ok<'a>( &'a self, dest: &Destination, futures: &mut FuturesUnordered<BoxFuture<'a, Result<Destination, (Destination, Error)>>>, statuses: &mut HashMap<Destination, TransactionStatus>, )

async fn handle_request<'a>( &'a self, msg: Msg, futures: &mut FuturesUnordered<BoxFuture<'a, Result<Destination, (Destination, Error)>>>, statuses: &mut HashMap<Destination, TransactionStatus>, )

async fn finish_responses<'a>( &'a self, futures: &mut FuturesUnordered<BoxFuture<'a, Result<Destination, (Destination, Error)>>>, )

async fn startup_netburst<'a>( &'a self, id: usize, futures: &mut FuturesUnordered<BoxFuture<'a, Result<Destination, (Destination, Error)>>>, statuses: &mut HashMap<Destination, TransactionStatus>, )

async fn select_events( &self, dest: &Destination, new_events: Vec<(Vec<u8>, SendingEvent)>, statuses: &mut HashMap<Destination, TransactionStatus>, ) -> Result<Option<Vec<SendingEvent>>>

fn select_events_current( &self, dest: &Destination, statuses: &mut HashMap<Destination, TransactionStatus>, ) -> Result<(bool, bool)>

async fn select_edus(&self, server_name: &ServerName) -> Result<(EduVec, u64)>

async fn select_edus_device_changes( &self, server_name: &ServerName, since: (u64, u64), max_edu_count: &AtomicU64, events_len: &AtomicUsize, ) -> EduVec

Look for device changes

async fn select_edus_receipts( &self, server_name: &ServerName, since: (u64, u64), max_edu_count: &AtomicU64, ) -> EduVec

Look for read receipts in this room

MSC3771 lets a user emit multiple receipts in the same EDU window, one per thread context. The federation EDU shape allows only one ReceiptData per (room, user) slot, so a user with N parallel thread receipts ships across N parallel Edu::Receipt buffers within the same transaction. Each buffer is shape-compliant; receivers process them as independent receipt EDUs and our storage keeps each thread distinct.

async fn select_edus_receipts_room( &self, room_id: &RoomId, since: (u64, u64), max_edu_count: &AtomicU64, num: &AtomicUsize, ) -> SmallVec<[ReceiptMap; 1]>

Look for read receipts in this room.

Returns a per-rank vector of [ReceiptMap]s. Each user’s receipts in the window (one per thread context, count-ordered) are placed into successive ranks, so rank 0 carries each user’s earliest receipt, rank 1 the next, and so on. The receipt-limit budget bounds distinct users only; subsequent thread receipts for an already-counted user do not consume additional budget.

async fn select_edus_presence( &self, server_name: &ServerName, since: (u64, u64), max_edu_count: &AtomicU64, ) -> Option<EduBuf>

Look for presence

fn send_events( &self, dest: Destination, events: Vec<SendingEvent>, ) -> BoxFuture<'_, Result<Destination, (Destination, Error)>>

async fn send_events_dest_appservice( &self, id: String, events: Vec<SendingEvent>, ) -> Result<Destination, (Destination, Error)>

async fn send_events_dest_push( &self, user_id: OwnedUserId, pushkey: String, events: Vec<SendingEvent>, ) -> Result<Destination, (Destination, Error)>

pub fn schedule_flush_suppressed_for_pushkey( &self, user_id: OwnedUserId, pushkey: String, reason: &'static str, )

pub fn schedule_flush_suppressed_for_user( &self, user_id: OwnedUserId, reason: &'static str, )

async fn enqueue_suppressed_push_events( &self, user_id: &UserId, pushkey: &str, events: &[SendingEvent], ) -> usize

async fn flush_suppressed_rooms( &self, user_id: &UserId, pushkey: &str, pusher: &Pusher, rules_for_user: &Ruleset, rooms: Vec<(OwnedRoomId, Vec<RawPduId>)>, reason: &'static str, )

async fn flush_suppressed_for_pushkey( &self, user_id: OwnedUserId, pushkey: String, reason: &'static str, )

pub async fn flush_suppressed_for_user( &self, user_id: OwnedUserId, reason: &'static str, )

async fn pushing_suppressed(&self, user_id: &UserId) -> bool

async fn send_events_dest_federation( &self, server: OwnedServerName, events: Vec<SendingEvent>, ) -> Result<Destination, (Destination, Error)>

impl Service

pub fn send_pdu_push( &self, pdu_id: &RawPduId, user: &UserId, pushkey: String, ) -> Result

pub fn send_pdu_appservice( &self, appservice_id: String, pdu_id: RawPduId, ) -> Result

pub async fn send_pdu_room(&self, room_id: &RoomId, pdu_id: &RawPduId) -> Result

pub async fn send_pdu_servers<'a, S>( &self, servers: S, pdu_id: &RawPduId, ) -> Result

where S: Stream<Item = &'a ServerName> + Send + 'a,

pub fn send_edu_server(&self, server: &ServerName, serialized: EduBuf) -> Result

pub async fn send_edu_room( &self, room_id: &RoomId, serialized: EduBuf, ) -> Result

pub fn send_edu_appservice( &self, appservice_id: String, serialized: EduBuf, ) -> Result

Queue an EDU for delivery to a specific appservice.

pub async fn send_edu_room_appservices<'a, F>( &self, room_id: &RoomId, serializer: F, ) -> Result

where F: Fn(&mut dyn Write) -> Result + Send + 'a, &'a F: Send + Sync,

Sends an EDU to all appservices interested in a room. The serialized data must be in EphemeralData format, not federation Edu.

pub async fn send_edu_servers<'a, S>( &self, servers: S, serialized: EduBuf, ) -> Result

where S: Stream<Item = &'a ServerName> + Send + 'a,

pub async fn flush_room(&self, room_id: &RoomId) -> Result

pub async fn flush_servers<'a, S>(&self, servers: S) -> Result

where S: Stream<Item = &'a ServerName> + Send + 'a,

pub async fn cleanup_events( &self, appservice_id: Option<&str>, user_id: Option<&UserId>, push_key: Option<&str>, ) -> Result

Clean up queued sending event data

Used after we remove an appservice registration or a user deletes a push key

fn dispatch(&self, msg: Msg) -> Result

pub(crate) fn shard_id(&self, dest: &Destination) -> usize

Trait Implementations

impl Service for Service

fn build(args: &Args<'_>) -> Result<Arc<Self>>

Implement the construction of the service instance. Services are generally singletons so expect this to only be called once for a service type. Note that it may be called again after a server reload, but the prior instance will have been dropped first. Failure will shutdown the server with an error.

fn worker<'async_trait>( self: Arc<Self>, ) -> Pin<Box<dyn Future<Output = Result> + Send + 'async_trait>>

where Self: 'async_trait,

Implement the service’s worker loop. The service manager spawns a task and calls this function after all services have been built.

fn interrupt<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Interrupt the service. This is sent to initiate a graceful shutdown. The service worker should return from its work loop.

fn name(&self) -> &str

Return the name of the service. i.e. crate::service::make_name(std::module_path!())

fn unconstrained(&self) -> bool

Return true if the service worker opts out of the tokio cooperative budgeting. This can reduce tail latency at the risk of event loop starvation.

fn clear_cache<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Clear any caches or similar runtime state.

fn memory_usage<'life0, 'life1, 'async_trait>( &'life0 self, _out: &'life1 mut (dyn Write + Send), ) -> Pin<Box<dyn Future<Output = Result> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Memory usage report in a markdown string.