<commit_before>namespace MassTransit.RabbitMqTransport.Pipeline
{
    using System;
    using System.Linq;
    using System.Threading.Tasks;
    using Context;
    using GreenPipes;
    using Topology.Builders;
    using Topology.Entities;


    /// <summary>
    /// Configures the broker with the supplied topology once the model is created, to ensure
    /// that the exchanges, queues, and bindings for the model are properly configured in RabbitMQ.
    /// </summary>
    public class ConfigureTopologyFilter<TSettings> :
        IFilter<ModelContext>
        where TSettings : class
    {
        readonly BrokerTopology _brokerTopology;
        readonly TSettings _settings;

        public ConfigureTopologyFilter(TSettings settings, BrokerTopology brokerTopology)
        {
            _settings = settings;
            _brokerTopology = brokerTopology;
        }

        async Task IFilter<ModelContext>.Send(ModelContext context, IPipe<ModelContext> next)
        {
            await context.OneTimeSetup<ConfigureTopologyContext<TSettings>>(async payload =>
            {
                await ConfigureTopology(context).ConfigureAwait(false);

                context.GetOrAddPayload(() => _settings);
            }, () => new Context()).ConfigureAwait(false);

            await next.Send(context).ConfigureAwait(false);
        }

        void IProbeSite.Probe(ProbeContext context)
        {
            var scope = context.CreateFilterScope("configureTopology");

            _brokerTopology.Probe(scope);
        }

        async Task ConfigureTopology(ModelContext context)
        {
            await Task.WhenAll(_brokerTopology.Exchanges.Select(exchange => Declare(context, exchange))).ConfigureAwait(false);

            await Task.WhenAll(_brokerTopology.ExchangeBindings.Select(binding => Bind(context, binding))).ConfigureAwait(false);

            await Task.WhenAll(_brokerTopology.Queues.Select(queue => Declare(context, queue))).ConfigureAwait(false);

            await Task.WhenAll(_brokerTopology.QueueBindings.Select(binding => Bind(context, binding))).ConfigureAwait(false);
        }

        Task Declare(ModelContext context, Exchange exchange)
        {
            RabbitMqLogMessages.DeclareExchange(exchange);

            return context.ExchangeDeclare(exchange.ExchangeName, exchange.ExchangeType, exchange.Durable, exchange.AutoDelete, exchange.ExchangeArguments);
        }

        async Task Declare(ModelContext context, Queue queue)
        {
            try
            {
                var ok = await context.QueueDeclare(queue.QueueName, queue.Durable, queue.Exclusive, queue.AutoDelete, queue.QueueArguments)
                    .ConfigureAwait(false);

                RabbitMqLogMessages.DeclareQueue(queue, ok.ConsumerCount, ok.MessageCount);
            }
            catch (Exception exception)
            {
                LogContext.Error?.Log(exception, "Declare queue faulted: {Queue}", queue);

                throw;
            }
        }

        Task Bind(ModelContext context, ExchangeToExchangeBinding binding)
        {
            RabbitMqLogMessages.BindToExchange(binding);

            return context.ExchangeBind(binding.Destination.ExchangeName, binding.Source.ExchangeName, binding.RoutingKey, binding.Arguments);
        }

        Task Bind(ModelContext context, ExchangeToQueueBinding binding)
        {
            RabbitMqLogMessages.BindToQueue(binding);

            return context.QueueBind(binding.Destination.QueueName, binding.Source.ExchangeName, binding.RoutingKey, binding.Arguments);
        }


        class Context :
            ConfigureTopologyContext<TSettings>
        {
        }
    }
}<commit_msg>Changed topology configuration order and add settle time for RabbitMQ to be ready to accept messages<commit_after>namespace MassTransit.RabbitMqTransport.Pipeline
{
    using System;
    using System.Linq;
    using System.Threading.Tasks;
    using Context;
    using GreenPipes;
    using Topology.Builders;
    using Topology.Entities;


    /// <summary>
    /// Configures the broker with the supplied topology once the model is created, to ensure
    /// that the exchanges, queues, and bindings for the model are properly configured in RabbitMQ.
    /// </summary>
    public class ConfigureTopologyFilter<TSettings> :
        IFilter<ModelContext>
        where TSettings : class
    {
        readonly BrokerTopology _brokerTopology;
        readonly TSettings _settings;

        public ConfigureTopologyFilter(TSettings settings, BrokerTopology brokerTopology)
        {
            _settings = settings;
            _brokerTopology = brokerTopology;
        }

        async Task IFilter<ModelContext>.Send(ModelContext context, IPipe<ModelContext> next)
        {
            await context.OneTimeSetup<ConfigureTopologyContext<TSettings>>(async payload =>
            {
                await ConfigureTopology(context).ConfigureAwait(false);

                context.GetOrAddPayload(() => _settings);
            }, () => new Context()).ConfigureAwait(false);

            await next.Send(context).ConfigureAwait(false);
        }

        void IProbeSite.Probe(ProbeContext context)
        {
            var scope = context.CreateFilterScope("configureTopology");

            _brokerTopology.Probe(scope);
        }

        async Task ConfigureTopology(ModelContext context)
        {
            await Task.WhenAll(_brokerTopology.Queues.Select(queue => Declare(context, queue))).ConfigureAwait(false);

            await Task.WhenAll(_brokerTopology.Exchanges.Select(exchange => Declare(context, exchange))).ConfigureAwait(false);

            await Task.WhenAll(_brokerTopology.QueueBindings.Select(binding => Bind(context, binding))).ConfigureAwait(false);

            await Task.WhenAll(_brokerTopology.ExchangeBindings.Select(binding => Bind(context, binding))).ConfigureAwait(false);
        }

        static Task Declare(ModelContext context, Exchange exchange)
        {
            RabbitMqLogMessages.DeclareExchange(exchange);

            return context.ExchangeDeclare(exchange.ExchangeName, exchange.ExchangeType, exchange.Durable, exchange.AutoDelete, exchange.ExchangeArguments);
        }

        static async Task Declare(ModelContext context, Queue queue)
        {
            try
            {
                var ok = await context.QueueDeclare(queue.QueueName, queue.Durable, queue.Exclusive, queue.AutoDelete, queue.QueueArguments)
                    .ConfigureAwait(false);

                RabbitMqLogMessages.DeclareQueue(queue, ok.ConsumerCount, ok.MessageCount);

                await Task.Yield();
                await Task.Delay(10).ConfigureAwait(false);
            }
            catch (Exception exception)
            {
                LogContext.Error?.Log(exception, "Declare queue faulted: {Queue}", queue);

                throw;
            }
        }

        static async Task Bind(ModelContext context, ExchangeToExchangeBinding binding)
        {
            RabbitMqLogMessages.BindToExchange(binding);

            await context.ExchangeBind(binding.Destination.ExchangeName, binding.Source.ExchangeName, binding.RoutingKey, binding.Arguments)
                .ConfigureAwait(false);

            await Task.Yield();
            await Task.Delay(10).ConfigureAwait(false);
        }

        static async Task Bind(ModelContext context, ExchangeToQueueBinding binding)
        {
            RabbitMqLogMessages.BindToQueue(binding);

            await context.QueueBind(binding.Destination.QueueName, binding.Source.ExchangeName, binding.RoutingKey, binding.Arguments).ConfigureAwait(false);

            await Task.Yield();
            await Task.Delay(10).ConfigureAwait(false);
        }


        class Context :
            ConfigureTopologyContext<TSettings>
        {
        }
    }
}
{'fraction_non_alphanumeric': 0.08552547973058838, 'fraction_numerical': 0.0007624857033930614, 'mean_word_length': 2.5086937137761924, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5015479', 'n_tokens_mistral': 2074, 'n_tokens_neox': 1988, 'n_words': 390}