arminius_lib_nats/client/
room.rs

1// SPDX-FileCopyrightText: 2025 BWI GmbH
2//
3// SPDX-License-Identifier: Apache-2.0
4
5//! Handles room related NATS operations
6
7use async_nats::{jetstream, jetstream::stream};
8
9use crate::{NatsClient, NatsError};
10
11/// The name of the stream used by the homeserver
12const ROOMSERVER_STREAM_NAME: &str = "roomserverstream";
13
14/// The amount of partitions in the roomserver stream
15pub const ROOMSERVER_STREAM_PARTITION_PARTS: usize = 100;
16
17impl NatsClient {
18	/// Get the stream for the roomserver
19	pub async fn get_roomserver_stream(&self) -> Result<stream::Stream, NatsError> {
20		let jetstream = jetstream::new(self.client.clone());
21		Ok(jetstream.get_or_create_stream(ROOMSERVER_STREAM_NAME).await?)
22	}
23
24	/// Sets up all required streams in NATS
25	pub async fn set_up_streams(&self) -> Result<(), NatsError> {
26		let jetstream = jetstream::new(self.client.clone());
27
28		// This stream is used by clients to make the roomserver execute
29		// commands
30		jetstream
31			.get_or_create_stream(stream::Config {
32				name: ROOMSERVER_STREAM_NAME.to_owned(),
33				subjects: vec!["rooms.>".to_owned()],
34				retention: stream::RetentionPolicy::WorkQueue,
35				subject_transform: Some(stream::SubjectTransform {
36					source: "rooms.*.>".to_owned(),
37					destination: format!(
38						"rooms.{{{{partition({}, 1)}}}}.{{{{wildcard(1)}}}}.>",
39						ROOMSERVER_STREAM_PARTITION_PARTS
40					),
41				}),
42				..Default::default()
43			})
44			.await?;
45
46		Ok(())
47	}
48}
49
50#[cfg(test)]
51mod tests;
arminius_lib_nats/client/room.rs

arminius_lib_nats/client/
room.rs
