Управление сообщениями очереди с помощью клиента RabbitMQ

Получать сообщения из очереди для просмотра

Этот метод будет вызван, как только мы нажмем на любую очередь, чтобы увидеть сообщения. Для этого мы удалим каждое сообщение и NACK, чтобы вернуть его в очередь. Это эквивалентно функциональности PEEK, но, поскольку RabbitMQ Client не предоставляет этого готового решения, мы должны сделать этот обходной путь. Нам нужно только просмотреть эти данные и затем решить, какую операцию нам нужно выполнить (очистить / переместить). Кроме того, мы передаем количество сообщений, которые нам нужно просмотреть — это будет полезно при поиске страниц.

public List<Message> GetMessagesFromQueueNoAck ( string queueName, int messageCount = -1 )
        {
            QueueingBasicConsumer consumer = null;          
            var responseMessages = new List<Message> ();
            BasicDeliverEventArgs result = null;
            using (var rmqConnection = GetRabbitMqConnection ())
            {
                using (var channel = rmqConnection.CreateModel ())
                {
                    try
                    {
                        var queueMessageCount = (int)channel.MessageCount (queueName);
                        var queueInfo = GetAllQueueData ("", "", queueName).FirstOrDefault ();
                        var count = messageCount > -1 ? messageCount <= queueMessageCount ? messageCount : queueMessageCount : queueMessageCount;
                        var pfCount = count >= PREFETCH_SIZE ? PREFETCH_SIZE : count;
                        channel.BasicQos (0, (ushort)pfCount, false);
                        consumer = new QueueingBasicConsumer (channel);
                        channel.BasicConsume (queueName, false, consumer);
                        for (int i = 0; i < pfCount; i++)
                        {
                            if (!channel.IsOpen)
                            {
                                throw new ApplicationException ("Channel is closed");
                            }
                            result = consumer.Queue.Dequeue ();
                            try
                            {
                                string messageData = System.Text.Encoding.UTF8.GetString (result.Body);
                                var rMessage = new Message (messageData);
                                RmqHeaderHandler.ReadRmqMessageProperties (result.BasicProperties, rMessage);
                                channel.BasicNack (result.DeliveryTag, false, true);
                                ////Set Message properties
                                Type t = queueInfo.GetType ();
                                foreach (PropertyInfo pi in t.GetProperties ())
                                {
                                    rMessage.Header.Properties.Add (pi.Name, pi.GetValue (queueInfo, null).ToString ());
                                }
                                responseMessages.Add (rMessage);
                            }
                            catch (Exception ex)
                            {
                                ////Nack the message in case of any exception while reading the message.
                                channel.BasicNack (result.DeliveryTag, false, true);
                                this.logger.Warn ("Error Occured while getting message for message ID" + result.BasicProperties.MessageId, ex);
                            }
                        }
                    }
                    catch (Exception)
                    {
                        ////Nack the message back to queue in case of exception
                        if (result != null)
                        {
                            channel.BasicNack (result.DeliveryTag, false, true);
                        }
                        throw;
                    }
                }
            }
            return responseMessages;
        }

public static class RmqHeaderHandler
    {
        private const byte NonPersistentDeliveryMode = 1;
        private const byte PersistentDeliveryMode = 2;
        private const string SecurityTokenKey = "SecurityToken";
        private const string Properties = "properties";
        private const string MessageNameKey = "MessageName";
        private const string SystemPropertiesKey = "SystemProperties";
        private const string ApplicationPropertiesKey = "ApplicationProperties";
        #region Public Methods
        public static void ReadDynamicMessageProperties(dynamic messageProperties, Message message)
        {
            try
            {                   
                message.Header.AppId = messageProperties.appId;
                message.Header.MessageId = messageProperties.messageId;  
                message.Header.GeneratedAtUtc = messageProperties.generatedAtUtc;
                message.Header.ExpirationInMilliseconds = messageProperties.expirationInMilliseconds;
                message.Header.IsPersistent = messageProperties.isPersistent;
                message.Header.Delete = Convert.ToBoolean(messageProperties.delete);
                message.Header.Move = Convert.ToBoolean(messageProperties.move);
                if (messageProperties.ContainsKey ("messageName"))
                {
                    message.Header.MessageName = messageProperties.messageName;
                }                
                if (messageProperties.ContainsKey("properties"))
                {  
                    var customProperties = Newtonsoft.Json.JsonConvert.DeserializeObject<Dictionary<string, string>>(Convert.ToString(messageProperties.properties));
                    foreach (var propPair in customProperties)
                    {
                        message.Header.Properties.Add(propPair.Key, propPair.Value);
                    }
                }
            }
            catch(Exception)
            {
                throw;
            }
        }

        public static void ReadRmqMessageProperties(IBasicProperties messageProperties, Message message)
        {
            message.Header.AppId = messageProperties.AppId;
            message.Header.MessageId = messageProperties.MessageId;
            message.Header.GeneratedAtUtc = new DateTime(messageProperties.Timestamp.UnixTime);
            message.Header.ExpirationInMilliseconds = messageProperties.Expiration;
            message.Header.IsPersistent = messageProperties.DeliveryMode == PersistentDeliveryMode;
            if (messageProperties.Headers.ContainsKey (SystemPropertiesKey))
            {
                var systemProperties = DeserializeMessageProperties ((byte[])messageProperties.Headers[SystemPropertiesKey]);
                if (systemProperties.ContainsKey (MessageNameKey))
                {
                    message.Header.MessageName = systemProperties[MessageNameKey];
                }
            }
            if (messageProperties.Headers.ContainsKey(ApplicationPropertiesKey))
            {
                var applicationProperties = DeserializeMessageProperties((byte[])messageProperties.Headers[ApplicationPropertiesKey]);
                foreach (var propPair in applicationProperties)
                {
                    message.Header.Properties.Add(propPair.Key, propPair.Value);
                }
            }
        }

        #endregion
        #region private methods        
        private static Dictionary<string, string> DeserializeMessageProperties(byte[] properties)
        {
            var serializer = new JsonMessageSerializer();
            var serializedText = serializer.Serialize(properties);
            return serializer.Deserialize<Dictionary<string, string>>(serializedText);
        }
        #endregion
}