1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
|
//! Abstraction for the messaging system to increase overall type safety and
//! code readability.
use std::sync::Arc;
use axum::extract::ws::{Message, WebSocket};
use futures::{stream::SplitSink, SinkExt};
use messenger_common::{client::MessageType as ClientMessageType, server::MessageType};
use tokio::sync::broadcast::Sender;
use tracing::error;
use crate::app::State;
/// Represents messages which can be sent through a [`WebSocket`].
pub trait Server
where
Self: std::marker::Sized + serde::Serialize,
{
/// Adds a message to the message history of the app state.
async fn append_to_history(&self, state: &Arc<State>) -> serde_json::Result<()>;
/// Performs serialization of a message.
/// Semantic alternative to `serde_json::to_string()`.
fn serialize(&self) -> serde_json::Result<String> {
serde_json::to_string(&self)
}
/// Sends a message through the provided sender.
///
/// The main purpose of this function is to enforce type-safety when sending
/// a message. This prevents accidentally sending non-`MessageType` messages
/// through the server.
fn send(&self, tx: &Sender<String>);
}
impl Server for MessageType {
/// Adds a message to the message history of the app state.
async fn append_to_history(&self, state: &Arc<State>) -> serde_json::Result<()> {
// Join and leave messages shouldn't be saved to history
// We ignore the inner message here as we want to serialize the entire message
if let Self::UserMessage(..) = *self {
let mut history_guard = state.message_history.lock().await;
// Only save the last 100 messages
if history_guard.len() > 99 {
history_guard.pop_front();
}
// Add a new message to the back of the queue
history_guard.push_back(serde_json::to_string(self)?);
}
Ok(())
}
fn serialize(&self) -> serde_json::Result<String> {
serde_json::to_string(&self)
}
fn send(&self, tx: &Sender<String>) {
tracing::debug!("sending message, content: {:?}", &self);
match self.serialize() {
Ok(json_message) => {
if let Err(error) = tx.send(json_message) {
error!("error occurred while sending a message through a channel: {error}");
}
}
Err(error) => {
error!("error occurred while converting message to json: {error}");
}
};
}
}
/// Performs deserialization of a message.
/// Semantic alternative to `serde_json::from_str::<MessageType>()`.
pub fn deserialize(message: &str) -> serde_json::Result<ClientMessageType> {
serde_json::from_str(message)
}
/// Sends an error through a websocket to the client.
/// Contains error handling to reduce overall code bloat.
pub async fn send_error(
sender: &mut SplitSink<WebSocket, Message>,
error: messenger_common::server::Error,
) {
// Log error through tracing to show invalid client behaviour
error!("received message from client that is considered an error: {error}");
// Handle deserialization errors correctly to avoid a panic
match MessageType::Error(error).serialize() {
Ok(outbound_error) => {
if let Err(error) = sender.send(Message::Text(outbound_error)).await {
error!("unable to send error message through a websocket: {error}");
}
}
Err(error) => {
// Errors can also occur during serialization, so these should be covered
error!("unable to serialize outbound error message: {error}");
}
};
}
|