<commit_before>﻿using System;
using System.IO;
using System.Text;
using System.Threading;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace Mike.AmqpSpike
{
    public class AckNackSpike
    {
        private const string ackNackQueue = "ackNackQueue";

        public void PublishSequence()
        {
            WithChannel.Do(channel =>
            {
                for (int i = 0; i < 10; i++)
                {
                    var properties = channel.CreateBasicProperties();
                    var body = Encoding.UTF8.GetBytes(i.ToString());
                    channel.BasicPublish("", ackNackQueue, properties, body);
                }
            });
        }

        public void InitializeQueue()
        {
            WithChannel.Do(channel => 
                channel.QueueDeclare(
                    ackNackQueue, 
                    durable: true, 
                    exclusive: false, 
                    autoDelete: false, 
                    arguments: null));
        }

        public void SubscribeAck()
        {
            const int prefetchCount = 0;
            const bool noAck = false;
            const int numberOfMessagesToConsume = 10;

            WithChannel.Do(channel =>
            {
                channel.BasicQos(0, prefetchCount, false);

                var consumer = new QueueingBasicConsumer(channel);
                channel.BasicConsume(ackNackQueue, noAck, consumer);

                var running = true;
                var thread = new Thread(_ =>
                {
                    var count = 0;
                    while (running && count++ < numberOfMessagesToConsume)
                    {
                        try
                        {
                            var basicDeliverEventArgs = (BasicDeliverEventArgs) consumer.Queue.Dequeue();

                            if (basicDeliverEventArgs != null)
                            {
                                var message = Encoding.UTF8.GetString(basicDeliverEventArgs.Body);
                                Console.Out.WriteLine("message = {0}", message);

                                consumer.Model.BasicAck(basicDeliverEventArgs.DeliveryTag, false);
                                //consumer.Model.BasicNack(basicDeliverEventArgs.DeliveryTag, false, requeue:true);
                            }
                        }
                        catch (EndOfStreamException exception)
                        {
                            break;
                        }
                    }

                });
                thread.Start();

                Thread.Sleep(1000);
                running = false;
            });
        }
    }
}<commit_msg>Spiking QoS prefetch count - logging HandleBasicDeliver<commit_after>﻿using System;
using System.IO;
using System.Text;
using System.Threading;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Util;

namespace Mike.AmqpSpike
{
    public class AckNackSpike
    {
        private const string ackNackQueue = "ackNackQueue";

        public void PublishSequence()
        {
            WithChannel.Do(channel =>
            {
                for (int i = 0; i < 10; i++)
                {
                    var properties = channel.CreateBasicProperties();
                    var body = Encoding.UTF8.GetBytes(i.ToString());
                    channel.BasicPublish("", ackNackQueue, properties, body);
                }
            });
        }

        public void InitializeQueue()
        {
            WithChannel.Do(channel => 
                channel.QueueDeclare(
                    ackNackQueue, 
                    durable: true, 
                    exclusive: false, 
                    autoDelete: false, 
                    arguments: null));
        }

        public void SubscribeAck()
        {
            const int prefetchCount = 1;
            const bool noAck = false;
            const int numberOfMessagesToConsume = 1;

            WithChannel.Do(channel =>
            {
                channel.BasicQos(0, prefetchCount, false);

                var consumer = new LocalQueueingBasicConsumer(channel);
                channel.BasicConsume(ackNackQueue, noAck, consumer);

                var running = true;
                var thread = new Thread(_ =>
                {
                    var count = 0;
                    while (running && count++ < numberOfMessagesToConsume)
                    {
                        try
                        {
                            var basicDeliverEventArgs = (BasicDeliverEventArgs) consumer.Queue.Dequeue();

                            if (basicDeliverEventArgs != null)
                            {
                                var message = Encoding.UTF8.GetString(basicDeliverEventArgs.Body);
                                Console.Out.WriteLine("message = {0}", message);

                                consumer.Model.BasicAck(basicDeliverEventArgs.DeliveryTag, false);
                                //consumer.Model.BasicNack(basicDeliverEventArgs.DeliveryTag, false, requeue:true);
                            }
                        }
                        catch (EndOfStreamException exception)
                        {
                            break;
                        }
                    }

                });
                thread.Start();

                Thread.Sleep(1000);
                running = false;
            });
        }
    }

    public class LocalQueueingBasicConsumer : DefaultBasicConsumer
    {
        protected SharedQueue m_queue;

        public SharedQueue Queue
        {
            get
            {
                return this.m_queue;
            }
        }

        public LocalQueueingBasicConsumer()
            : this((IModel)null)
        {
        }

        public LocalQueueingBasicConsumer(IModel model)
            : this(model, new SharedQueue())
        {
        }

        public LocalQueueingBasicConsumer(IModel model, SharedQueue queue)
            : base(model)
        {
            this.m_queue = queue;
        }

        public override void OnCancel()
        {
            this.m_queue.Close();
            base.OnCancel();
        }

        public override void HandleBasicDeliver(string consumerTag, ulong deliveryTag, bool redelivered, string exchange, string routingKey, IBasicProperties properties, byte[] body)
        {
            Console.Out.WriteLine("HandleBasicDeliver");
            this.m_queue.Enqueue((object)new BasicDeliverEventArgs()
            {
                ConsumerTag = consumerTag,
                DeliveryTag = deliveryTag,
                Redelivered = redelivered,
                Exchange = exchange,
                RoutingKey = routingKey,
                BasicProperties = properties,
                Body = body
            });
        }
    }

}
{'fraction_non_alphanumeric': 0.07598784194528875, 'fraction_numerical': 0.004197423650311188, 'mean_word_length': 1.1607254534083802, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '336714', 'n_tokens_mistral': 1701, 'n_tokens_neox': 1540, 'n_words': 325}