C# Rabbitmq single consumer consume multiple message in single Queue

Viewed 20

I have single queue with more than message and i want to consume that message sequencial in

order for another transaction with parse payload foreach message

the problem is when i loop all payload list after get the first paylod why first message

is always still count on list after being loop and always being proceed

and the second message and so on not being consumed

my sample list messsage array is "

first message : ["test123@gmail.com","smtp.email.io","admin","2525","123","Test_200999@yahoo.com","Email Confirmation","Hello World"]

second message :

["test123@gmail.com","smtp.email.io","admin","2525","123","Test_200998@yahoo.com","Email Confirmation","Hello World"]

third message

["test123@gmail.com","smtp.email.io","admin","2525","123","Test_200997@yahoo.com","Email Confirmation","Hello World"]

here's my code

private static async void ExcecutePayload(CancellationToken cancellationToken)
{
  var factory = new ConnectionFactory
        {
            HostName = "localhost"
        };

        var connection = factory.CreateConnection();

        using var channel = connection.CreateModel();

        channel.QueueDeclare("email_confirmation_notifications", durable: true, exclusive: false, autoDelete: false,arguments: null);

        channel.BasicQos(0, 1, false);

        var consumer = new EventingBasicConsumer(channel);

        EmailModel _model = new EmailModel();

        var s1 = new List<string>();

        string s2 = "";

        bool
            first = false,
            second = false;

        var random = new Random();

        consumer.Received += (model, eventArgs) =>
        {
            var processingtime = random.Next(1, 10);

            var body = eventArgs.Body.ToArray();

            var message = Encoding.UTF8.GetString(body);

            foreach (char c in message)
            {
                if (second)
                {
                    s1.Add(s2);
                    s2 = "";
                    first = false;
                    second = false;
                }
                if (first)
                {
                    if (c == '"')
                    {
                        second = true;
                    }
                    else
                    {
                        s2 += c;
                    }
                }
                if (c == '"' && !first && !second)
                {
                    first = true;
                }
            }

            if (second && first)
            {
                s1.Add(s2);
            }

            foreach (var item in s1.Select((value, i) => new { i, value }).ToList())
            {
                var value = item.value;
                var index = item.i;

                if (index == 0) { _model.Sender = value; };
                if (index == 1) { _model.Host = value; };
                if (index == 2) { _model.UserName = value; };
                if (index == 3) { _model.Port = value; };
                if (index == 4) { _model.Password = value; };
                if (index == 5) { _model.To = value; };
                if (index == 6) { _model.Subject = value; };
                if (index == 7) { _model.Body = value; };
            }

            try
            {
                InsertTransaction(_model, cancellationToken);
                SendEmailAsync(_model, cancellationToken);
            }
            catch (Exception ex)
            {
                throw ex;
            }

            Console.WriteLine($"Notification email has sent to: {string.Join(", ", _model.To)}");

            Task.Delay(TimeSpan.FromSeconds(processingtime), cancellationToken).Wait();

            channel.BasicAck(deliveryTag: eventArgs.DeliveryTag, multiple: false);
        };

        await Task.CompletedTask;

        channel.BasicConsume(queue: "email_confirmation_notifications", autoAck: false, consumer: consumer);

        Console.ReadKey();
}
0 Answers
Related