#[cfg(feature = "e2e-encryption")]
use std::sync::{Arc, RwLock};
#[cfg(feature = "e2e-encryption")]
use futures_core::Stream;
#[cfg(feature = "e2e-encryption")]
use futures_util::StreamExt;
use ruma::{
events::{AnyMessageLikeEventContent, AnyStateEventContent},
serde::Raw,
};
use serde_json::value::{RawValue as RawJsonValue, Value as JsonValue};
#[cfg(feature = "e2e-encryption")]
use tokio::sync::broadcast;
#[cfg(feature = "e2e-encryption")]
use tokio_stream::wrappers::{errors::BroadcastStreamRecvError, BroadcastStream};
#[cfg(doc)]
use crate::Room;
#[cfg(feature = "e2e-encryption")]
#[derive(Clone, Debug)]
pub(crate) struct ChannelObservable<T: Clone + Send> {
value: Arc<RwLock<T>>,
channel: broadcast::Sender<T>,
}
#[cfg(feature = "e2e-encryption")]
impl<T: Default + Clone + Send + 'static> Default for ChannelObservable<T> {
fn default() -> Self {
let value = Default::default();
Self::new(value)
}
}
#[cfg(feature = "e2e-encryption")]
impl<T: 'static + Send + Clone> ChannelObservable<T> {
pub(crate) fn new(value: T) -> Self {
let channel = broadcast::Sender::new(100);
Self { value: RwLock::new(value).into(), channel }
}
pub(crate) fn subscribe(&self) -> impl Stream<Item = Result<T, BroadcastStreamRecvError>> {
let current_value = self.value.read().unwrap().to_owned();
let initial_stream = tokio_stream::once(Ok(current_value));
let broadcast_stream = BroadcastStream::new(self.channel.subscribe());
initial_stream.chain(broadcast_stream)
}
pub(crate) fn set(&self, new_value: T) -> T {
let old_value = {
let mut guard = self.value.write().unwrap();
std::mem::replace(&mut (*guard), new_value.clone())
};
let _ = self.channel.send(new_value);
old_value
}
pub(crate) fn get(&self) -> T {
self.value.read().unwrap().to_owned()
}
}
pub trait IntoRawMessageLikeEventContent {
#[doc(hidden)]
fn into_raw_message_like_event_content(self) -> Raw<AnyMessageLikeEventContent>;
}
impl IntoRawMessageLikeEventContent for Raw<AnyMessageLikeEventContent> {
fn into_raw_message_like_event_content(self) -> Raw<AnyMessageLikeEventContent> {
self
}
}
impl IntoRawMessageLikeEventContent for &Raw<AnyMessageLikeEventContent> {
fn into_raw_message_like_event_content(self) -> Raw<AnyMessageLikeEventContent> {
self.clone()
}
}
impl IntoRawMessageLikeEventContent for JsonValue {
fn into_raw_message_like_event_content(self) -> Raw<AnyMessageLikeEventContent> {
(&self).into_raw_message_like_event_content()
}
}
impl IntoRawMessageLikeEventContent for &JsonValue {
fn into_raw_message_like_event_content(self) -> Raw<AnyMessageLikeEventContent> {
Raw::new(self).expect("serde_json::Value never fails to serialize").cast()
}
}
impl IntoRawMessageLikeEventContent for Box<RawJsonValue> {
fn into_raw_message_like_event_content(self) -> Raw<AnyMessageLikeEventContent> {
Raw::from_json(self)
}
}
impl IntoRawMessageLikeEventContent for &RawJsonValue {
fn into_raw_message_like_event_content(self) -> Raw<AnyMessageLikeEventContent> {
self.to_owned().into_raw_message_like_event_content()
}
}
impl IntoRawMessageLikeEventContent for &Box<RawJsonValue> {
fn into_raw_message_like_event_content(self) -> Raw<AnyMessageLikeEventContent> {
self.clone().into_raw_message_like_event_content()
}
}
pub trait IntoRawStateEventContent {
#[doc(hidden)]
fn into_raw_state_event_content(self) -> Raw<AnyStateEventContent>;
}
impl IntoRawStateEventContent for Raw<AnyStateEventContent> {
fn into_raw_state_event_content(self) -> Raw<AnyStateEventContent> {
self
}
}
impl IntoRawStateEventContent for &Raw<AnyStateEventContent> {
fn into_raw_state_event_content(self) -> Raw<AnyStateEventContent> {
self.clone()
}
}
impl IntoRawStateEventContent for JsonValue {
fn into_raw_state_event_content(self) -> Raw<AnyStateEventContent> {
(&self).into_raw_state_event_content()
}
}
impl IntoRawStateEventContent for &JsonValue {
fn into_raw_state_event_content(self) -> Raw<AnyStateEventContent> {
Raw::new(self).expect("serde_json::Value never fails to serialize").cast()
}
}
impl IntoRawStateEventContent for Box<RawJsonValue> {
fn into_raw_state_event_content(self) -> Raw<AnyStateEventContent> {
Raw::from_json(self)
}
}
impl IntoRawStateEventContent for &RawJsonValue {
fn into_raw_state_event_content(self) -> Raw<AnyStateEventContent> {
self.to_owned().into_raw_state_event_content()
}
}
impl IntoRawStateEventContent for &Box<RawJsonValue> {
fn into_raw_state_event_content(self) -> Raw<AnyStateEventContent> {
self.clone().into_raw_state_event_content()
}
}
#[cfg(test)]
mod test {
#[cfg(feature = "e2e-encryption")]
#[test]
fn test_channel_observable_get_set() {
let observable = super::ChannelObservable::new(0);
assert_eq!(observable.get(), 0);
assert_eq!(observable.set(1), 0);
assert_eq!(observable.set(10), 1);
assert_eq!(observable.get(), 10);
}
}