Compare commits

...

3 Commits

Author SHA1 Message Date
Win 7f7fd7c39c added stuff 2025-05-01 07:52:12 +07:00
Win 7804b95189 finished SEC-101 2025-05-01 04:47:28 +07:00
Win 90d7d69517 added SEC-101 work 2025-04-29 18:36:47 +07:00
12 changed files with 103 additions and 35 deletions

@ -0,0 +1 @@
Subproject commit 12b88508ce69530bdebfef69d641e222842e93b1

View File

@ -27,8 +27,11 @@ pub struct SendMessageCommand {
pub content: String,
}
pub struct GetHistory {
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct GetHistoryCommand {
// Get history from this room
#[serde(rename = "r")]
pub room: String,
}
/// User Command for quitting the whole chat session.

View File

@ -11,6 +11,19 @@ pub struct RoomDetail {
pub description: String,
}
// Chat history that contains the last x messages
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ChatHistoryEvent
{
// Room
#[serde(rename = "r")]
pub room: String,
// Last x messages
#[serde(rename = "m")]
pub messages: Vec<UserMessageBroadcastEvent>,
}
/// A user has successfully logged in
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct LoginSuccessfulReplyEvent {
@ -81,6 +94,7 @@ pub enum Event {
RoomParticipation(RoomParticipationBroadcastEvent),
UserJoinedRoom(UserJoinedRoomReplyEvent),
UserMessage(UserMessageBroadcastEvent),
ChatHistory(ChatHistoryEvent)
}
#[cfg(test)]
@ -166,4 +180,31 @@ mod tests {
r#"{"_et":"user_message","r":"test","u":"test","c":"test"}"#,
);
}
#[test]
fn test_chat_history_event()
{
let event = Event::ChatHistory(ChatHistoryEvent {
room: "test".to_string(),
messages: vec![
UserMessageBroadcastEvent
{
room: "test".to_string(),
user_id: "user1".to_string(),
content: "Whats up".to_string(),
},
UserMessageBroadcastEvent
{
room: "test".to_string(),
user_id: "user2".to_string(),
content: "Hi there".to_string(),
}
]
});
assert_event_serialization(
&event,
r#"{"_et":"chat_history","r":"test","m":[{"r":"test","u":"user1","c":"Whats up"},{"r":"test","u":"user2","c":"Hi there"}]}"#
);
}
}

View File

@ -42,8 +42,7 @@ impl RoomManagerBuilder {
}
self.chat_rooms.push((metadata.clone(), chat_room));
ChatRoom::listen_messages(&mut ChatRoom::new(metadata));
self
}

View File

@ -1,7 +1,8 @@
use std::collections::VecDeque;
use comms::event::{self, Event};
use serde::{Deserialize, Serialize};
use tokio::sync::broadcast;
use circular_queue::CircularQueue;
use super::{
user_registry::UserRegistry, user_session_handle::UserSessionHandle, SessionAndUserId,
@ -14,13 +15,8 @@ pub struct ChatRoomMetadata {
pub description: String,
}
#[derive(Debug, Clone)]
pub enum MessageBoxItem {
Message { user_id: String, content: String },
Notification(String),
}
const BROADCAST_CHANNEL_CAPACITY: usize = 100;
const MESSAGE_HISTORY_LIMIT: usize = 10; // Store last 10 messages
#[derive(Debug)]
/// [ChatRoom] handles the participants of a chat room and the primary broadcast channel
@ -29,7 +25,7 @@ pub struct ChatRoom {
metadata: ChatRoomMetadata,
broadcast_tx: broadcast::Sender<Event>,
user_registry: UserRegistry,
messages: CircularQueue<MessageBoxItem>
message_history: VecDeque<Event>
}
impl ChatRoom {
@ -40,7 +36,7 @@ impl ChatRoom {
metadata,
broadcast_tx,
user_registry: UserRegistry::new(),
messages: CircularQueue::with_capacity(10)
message_history: VecDeque::with_capacity(MESSAGE_HISTORY_LIMIT)
}
}
@ -48,23 +44,21 @@ impl ChatRoom {
self.user_registry.get_unique_user_ids()
}
pub fn store_message(&mut self, user_id: String, content: String)
// Remove old messages and push in new ones after the history limit has reached
// MESSAGE_HISTORY_LIMIT
pub fn store_message(&mut self, message: Event)
{
self.messages.push(MessageBoxItem::Message{user_id, content});
println!("{:#?}", self.messages);
if self.message_history.len() == MESSAGE_HISTORY_LIMIT
{
self.message_history.pop_front();
}
self.message_history.push_back(message);
}
pub async fn listen_messages(&mut self)
// Get Chat history
pub fn get_chat_history(&self) -> Vec<Event>
{
let mut broadcast_rx = self.broadcast_tx.subscribe();
while let Ok(event) = broadcast_rx.recv().await
{
if let Event::UserMessage(event) = event
{
self.store_message(event.user_id.clone(), event.content.clone());
}
}
self.message_history.iter().cloned().collect()
}
/// Add a participant to the room and broadcast that they joined
@ -85,6 +79,12 @@ impl ChatRoom {
session_and_user_id.clone(),
);
let history_events = self.get_chat_history();
for event in history_events
{
let _ = self.broadcast_tx.send(event) ;
}
// If the user is new e.g. they do not have another session with same user id,
// broadcast that they joined to all users
if self.user_registry.insert(&user_session_handle) {

View File

@ -1,6 +1,7 @@
use anyhow::Context;
use comms::event;
use tokio::sync::broadcast;
use crate::room_manager::ChatRoom;
#[derive(Debug, Clone)]
pub struct SessionAndUserId {
@ -48,15 +49,19 @@ impl UserSessionHandle {
}
/// Send a message to the room
pub fn send_message(&self, content: String) -> anyhow::Result<()> {
pub fn send_message(&self, chat_room: &mut ChatRoom, content: String) -> anyhow::Result<()> {
// create a distinct message event
let message_event = event::Event::UserMessage(event::UserMessageBroadcastEvent
{
room: self.room.clone(),
user_id: self.session_and_user_id.user_id.clone(),
content,
});
chat_room.store_message(message_event.clone());
self.broadcast_tx
.send(event::Event::UserMessage(
event::UserMessageBroadcastEvent {
room: self.room.clone(),
user_id: self.session_and_user_id.user_id.clone(),
content,
},
))
.send(message_event)
.context("could not write to the broadcast channel")?;
Ok(())

View File

@ -66,4 +66,8 @@ impl RoomManager {
Ok(())
}
pub async fn get_room(&self, room_name: &str) -> Option<tokio::sync::MutexGuard<'_, ChatRoom>> {
self.chat_rooms.get(room_name)?.lock().await.into()
}
}

View File

@ -79,7 +79,9 @@ impl ChatSession {
}
UserCommand::SendMessage(cmd) => {
if let Some((user_session_handle, _)) = self.joined_rooms.get(&cmd.room) {
let _ = user_session_handle.send_message(cmd.content);
if let Some(mut chat_room) = self.room_manager.get_room(&cmd.room).await {
let _ = user_session_handle.send_message(&mut chat_room, cmd.content);
}
}
}
UserCommand::LeaveRoom(cmd) => {

View File

@ -156,6 +156,19 @@ impl State {
}
}
}
event::Event::ChatHistory(event) => {
if let Some(room_data) = self.room_data_map.get_mut(&event.room)
{
for msg in &event.messages
{
room_data.messages.push(MessageBoxItem::Message
{
user_id: msg.user_id.clone(),
content: msg.content.clone()
});
}
}
}
}
}