tinklu-paslaugos/Lab4/ChatRoomContract/ChatRoomRabbitMQClient.cs

using MessagePack;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace ChatRoomContract;

/// <summary>
/// RabbitMQ chat room client
/// </summary>
public class ChatRoomRabbitMQClient : IChatRoomService
{
    /// <summary>
    /// Exchange name
    /// </summary>
    private string exchangeName;
    /// <summary>
    /// Client queue name
    /// </summary>
    private string clientQueueName;
    /// <summary>
    /// Server queue name
    /// </summary>
    private string serverQueueName;
    /// <summary>
    /// RabbitMQ channel
    /// </summary>
    private IModel rmqChannel;

    /// <summary>
    /// Chat Room client constructor
    /// </summary>
    /// <param name="connection">RabbitMQ connection</param>
    /// <param name="exchangeName">Exchange name</param>
    /// <param name="clientQueueName">Client queue name</param>
    /// <param name="serverQueueName">Server queue name</param>
    public ChatRoomRabbitMQClient(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);
    }

    /// <summary>
    /// Send rpc message without waiting for a response
    /// </summary>
    /// <param name="method">Method name</param>
    /// <param name="args">Serialized arguments</param>
    /// <param name="correlationId">Optional correlation ID</param>
    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)
        );
    }

    /// <summary>
    /// Send a rpc message and wait for a response
    /// </summary>
    /// <typeparam name="ResultType">Result type</typeparam>
    /// <param name="method">Method name</param>
    /// <param name="args">Serialized arguments</param>
    /// <returns>Result</returns>
    private ResultType Call<ResultType>(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<RPCMessage>(delivery.Body);
                if (msg.isResponse && msg.method == method)
                {
                    if (msg.args != null)
                    {
                        result = MessagePackSerializer.Deserialize<ResultType>(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;
    }

    /// <summary>
    /// Approve a message
    /// </summary>
    /// <param name="messageId">Message ID</param>
    public void ApproveMessage(int messageId)
    {
        CallVoid(
            nameof(IChatRoomService.ApproveMessage),
            MessagePackSerializer.Serialize(messageId)
        );
    }

    /// <summary>
    /// Get timestamp until when the client is blocked
    /// </summary>
    /// <param name="clientId">Client ID</param>
    /// <returns>Optional datetime object</returns>
    public DateTime? GetBlockedUntil(int clientId)
    {
        return Call<DateTime?>(
            nameof(IChatRoomService.GetBlockedUntil),
            MessagePackSerializer.Serialize(clientId)
        );
    }

    /// <summary>
    /// Get the next message which hasn't been approved or rejected
    /// </summary>
    /// <returns>Message object. Returns null if there is no message</returns>
    public Message? GetNewMessage()
    {
        return Call<Message?>(
            nameof(IChatRoomService.GetNewMessage),
            null
        );
    }

    /// <summary>
    /// Get number of strikes a participant has
    /// </summary>
    /// <param name="clientId">Client ID</param>
    /// <returns>Number of strikes</returns>
    public int GetStrikes(int clientId)
    {
        return Call<int>(
            nameof(IChatRoomService.GetStrikes),
            MessagePackSerializer.Serialize(clientId)
        );
    }

    /// <summary>
    /// Register client with a name
    /// </summary>
    /// <param name="name">Name of client, can be duplicate between clients</param>
    /// <returns>Client ID</returns>
    public int RegisterClient(string name)
    {
        return Call<int>(
            nameof(IChatRoomService.RegisterClient),
            MessagePackSerializer.Serialize(name)
        );
    }

    /// <summary>
    /// Reject a message
    /// </summary>
    /// <param name="messageId">Message ID</param>
    public void RejectMessage(int messageId)
    {
        CallVoid(
            nameof(IChatRoomService.RejectMessage),
            MessagePackSerializer.Serialize(messageId)
        );
    }

    /// <summary>
    /// Send a message, will be given to a moderator to be approved
    /// </summary>
    /// <param name="clientId">Client ID</param>
    /// <param name="contents">Message contents</param>
    /// <param name="needsToBeCensored">Does this message need to be censored?</param>
    /// <returns>Was sending successful, can fail if user is blocked</returns>
    public bool SendMessage(int clientId, string contents, bool needsToBeCensored)
    {
        var args = new SendMessageArgs
        {
            clientId = clientId,
            contents = contents,
            needsToBeCensored = needsToBeCensored
        };

        return Call<bool>(
            nameof(IChatRoomService.SendMessage),
            MessagePackSerializer.Serialize(args)
        );
    }
}