diff --git a/Lab3/.vscode/launch.json b/Lab3/.vscode/launch.json new file mode 100644 index 0000000..54b90be --- /dev/null +++ b/Lab3/.vscode/launch.json @@ -0,0 +1,38 @@ +{ + "version": "0.2.0", + "configurations": [ + { + "name": "ChatRoom", + "type": "coreclr", + "request": "launch", + "preLaunchTask": "build-ChatRoom", + "program": "${workspaceFolder}/ChatRoom/bin/Debug/net6.0/ChatRoom.dll", + "args": [], + "cwd": "${workspaceFolder}/ChatRoom", + "console": "externalTerminal", + "stopAtEntry": false + }, + { + "name": "Moderator", + "type": "coreclr", + "request": "launch", + "preLaunchTask": "build-Moderator", + "program": "${workspaceFolder}/Moderator/bin/Debug/net6.0/Moderator.dll", + "args": [], + "cwd": "${workspaceFolder}/Moderator", + "console": "externalTerminal", + "stopAtEntry": false + }, + { + "name": "Participant", + "type": "coreclr", + "request": "launch", + "preLaunchTask": "build-Participant", + "program": "${workspaceFolder}/Participant/bin/Debug/net6.0/Participant.dll", + "args": [], + "cwd": "${workspaceFolder}/Participant", + "console": "externalTerminal", + "stopAtEntry": false + } + ] +} \ No newline at end of file diff --git a/Lab3/.vscode/tasks.json b/Lab3/.vscode/tasks.json new file mode 100644 index 0000000..cf312c1 --- /dev/null +++ b/Lab3/.vscode/tasks.json @@ -0,0 +1,41 @@ +{ + "version": "2.0.0", + "tasks": [ + { + "label": "build-ChatRoom", + "command": "dotnet", + "type": "process", + "args": [ + "build", + "${workspaceFolder}/ChatRoom/ChatRoom.csproj", + "/property:GenerateFullPaths=true", + "/consoleloggerparameters:NoSummary" + ], + "problemMatcher": "$msCompile" + }, + { + "label": "build-Moderator", + "command": "dotnet", + "type": "process", + "args": [ + "build", + "${workspaceFolder}/Moderator/Moderator.csproj", + "/property:GenerateFullPaths=true", + "/consoleloggerparameters:NoSummary" + ], + "problemMatcher": "$msCompile" + }, + { + "label": "build-Participant", + "command": "dotnet", + "type": "process", + "args": [ + "build", + "${workspaceFolder}/Participant/Participant.csproj", + "/property:GenerateFullPaths=true", + "/consoleloggerparameters:NoSummary" + ], + "problemMatcher": "$msCompile" + } + ] +} \ No newline at end of file diff --git a/Lab3/ChatRoom/ChatRoom.csproj b/Lab3/ChatRoom/ChatRoom.csproj new file mode 100644 index 0000000..30cd58d --- /dev/null +++ b/Lab3/ChatRoom/ChatRoom.csproj @@ -0,0 +1,20 @@ + + + + Exe + net6.0 + enable + enable + + + + + + + + + + + + + diff --git a/Lab3/ChatRoom/ChatRoomLogic.cs b/Lab3/ChatRoom/ChatRoomLogic.cs new file mode 100644 index 0000000..7d58b0f --- /dev/null +++ b/Lab3/ChatRoom/ChatRoomLogic.cs @@ -0,0 +1,296 @@ +using NLog; + +namespace ChatRoom; + +/// +/// Chat room service logic +/// +public class ChatRoomLogic +{ + /// + /// Background thread of deleting approved or rejected messages + /// + private Thread thread; + /// + /// Chat Room state + /// + private ChatRoomState state = new ChatRoomState(); + /// + /// Logger + /// + private Logger log = LogManager.GetCurrentClassLogger(); + + /// + /// Chat room logic constructor + /// + public ChatRoomLogic() + { + thread = new Thread(BackgroundTask); + thread.Start(); + } + + /// + /// Generate the next incrementing ID + /// + /// Unique ID + int NextId() + { + int id = state.lastUniqueId; + state.lastUniqueId++; + return id; + } + + /// + /// Register a client + /// + /// Client name + /// Client ID + public int RegisterClient(string name) + { + lock (state.accessLock) + { + int clientId = NextId(); + state.clients.Add(new Client + { + id = clientId, + name = name + }); + log.Info($"Registered with client '{name}' with id {clientId}"); + return clientId; + } + } + + /// + /// Find a client by ID, can return NULL if not found + /// + /// Client ID + /// Optional client object + Client? FindClientById(int clientId) + { + foreach (var client in state.clients) + { + if (client.id == clientId) + { + return client; + } + } + return null; + } + + /// + /// Find message by ID, can return NULL if not found + /// + /// Message ID + /// Optional message object + Message? FindMessageById(int messageId) + { + foreach (var message in state.messages) + { + if (message.id == messageId) + { + return message; + } + } + return null; + } + + /// + /// Check if client is still blocked, will clear `blockedUntil` if client became unblocked + /// + /// Client object + /// Is client blocked? + bool GetAndUpdateBlockedState(Client client) + { + if (client.blockedUntil != null && DateTime.UtcNow >= client.blockedUntil) + { + client.blockedUntil = null; + } + + return client.blockedUntil != null; + } + + /// + /// Send a message + /// + /// Client ID + /// Message contents + /// Does this message need to be censored? + /// Was sending the message successful, can fail if client is blocked + public bool SendMessage(int clientId, string contents, bool needsToBeCensored) + { + lock (state.accessLock) + { + var client = FindClientById(clientId); + if (client == null) + { + return false; + } + + if (GetAndUpdateBlockedState(client)) + { + return false; + } + + var message = new Message + { + id = NextId(), + clientId = clientId, + contents = contents, + needsToBeCensored = needsToBeCensored, + status = MessageStatus.WaitingForModerator, + sentAt = DateTime.UtcNow + }; + state.messages.Add(message); + log.Info($"Client '{client.name}' ({client.id}) sent message '{contents}' ({message.id}). Needs to censored: {needsToBeCensored}"); + } + + return true; + } + + /// + /// Get next message which isin't approved or rejected + /// + /// Optional message object + public ChatRoomContract.Message? GetNewMessage() + { + lock (state.accessLock) + { + foreach (var message in state.messages) + { + if (message.status != MessageStatus.WaitingForModerator) continue; + + log.Info($"Message '{message.id}' given to moderator"); + message.status = MessageStatus.GivenToModerator; + return new ChatRoomContract.Message + { + id = message.id, + contents = message.contents, + needsToBeCensored = message.needsToBeCensored + }; + } + } + + return null; + } + + /// + /// Approve a message + /// + /// Message ID + public void ApproveMessage(int messageId) + { + lock (state.accessLock) + { + var message = FindMessageById(messageId); + if (message == null) + { + return; + } + + if (message.status != MessageStatus.GivenToModerator) + { + return; + } + + message.status = MessageStatus.Approved; + log.Info($"Message {message.id} was approved"); + } + } + + /// + /// Reject a message + /// + /// Message ID + public void RejectMessage(int messageId) + { + lock (state.accessLock) + { + var message = FindMessageById(messageId); + if (message == null) + { + return; + } + + if (message.status != MessageStatus.GivenToModerator) + { + return; + } + + message.status = MessageStatus.Rejected; + log.Info($"Message {message.id} was rejected"); + + var client = FindClientById(message.clientId); + if (client != null && !GetAndUpdateBlockedState(client)) + { + client.strikes++; + + var rnd = new Random(); + if (client.strikes > rnd.Next(0, 10)) + { + log.Info($"Client '{client.name}' ({client.id}) was blocked for {client.strikes}s"); + client.blockedUntil = DateTime.UtcNow.AddSeconds(client.strikes); + client.strikes = 0; + } + } + } + } + + /// + /// Get number of strikes a client has + /// + /// Client ID + /// Number of strikes + public int GetStrikes(int clientId) + { + lock (state.accessLock) + { + var client = FindClientById(clientId); + if (client == null) + { + return 0; + } + + return client.strikes; + } + } + + /// + /// Get timestamp until when the client is blocked + /// + /// Client ID + /// Optional datetime object + public DateTime? GetBlockedUntil(int clientId) + { + lock (state.accessLock) + { + var client = FindClientById(clientId); + if (client == null) + { + return null; + } + + return client.blockedUntil; + } + } + + /// + /// Main loop for background thread. Used for deleting approved or rejected messages. + /// + public void BackgroundTask() + { + while (true) + { + lock (state.accessLock) + { + var now = DateTime.UtcNow; + int count = state.messages.RemoveAll(msg => + (now - msg.sentAt).TotalSeconds > 5 && + (msg.status == MessageStatus.Approved || msg.status == MessageStatus.Rejected) + ); + log.Info($"Running periodic cleanup, removed {count} messages"); + } + + Thread.Sleep(10 * 1000); + } + } +} diff --git a/Lab3/ChatRoom/ChatRoomService.cs b/Lab3/ChatRoom/ChatRoomService.cs new file mode 100644 index 0000000..a9b5af2 --- /dev/null +++ b/Lab3/ChatRoom/ChatRoomService.cs @@ -0,0 +1,156 @@ +using RabbitMQ.Client; +using Newtonsoft.Json; +using RabbitMQ.Client.Events; +using ChatRoomContract; +using System.Text; +using MessagePack; +using System.Diagnostics; +using NLog.LayoutRenderers.Wrappers; +using NLog; + +namespace ChatRoom; + +internal class ChatRoomService +{ + /// + /// Logger for this class. + /// + private Logger log = LogManager.GetCurrentClassLogger(); + + /// + /// Communications channel to RabbitMQ message broker. + /// + private IModel rmqChannel; + + /// + /// Service logic. + /// + private ChatRoomLogic logic = new ChatRoomLogic(); + + public ChatRoomService(IConnection connection, string exchangeName, string serverQueueName) + { + //get channel, configure exchanges and request queue + rmqChannel = connection.CreateModel(); + + rmqChannel.ExchangeDeclare(exchange: exchangeName, type: ExchangeType.Direct); + rmqChannel.QueueDeclare(queue: serverQueueName, durable: true, exclusive: false, autoDelete: false, arguments: null); + rmqChannel.QueueBind(queue: serverQueueName, exchange: exchangeName, routingKey: serverQueueName, arguments: null); + + //connect to the queue as consumer + //XXX: see https://www.rabbitmq.com/dotnet-api-guide.html#concurrency for threading issues + var rmqConsumer = new EventingBasicConsumer(rmqChannel); + rmqConsumer.Received += (consumer, delivery) => OnMessageReceived(((EventingBasicConsumer)consumer).Model, delivery); + rmqChannel.BasicConsume(queue: serverQueueName, autoAck: true, consumer: rmqConsumer); + } + + /// + /// Is invoked to process messages received. + /// + /// Related communications channel. + /// Message deliver data. + private void OnMessageReceived(IModel channel, BasicDeliverEventArgs msgIn) + { + try + { + var msg = MessagePackSerializer.Deserialize(msgIn.Body); + Debug.Assert(msg != null); + + if (msg.isResponse == true) + { + return; + } + + bool hasResponse = false; + byte[]? response = null; + + switch (msg.method) + { + case nameof(IChatRoomService.RegisterClient): + { + var name = MessagePackSerializer.Deserialize(msg.args); + var clientId = logic.RegisterClient(name); + response = MessagePackSerializer.Serialize(clientId); + hasResponse = true; + break; + } + + case nameof(IChatRoomService.GetStrikes): + { + var clientId = MessagePackSerializer.Deserialize(msg.args); + var strikes = logic.GetStrikes(clientId); + response = MessagePackSerializer.Serialize(strikes); + hasResponse = true; + break; + } + + case nameof(IChatRoomService.GetBlockedUntil): + { + var clientId = MessagePackSerializer.Deserialize(msg.args); + var blockedUntil = logic.GetBlockedUntil(clientId); + response = MessagePackSerializer.Serialize(blockedUntil); + hasResponse = true; + break; + } + + case nameof(IChatRoomService.SendMessage): + { + var args = MessagePackSerializer.Deserialize(msg.args); + var success = logic.SendMessage(args.clientId, args.contents, args.needsToBeCensored); + response = MessagePackSerializer.Serialize(success); + hasResponse = true; + break; + } + + case nameof(IChatRoomService.GetNewMessage): + { + var newMessage = logic.GetNewMessage(); + response = MessagePackSerializer.Serialize(newMessage); + hasResponse = true; + break; + } + + case nameof(IChatRoomService.RejectMessage): + { + var messageId = MessagePackSerializer.Deserialize(msg.args); + logic.ApproveMessage(messageId); + break; + } + + case nameof(IChatRoomService.ApproveMessage): + { + var messageId = MessagePackSerializer.Deserialize(msg.args); + logic.ApproveMessage(messageId); + break; + } + + default: + { + throw new Exception("Unknown RPC method"); + } + } + + if (hasResponse) + { + var responseMsg = new RPCMessage + { + isResponse = true, + method = msg.method, + args = response + }; + + var properties = channel.CreateBasicProperties(); + properties.CorrelationId = msgIn.BasicProperties.CorrelationId; + + channel.BasicPublish( + exchange: msgIn.Exchange, + routingKey: msgIn.BasicProperties.ReplyTo, + basicProperties: properties, + body: MessagePackSerializer.Serialize(responseMsg) + ); + } + } catch (Exception e) + { + log.Error(e, "Unhandled exception caught when processing a message. The message is now lost."); + } + } +} diff --git a/Lab3/ChatRoom/ChatRoomState.cs b/Lab3/ChatRoom/ChatRoomState.cs new file mode 100644 index 0000000..f6616ea --- /dev/null +++ b/Lab3/ChatRoom/ChatRoomState.cs @@ -0,0 +1,97 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace ChatRoom; + +/// +/// Client information +/// +public class Client +{ + /// + /// Client ID + /// + public int id; + /// + /// Client name, can be the same between multiple clients + /// + public string name; + /// + /// Number of strikes + /// + public int strikes = 0; + /// + /// Until when is this client blocked from sending messages + /// + public DateTime? blockedUntil = null; +} + +/// +/// Describes the messages status/stage +/// +public enum MessageStatus +{ + WaitingForModerator, + GivenToModerator, + Approved, + Rejected +} + +/// +/// Message information +/// +public class Message +{ + /// + /// Message ID + /// + public int id; + /// + /// Client ID + /// + public int clientId; + /// + /// Message contents + /// + public string contents; + /// + /// Does this message need to be censored? + /// + public bool needsToBeCensored; + /// + /// Message status/stage + /// + public MessageStatus status = MessageStatus.WaitingForModerator; + + /// + /// When was this message sent + /// + public DateTime sentAt; +} + +public class ChatRoomState +{ + /// + /// Access lock. + /// + public readonly object accessLock = new object(); + + /// + /// Last unique ID value generated. + /// + public int lastUniqueId; + + /// + /// List of all registered clients + /// + public List clients = new List(); + + /// + /// List of messages + /// + public List messages = new List(); +} + diff --git a/Lab3/ChatRoom/Server.cs b/Lab3/ChatRoom/Server.cs new file mode 100644 index 0000000..779e68b --- /dev/null +++ b/Lab3/ChatRoom/Server.cs @@ -0,0 +1,73 @@ +using ChatRoomContract; +using NLog; + +namespace ChatRoom; + +internal class Server +{ + /// + /// Logger for this class. + /// + private Logger log = LogManager.GetCurrentClassLogger(); + + /// + /// Configure loggin subsystem. + /// + private void ConfigureLogging() + { + var config = new NLog.Config.LoggingConfiguration(); + + var console = + new NLog.Targets.ConsoleTarget("console") + { + Layout = @"${date:format=HH\:mm\:ss}|${level}| ${message} ${exception}" + }; + config.AddTarget(console); + config.AddRuleForAllLevels(console); + + LogManager.Configuration = config; + } + + /// + /// Program body. + /// + private void Run() + { + //configure logging + ConfigureLogging(); + + while (true) + { + try + { + var service = new ChatRoomService(Config.CreateConnection(), Config.ExchangeName, Config.ServerQueueName); + + log.Info("Server has been started."); + + //hang main thread + while (true) + { + Thread.Sleep(1000); + } + } + catch (Exception e) + { + //log exception + log.Error(e, "Unhandled exception caught. Server will now restart."); + + //prevent console spamming + Thread.Sleep(2000); + } + } + } + + /// + /// Program entry point. + /// + /// Command line arguments. + static void Main(string[] args) + { + var self = new Server(); + self.Run(); + } +} diff --git a/Lab3/ChatRoomContract/ChatRoomClient.cs b/Lab3/ChatRoomContract/ChatRoomClient.cs new file mode 100644 index 0000000..f2af0e9 --- /dev/null +++ b/Lab3/ChatRoomContract/ChatRoomClient.cs @@ -0,0 +1,241 @@ +using MessagePack; +using RabbitMQ.Client; +using RabbitMQ.Client.Events; + +namespace ChatRoomContract; + +/// +/// RabbitMQ chat room client +/// +public class ChatRoomClient : IChatRoomService +{ + /// + /// Exchange name + /// + private string exchangeName; + /// + /// Client queue name + /// + private string clientQueueName; + /// + /// Server queue name + /// + private string serverQueueName; + /// + /// RabbitMQ channel + /// + private IModel rmqChannel; + + /// + /// Chat Room client constructor + /// + /// RabbitMQ connection + /// Exchange name + /// Client queue name + /// Server queue name + public ChatRoomClient(IConnection connection, string exchangeName, string clientQueueName, string serverQueueName) + { + this.exchangeName = exchangeName; + this.clientQueueName = clientQueueName; + this.serverQueueName = serverQueueName; + rmqChannel = connection.CreateModel(); + + rmqChannel.ExchangeDeclare(exchange: exchangeName, type: ExchangeType.Direct); + rmqChannel.QueueDeclare(queue: clientQueueName, durable: false, exclusive: true, autoDelete: false, arguments: null); + rmqChannel.QueueBind(queue: clientQueueName, exchange: exchangeName, routingKey: clientQueueName, arguments: null); + } + + /// + /// Send rpc message without waiting for a response + /// + /// Method name + /// Serialized arguments + /// Optional correlation ID + private void CallVoid(string method, byte[] args, string? correlationId = null) + { + correlationId ??= Guid.NewGuid().ToString(); + + var requestProps = rmqChannel.CreateBasicProperties(); + requestProps.CorrelationId = correlationId; + requestProps.ReplyTo = clientQueueName; + + var msg = new RPCMessage{ + isResponse = false, + method = method, + args = args + }; + + rmqChannel.BasicPublish( + exchange: exchangeName, + routingKey: serverQueueName, + basicProperties: requestProps, + body: MessagePackSerializer.Serialize(msg) + ); + } + + /// + /// Send a rpc message and wait for a response + /// + /// Result type + /// Method name + /// Serialized arguments + /// Result + private ResultType Call(string method, byte[]? args) + { + var correlationId = Guid.NewGuid().ToString(); + var isResultReady = false; + var resultReadySignal = new AutoResetEvent(false); + + ResultType result = default; + + //ensure contents of variables set in main thread, are loadable by receiver thread + Thread.MemoryBarrier(); + + //create response message consumer + var consumer = new EventingBasicConsumer(rmqChannel); + consumer.Received += + (channel, delivery) => { + //ensure contents of variables set in main thread are loaded into this thread + Thread.MemoryBarrier(); + + if (isResultReady) + { + return; + } + + if (delivery.BasicProperties.CorrelationId != correlationId) + { + return; + } + + + var msg = MessagePackSerializer.Deserialize(delivery.Body); + if (msg.isResponse && msg.method == method) + { + if (msg.args != null) + { + result = MessagePackSerializer.Deserialize(msg.args); + } + + //indicate result has been received, ensure it is loadable by main thread + isResultReady = true; + Thread.MemoryBarrier(); + + //signal main thread that result has been received + resultReadySignal.Set(); + } + }; + + //attach message consumer to the response queue + var consumerTag = rmqChannel.BasicConsume(clientQueueName, true, consumer); + + CallVoid(method, args, correlationId); + + //wait for the result to be ready + resultReadySignal.WaitOne(); + + //ensure contents of variables set by the receiver are loaded into this thread + Thread.MemoryBarrier(); + + //detach message consumer from the response queue + rmqChannel.BasicCancel(consumerTag); + + return result; + } + + /// + /// Approve a message + /// + /// Message ID + public void ApproveMessage(int messageId) + { + CallVoid( + nameof(IChatRoomService.ApproveMessage), + MessagePackSerializer.Serialize(messageId) + ); + } + + /// + /// Get timestamp until when the client is blocked + /// + /// Client ID + /// Optional datetime object + public DateTime? GetBlockedUntil(int clientId) + { + return Call( + nameof(IChatRoomService.GetBlockedUntil), + MessagePackSerializer.Serialize(clientId) + ); + } + + /// + /// Get the next message which hasn't been approved or rejected + /// + /// Message object. Returns null if there is no message + public Message? GetNewMessage() + { + return Call( + nameof(IChatRoomService.GetNewMessage), + null + ); + } + + /// + /// Get number of strikes a participant has + /// + /// Client ID + /// Number of strikes + public int GetStrikes(int clientId) + { + return Call( + nameof(IChatRoomService.GetStrikes), + MessagePackSerializer.Serialize(clientId) + ); + } + + /// + /// Register client with a name + /// + /// Name of client, can be duplicate between clients + /// Client ID + public int RegisterClient(string name) + { + return Call( + nameof(IChatRoomService.RegisterClient), + MessagePackSerializer.Serialize(name) + ); + } + + /// + /// Reject a message + /// + /// Message ID + public void RejectMessage(int messageId) + { + CallVoid( + nameof(IChatRoomService.RejectMessage), + MessagePackSerializer.Serialize(messageId) + ); + } + + /// + /// Send a message, will be given to a moderator to be approved + /// + /// Client ID + /// Message contents + /// Does this message need to be censored? + /// Was sending successful, can fail if user is blocked + public bool SendMessage(int clientId, string contents, bool needsToBeCensored) + { + var args = new SendMessageArgs { + clientId = clientId, + contents = contents, + needsToBeCensored = needsToBeCensored + }; + + return Call( + nameof(IChatRoomService.SendMessage), + MessagePackSerializer.Serialize(args) + ); + } +} diff --git a/Lab3/ChatRoomContract/ChatRoomContract.csproj b/Lab3/ChatRoomContract/ChatRoomContract.csproj new file mode 100644 index 0000000..cabe6ba --- /dev/null +++ b/Lab3/ChatRoomContract/ChatRoomContract.csproj @@ -0,0 +1,14 @@ + + + + net6.0 + enable + enable + + + + + + + + diff --git a/Lab3/ChatRoomContract/Config.cs b/Lab3/ChatRoomContract/Config.cs new file mode 100644 index 0000000..4aeb1e2 --- /dev/null +++ b/Lab3/ChatRoomContract/Config.cs @@ -0,0 +1,30 @@ +using RabbitMQ.Client; + +namespace ChatRoomContract; + +public static class Config +{ + public static string ExchangeName = "T120B180.ChatRoom.Exchange"; + public static string ServerQueueName = "T120B180.ChatRoom.Server"; + public static string ClientQueueNamePrefix = "T120B180.ChatRoom.Client_"; + public static string HostName = "localhost"; + public static string Username = "guest"; + public static string Password = "guest"; + + public static string CreateClientQueueName() + { + var ClientId = Guid.NewGuid().ToString(); + return ClientQueueNamePrefix + ClientId; + } + + public static IConnection CreateConnection() + { + var rmqConnectionFactory = new ConnectionFactory + { + HostName = HostName, + UserName = Username, + Password = Password + }; + return rmqConnectionFactory.CreateConnection(); + } +} diff --git a/Lab3/ChatRoomContract/IChatRoomService.cs b/Lab3/ChatRoomContract/IChatRoomService.cs new file mode 100644 index 0000000..fbf96ed --- /dev/null +++ b/Lab3/ChatRoomContract/IChatRoomService.cs @@ -0,0 +1,81 @@ +using MessagePack; + +namespace ChatRoomContract; + +/// +/// Minimal message description +/// +[MessagePackObject] +public class Message +{ + /// + /// Message ID + /// + [Key(0)] + public int id; + /// + /// Message contents + /// + [Key(1)] + public string contents; + /// + /// Does this message need to be censored? + /// + [Key(2)] + public bool needsToBeCensored; +} + +/// +/// Chat room service contract +/// +public interface IChatRoomService +{ + /// + /// Register client with a name + /// + /// Name of client, can be duplicate between clients + /// Client ID + int RegisterClient(string name); + + /// + /// Get number of strikes a participant has + /// + /// Client ID + /// Number of strikes + int GetStrikes(int clientId); + + /// + /// Get timestamp until when the client is blocked + /// + /// Client ID + /// Optional datetime object + DateTime? GetBlockedUntil(int clientId); + + /// + /// Send a message, will be given to a moderator to be approved + /// + /// Client ID + /// Message contents + /// Does this message need to be censored? + /// Was sending successful, can fail if user is blocked + bool SendMessage(int clientId, string contents, bool needsToBeCensored); + + /// + /// Get the next message which hasn't been approved or rejected + /// + /// Message object. Returns null if there is no message + Message? GetNewMessage(); + + /// + /// Reject a message + /// + /// Message ID + void RejectMessage(int messageId); + + /// + /// Approve a message + /// + /// Message ID + void ApproveMessage(int messageId); +} + diff --git a/Lab3/ChatRoomContract/RPCMessage.cs b/Lab3/ChatRoomContract/RPCMessage.cs new file mode 100644 index 0000000..2acc5b1 --- /dev/null +++ b/Lab3/ChatRoomContract/RPCMessage.cs @@ -0,0 +1,49 @@ +using MessagePack; + +namespace ChatRoomContract; + +/// +/// RabbitMQ message +/// +[MessagePackObject] +public class RPCMessage +{ + /// + /// Is this message a response + /// + [Key(0)] + public bool isResponse; + /// + /// Method name + /// + [Key(1)] + public string method; + /// + /// Optional arguments + /// + [Key(2)] + public byte[]? args; +} + +/// +/// IChatRoomService.SendMessage arguments +/// +[MessagePackObject] +public class SendMessageArgs +{ + /// + /// Client ID + /// + [Key(0)] + public int clientId { get; set; } + /// + /// Message contents + /// + [Key(1)] + public string contents { get; set; } + /// + /// Does this message need to be censored? + /// + [Key(2)] + public bool needsToBeCensored { get; set; } +} \ No newline at end of file diff --git a/Lab3/Lab3.sln b/Lab3/Lab3.sln new file mode 100644 index 0000000..9c1103f --- /dev/null +++ b/Lab3/Lab3.sln @@ -0,0 +1,43 @@ + +Microsoft Visual Studio Solution File, Format Version 12.00 +# Visual Studio Version 17 +VisualStudioVersion = 17.11.35208.52 +MinimumVisualStudioVersion = 10.0.40219.1 +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ChatRoom", "ChatRoom\ChatRoom.csproj", "{BDE934D4-F8C9-484F-965F-FBBDE10818EE}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ChatRoomContract", "ChatRoomContract\ChatRoomContract.csproj", "{37F160C5-7D5A-4EF4-ADBB-C7F4065F2ECF}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Moderator", "Moderator\Moderator.csproj", "{CC693E3A-9508-4D87-8B5F-630DD268858F}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Participant", "Participant\Participant.csproj", "{F65D4958-FE80-4B62-ADAD-F27BC6B5C696}" +EndProject +Global + GlobalSection(SolutionConfigurationPlatforms) = preSolution + Debug|Any CPU = Debug|Any CPU + Release|Any CPU = Release|Any CPU + EndGlobalSection + GlobalSection(ProjectConfigurationPlatforms) = postSolution + {BDE934D4-F8C9-484F-965F-FBBDE10818EE}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {BDE934D4-F8C9-484F-965F-FBBDE10818EE}.Debug|Any CPU.Build.0 = Debug|Any CPU + {BDE934D4-F8C9-484F-965F-FBBDE10818EE}.Release|Any CPU.ActiveCfg = Release|Any CPU + {BDE934D4-F8C9-484F-965F-FBBDE10818EE}.Release|Any CPU.Build.0 = Release|Any CPU + {37F160C5-7D5A-4EF4-ADBB-C7F4065F2ECF}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {37F160C5-7D5A-4EF4-ADBB-C7F4065F2ECF}.Debug|Any CPU.Build.0 = Debug|Any CPU + {37F160C5-7D5A-4EF4-ADBB-C7F4065F2ECF}.Release|Any CPU.ActiveCfg = Release|Any CPU + {37F160C5-7D5A-4EF4-ADBB-C7F4065F2ECF}.Release|Any CPU.Build.0 = Release|Any CPU + {CC693E3A-9508-4D87-8B5F-630DD268858F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {CC693E3A-9508-4D87-8B5F-630DD268858F}.Debug|Any CPU.Build.0 = Debug|Any CPU + {CC693E3A-9508-4D87-8B5F-630DD268858F}.Release|Any CPU.ActiveCfg = Release|Any CPU + {CC693E3A-9508-4D87-8B5F-630DD268858F}.Release|Any CPU.Build.0 = Release|Any CPU + {F65D4958-FE80-4B62-ADAD-F27BC6B5C696}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {F65D4958-FE80-4B62-ADAD-F27BC6B5C696}.Debug|Any CPU.Build.0 = Debug|Any CPU + {F65D4958-FE80-4B62-ADAD-F27BC6B5C696}.Release|Any CPU.ActiveCfg = Release|Any CPU + {F65D4958-FE80-4B62-ADAD-F27BC6B5C696}.Release|Any CPU.Build.0 = Release|Any CPU + EndGlobalSection + GlobalSection(SolutionProperties) = preSolution + HideSolutionNode = FALSE + EndGlobalSection + GlobalSection(ExtensibilityGlobals) = postSolution + SolutionGuid = {B2EFC0A1-DB91-4736-B795-29ACE70DCBA4} + EndGlobalSection +EndGlobal diff --git a/Lab3/Moderator/Moderator.cs b/Lab3/Moderator/Moderator.cs new file mode 100644 index 0000000..15587aa --- /dev/null +++ b/Lab3/Moderator/Moderator.cs @@ -0,0 +1,102 @@ +using ChatRoomContract; +using NLog; +using Bogus; + +namespace Moderator; + +internal class Moderator +{ + /// + /// Logger for this class. + /// + Logger log = LogManager.GetCurrentClassLogger(); + + /// + /// Configures logging subsystem. + /// + private void ConfigureLogging() + { + var config = new NLog.Config.LoggingConfiguration(); + + var console = + new NLog.Targets.ConsoleTarget("console") + { + Layout = @"${date:format=HH\:mm\:ss}|${level}| ${message} ${exception}" + }; + config.AddTarget(console); + config.AddRuleForAllLevels(console); + + LogManager.Configuration = config; + } + + /// + /// Run with a given service + /// + /// Chat room service + private void RunConnection(IChatRoomService chatRoom) + { + var faker = new Faker("en"); + + var name = faker.Name.FullName(); + int clientId = chatRoom.RegisterClient(name); + log.Info($"Registered with client id {clientId}"); + + Console.Title = $"Moderator | {name} | {clientId}"; + + while (true) + { + var message = chatRoom.GetNewMessage(); + if (message != null) + { + log.Info($"Checking message ({message.id}): {message.contents}"); + Thread.Sleep(500); + + if (message.needsToBeCensored) + { + chatRoom.RejectMessage(message.id); + } + else + { + chatRoom.ApproveMessage(message.id); + } + } + + Thread.Sleep(1 * 1000); + } + } + + /// + /// Main loop + /// + private void Run() + { + ConfigureLogging(); + + while (true) + { + var chatRoom = new ChatRoomClient(Config.CreateConnection(), Config.ExchangeName, Config.CreateClientQueueName(), Config.ServerQueueName); + + try + { + RunConnection(chatRoom); + } + catch (Exception e) + { + //log whatever exception to console + log.Warn(e, "Unhandled exception caught. Will restart main loop."); + + //prevent console spamming + Thread.Sleep(2000); + } + } + } + + /// + /// Entry point + /// + static void Main() + { + var self = new Moderator(); + self.Run(); + } +} diff --git a/Lab3/Moderator/Moderator.csproj b/Lab3/Moderator/Moderator.csproj new file mode 100644 index 0000000..ac175f5 --- /dev/null +++ b/Lab3/Moderator/Moderator.csproj @@ -0,0 +1,19 @@ + + + + Exe + net6.0 + enable + enable + + + + + + + + + + + + diff --git a/Lab3/Participant/Participant.cs b/Lab3/Participant/Participant.cs new file mode 100644 index 0000000..44d579c --- /dev/null +++ b/Lab3/Participant/Participant.cs @@ -0,0 +1,107 @@ +using ChatRoomContract; +using NLog; +using Bogus; + +namespace Participant; + +internal class Participant +{ + /// + /// Logger for this class. + /// + Logger log = LogManager.GetCurrentClassLogger(); + + /// + /// Configures logging subsystem. + /// + private void ConfigureLogging() + { + var config = new NLog.Config.LoggingConfiguration(); + + var console = + new NLog.Targets.ConsoleTarget("console") + { + Layout = @"${date:format=HH\:mm\:ss}|${level}| ${message} ${exception}" + }; + config.AddTarget(console); + config.AddRuleForAllLevels(console); + + LogManager.Configuration = config; + } + + /// + /// Run with a given service + /// + /// Chat room service + private void RunConnection(IChatRoomService chatRoom) + { + var faker = new Faker("en"); + var rnd = new Random(); + + var name = faker.Name.FullName(); + int clientId = chatRoom.RegisterClient(name); + log.Info($"Registered with client id {clientId}"); + + while (true) + { + int strikes = chatRoom.GetStrikes(clientId); + Console.Title = $"Participant | {name} | {clientId} | {strikes} Strikes"; + + var message = string.Join(" ", faker.Lorem.Words(5)); + bool needsToBeCensored = rnd.Next(0, 100) > 50; + if (chatRoom.SendMessage(clientId, message, needsToBeCensored)) + { + log.Info("Sent message"); + } + else + { + log.Info("Failed to send message, blocked"); + var blockedUntil = chatRoom.GetBlockedUntil(clientId); + var now = DateTime.UtcNow; + if (blockedUntil != null && blockedUntil > now) + { + var delta = blockedUntil.Value - now; + log.Info($"Waiting {delta.TotalSeconds:F3}s until block expires"); + Thread.Sleep((int)delta.TotalMilliseconds); + } + } + + Thread.Sleep(2 * 1000); + } + } + + /// + /// Main loop + /// + private void Run() + { + ConfigureLogging(); + + while (true) + { + var chatRoom = new ChatRoomClient(Config.CreateConnection(), Config.ExchangeName, Config.CreateClientQueueName(), Config.ServerQueueName); + + try + { + RunConnection(chatRoom); + } + catch (Exception e) + { + //log whatever exception to console + log.Warn(e, "Unhandled exception caught. Will restart main loop."); + + //prevent console spamming + Thread.Sleep(2000); + } + } + } + + /// + /// Entry point + /// + static void Main() + { + var self = new Participant(); + self.Run(); + } +} diff --git a/Lab3/Participant/Participant.csproj b/Lab3/Participant/Participant.csproj new file mode 100644 index 0000000..ac175f5 --- /dev/null +++ b/Lab3/Participant/Participant.csproj @@ -0,0 +1,19 @@ + + + + Exe + net6.0 + enable + enable + + + + + + + + + + + +