Compare commits
No commits in common. "7f7fd7c39c39891276a7dce495ac8c6b58926c03" and "32d98bc5d1e1ddb83449411b58fd597e2a9bd98e" have entirely different histories.
7f7fd7c39c
...
32d98bc5d1
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -1 +0,0 @@
|
||||||
Subproject commit 12b88508ce69530bdebfef69d641e222842e93b1
|
|
||||||
|
|
@ -27,11 +27,8 @@ pub struct SendMessageCommand {
|
||||||
pub content: String,
|
pub content: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
pub struct GetHistory {
|
||||||
pub struct GetHistoryCommand {
|
|
||||||
// Get history from this room
|
|
||||||
#[serde(rename = "r")]
|
|
||||||
pub room: String,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// User Command for quitting the whole chat session.
|
/// User Command for quitting the whole chat session.
|
||||||
|
|
|
||||||
|
|
@ -11,19 +11,6 @@ pub struct RoomDetail {
|
||||||
pub description: String,
|
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
|
/// A user has successfully logged in
|
||||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||||
pub struct LoginSuccessfulReplyEvent {
|
pub struct LoginSuccessfulReplyEvent {
|
||||||
|
|
@ -94,7 +81,6 @@ pub enum Event {
|
||||||
RoomParticipation(RoomParticipationBroadcastEvent),
|
RoomParticipation(RoomParticipationBroadcastEvent),
|
||||||
UserJoinedRoom(UserJoinedRoomReplyEvent),
|
UserJoinedRoom(UserJoinedRoomReplyEvent),
|
||||||
UserMessage(UserMessageBroadcastEvent),
|
UserMessage(UserMessageBroadcastEvent),
|
||||||
ChatHistory(ChatHistoryEvent)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
|
|
@ -180,31 +166,4 @@ mod tests {
|
||||||
r#"{"_et":"user_message","r":"test","u":"test","c":"test"}"#,
|
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"}]}"#
|
|
||||||
);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -43,6 +43,7 @@ impl RoomManagerBuilder {
|
||||||
|
|
||||||
self.chat_rooms.push((metadata.clone(), chat_room));
|
self.chat_rooms.push((metadata.clone(), chat_room));
|
||||||
|
|
||||||
|
ChatRoom::listen_messages(&mut ChatRoom::new(metadata));
|
||||||
|
|
||||||
self
|
self
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -1,8 +1,7 @@
|
||||||
use std::collections::VecDeque;
|
|
||||||
|
|
||||||
use comms::event::{self, Event};
|
use comms::event::{self, Event};
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
use tokio::sync::broadcast;
|
use tokio::sync::broadcast;
|
||||||
|
use circular_queue::CircularQueue;
|
||||||
|
|
||||||
use super::{
|
use super::{
|
||||||
user_registry::UserRegistry, user_session_handle::UserSessionHandle, SessionAndUserId,
|
user_registry::UserRegistry, user_session_handle::UserSessionHandle, SessionAndUserId,
|
||||||
|
|
@ -15,8 +14,13 @@ pub struct ChatRoomMetadata {
|
||||||
pub description: String,
|
pub description: String,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone)]
|
||||||
|
pub enum MessageBoxItem {
|
||||||
|
Message { user_id: String, content: String },
|
||||||
|
Notification(String),
|
||||||
|
}
|
||||||
|
|
||||||
const BROADCAST_CHANNEL_CAPACITY: usize = 100;
|
const BROADCAST_CHANNEL_CAPACITY: usize = 100;
|
||||||
const MESSAGE_HISTORY_LIMIT: usize = 10; // Store last 10 messages
|
|
||||||
|
|
||||||
#[derive(Debug)]
|
#[derive(Debug)]
|
||||||
/// [ChatRoom] handles the participants of a chat room and the primary broadcast channel
|
/// [ChatRoom] handles the participants of a chat room and the primary broadcast channel
|
||||||
|
|
@ -25,7 +29,7 @@ pub struct ChatRoom {
|
||||||
metadata: ChatRoomMetadata,
|
metadata: ChatRoomMetadata,
|
||||||
broadcast_tx: broadcast::Sender<Event>,
|
broadcast_tx: broadcast::Sender<Event>,
|
||||||
user_registry: UserRegistry,
|
user_registry: UserRegistry,
|
||||||
message_history: VecDeque<Event>
|
messages: CircularQueue<MessageBoxItem>
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ChatRoom {
|
impl ChatRoom {
|
||||||
|
|
@ -36,7 +40,7 @@ impl ChatRoom {
|
||||||
metadata,
|
metadata,
|
||||||
broadcast_tx,
|
broadcast_tx,
|
||||||
user_registry: UserRegistry::new(),
|
user_registry: UserRegistry::new(),
|
||||||
message_history: VecDeque::with_capacity(MESSAGE_HISTORY_LIMIT)
|
messages: CircularQueue::with_capacity(10)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -44,21 +48,23 @@ impl ChatRoom {
|
||||||
self.user_registry.get_unique_user_ids()
|
self.user_registry.get_unique_user_ids()
|
||||||
}
|
}
|
||||||
|
|
||||||
// Remove old messages and push in new ones after the history limit has reached
|
pub fn store_message(&mut self, user_id: String, content: String)
|
||||||
// MESSAGE_HISTORY_LIMIT
|
|
||||||
pub fn store_message(&mut self, message: Event)
|
|
||||||
{
|
{
|
||||||
if self.message_history.len() == MESSAGE_HISTORY_LIMIT
|
self.messages.push(MessageBoxItem::Message{user_id, content});
|
||||||
{
|
println!("{:#?}", self.messages);
|
||||||
self.message_history.pop_front();
|
|
||||||
}
|
|
||||||
self.message_history.push_back(message);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Get Chat history
|
pub async fn listen_messages(&mut self)
|
||||||
pub fn get_chat_history(&self) -> Vec<Event>
|
|
||||||
{
|
{
|
||||||
self.message_history.iter().cloned().collect()
|
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());
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Add a participant to the room and broadcast that they joined
|
/// Add a participant to the room and broadcast that they joined
|
||||||
|
|
@ -79,12 +85,6 @@ impl ChatRoom {
|
||||||
session_and_user_id.clone(),
|
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,
|
// If the user is new e.g. they do not have another session with same user id,
|
||||||
// broadcast that they joined to all users
|
// broadcast that they joined to all users
|
||||||
if self.user_registry.insert(&user_session_handle) {
|
if self.user_registry.insert(&user_session_handle) {
|
||||||
|
|
|
||||||
|
|
@ -1,7 +1,6 @@
|
||||||
use anyhow::Context;
|
use anyhow::Context;
|
||||||
use comms::event;
|
use comms::event;
|
||||||
use tokio::sync::broadcast;
|
use tokio::sync::broadcast;
|
||||||
use crate::room_manager::ChatRoom;
|
|
||||||
|
|
||||||
#[derive(Debug, Clone)]
|
#[derive(Debug, Clone)]
|
||||||
pub struct SessionAndUserId {
|
pub struct SessionAndUserId {
|
||||||
|
|
@ -49,19 +48,15 @@ impl UserSessionHandle {
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Send a message to the room
|
/// Send a message to the room
|
||||||
pub fn send_message(&self, chat_room: &mut ChatRoom, content: String) -> anyhow::Result<()> {
|
pub fn send_message(&self, content: String) -> anyhow::Result<()> {
|
||||||
// create a distinct message event
|
self.broadcast_tx
|
||||||
let message_event = event::Event::UserMessage(event::UserMessageBroadcastEvent
|
.send(event::Event::UserMessage(
|
||||||
{
|
event::UserMessageBroadcastEvent {
|
||||||
room: self.room.clone(),
|
room: self.room.clone(),
|
||||||
user_id: self.session_and_user_id.user_id.clone(),
|
user_id: self.session_and_user_id.user_id.clone(),
|
||||||
content,
|
content,
|
||||||
});
|
},
|
||||||
|
))
|
||||||
chat_room.store_message(message_event.clone());
|
|
||||||
|
|
||||||
self.broadcast_tx
|
|
||||||
.send(message_event)
|
|
||||||
.context("could not write to the broadcast channel")?;
|
.context("could not write to the broadcast channel")?;
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
|
|
|
||||||
|
|
@ -66,8 +66,4 @@ impl RoomManager {
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn get_room(&self, room_name: &str) -> Option<tokio::sync::MutexGuard<'_, ChatRoom>> {
|
|
||||||
self.chat_rooms.get(room_name)?.lock().await.into()
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -79,9 +79,7 @@ impl ChatSession {
|
||||||
}
|
}
|
||||||
UserCommand::SendMessage(cmd) => {
|
UserCommand::SendMessage(cmd) => {
|
||||||
if let Some((user_session_handle, _)) = self.joined_rooms.get(&cmd.room) {
|
if let Some((user_session_handle, _)) = self.joined_rooms.get(&cmd.room) {
|
||||||
if let Some(mut chat_room) = self.room_manager.get_room(&cmd.room).await {
|
let _ = user_session_handle.send_message(cmd.content);
|
||||||
let _ = user_session_handle.send_message(&mut chat_room, cmd.content);
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
UserCommand::LeaveRoom(cmd) => {
|
UserCommand::LeaveRoom(cmd) => {
|
||||||
|
|
|
||||||
|
|
@ -156,19 +156,6 @@ 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()
|
|
||||||
});
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue