JN.RabbitMQClient 2.3.0

Simple implementation of RabbitMQ consumer and sender.

Install-Package JN.RabbitMQClient -Version 2.3.0
dotnet add package JN.RabbitMQClient --version 2.3.0
<PackageReference Include="JN.RabbitMQClient" Version="2.3.0" />
For projects that support PackageReference, copy this XML node into the project file to reference the package.
paket add JN.RabbitMQClient --version 2.3.0
The NuGet Team does not provide support for this client. Please contact its maintainers for support.

JN.RabbitMQClient

Simple implementation of RabbitMQ consumer and sender.

Features

  • Sender implementation
  • Multiple consumer instances supported
  • Multiple processing options for received messages
  • Random expiration for messages sent to an holding queue (depending on the processing option)
  • TLS connection support
  • Limiter for processing messages

More details available on the project website.

Install

Download the package from NuGet:

Install-Package JN.RabbitMQClient -Version [version number]

Usage

First, you must create the RabbitMqConsumerService and then define delegates for ReceiveMessage, ShutdownConsumer and ReceiveMessageError. The service will start the required number of consumers when StartConsumers is called.

To use a retry queue, the method StartConsumers should be called with a RetryQueueDetails object.

Message processing instructions

OK - message is considered as successfully processed

RequeueMessageWithDelay - message is removed from the queue, but sent to a holding queue for later processing (typically with a dead letter configuration)

IgnoreMessage - message is removed from the queue and ignored

IgnoreMessageWithRequeue - message is rejected and sent back to the queue

Example

Example for consumer and sender services:

    class Program
    {
        static void Main(string[] args)
        {
            // consumer

            var consumerService = new RabbitMqConsumerService(GetBrokerConfigConsumers());

            consumerService.ReceiveMessage += ReceiveMessage;
            consumerService.ShutdownConsumer += ShutdownConsumer;
            consumerService.ReceiveMessageError += ReceiveMessageError;

            consumerService.StartConsumers("my consumer");

            // sender

            var senderService = new RabbitMqSenderService(GetBrokerConfigSender());

            senderService.Send("my message");

            Console.WriteLine("Press any key to exit...");
            Console.ReadKey();

            consumerService.Dispose();
        }

        private static IBrokerConfigSender GetBrokerConfigSender()
        {
            IBrokerConfigSender configSender = new BrokerConfigSender()
            {
                Username = "test",
                Password = "123",
                Host = "localhost",
                VirtualHost = "/",
                RoutingKeyOrQueueName = "MyTestQueue"
            };
            return configSender;
        }

        private static IBrokerConfigConsumers GetBrokerConfigConsumers()
        {
            IBrokerConfigConsumers configConsumers = new BrokerConfigConsumers()
            {
                Username = "test",
                Password = "123",
                Host = "localhost",
                VirtualHost = "/",
                RoutingKeyOrQueueName = "MyTestQueue",
                ShuffleHostList = false,
                Port = 0,
                TotalInstances = 3
            };
            return configConsumers;
        }

        private static async Task ReceiveMessageError(string routingKeyOrQueueName, string consumerTag, string exchange, string message, string errorMessage)
        {
            await Console.Out.WriteLineAsync($"Error: '{consumerTag}' | {errorMessage}");
        }

        private static async Task ShutdownConsumer(string consumerTag, ushort errorCode, string shutdownInitiator, string errorMessage)
        {
            await Console.Out.WriteLineAsync($"Shutdown '{consumerTag}' | {errorCode} | {shutdownInitiator} | {errorMessage}");
        }

        private static async Task<Constants.MessageProcessInstruction> ReceiveMessage(string routingKeyOrQueueName, string consumerTag, long firstErrorTimestamp, string exchange, string message)
        {
            await Console.Out.WriteLineAsync($"Message received from '{consumerTag}': {message}");
            return Constants.MessageProcessInstruction.OK;
        }
    }

JN.RabbitMQClient

Simple implementation of RabbitMQ consumer and sender.

Features

  • Sender implementation
  • Multiple consumer instances supported
  • Multiple processing options for received messages
  • Random expiration for messages sent to an holding queue (depending on the processing option)
  • TLS connection support
  • Limiter for processing messages

More details available on the project website.

Install

Download the package from NuGet:

Install-Package JN.RabbitMQClient -Version [version number]

Usage

First, you must create the RabbitMqConsumerService and then define delegates for ReceiveMessage, ShutdownConsumer and ReceiveMessageError. The service will start the required number of consumers when StartConsumers is called.

To use a retry queue, the method StartConsumers should be called with a RetryQueueDetails object.

Message processing instructions

OK - message is considered as successfully processed

RequeueMessageWithDelay - message is removed from the queue, but sent to a holding queue for later processing (typically with a dead letter configuration)

IgnoreMessage - message is removed from the queue and ignored

IgnoreMessageWithRequeue - message is rejected and sent back to the queue

Example

Example for consumer and sender services:

    class Program
    {
        static void Main(string[] args)
        {
            // consumer

            var consumerService = new RabbitMqConsumerService(GetBrokerConfigConsumers());

            consumerService.ReceiveMessage += ReceiveMessage;
            consumerService.ShutdownConsumer += ShutdownConsumer;
            consumerService.ReceiveMessageError += ReceiveMessageError;

            consumerService.StartConsumers("my consumer");

            // sender

            var senderService = new RabbitMqSenderService(GetBrokerConfigSender());

            senderService.Send("my message");

            Console.WriteLine("Press any key to exit...");
            Console.ReadKey();

            consumerService.Dispose();
        }

        private static IBrokerConfigSender GetBrokerConfigSender()
        {
            IBrokerConfigSender configSender = new BrokerConfigSender()
            {
                Username = "test",
                Password = "123",
                Host = "localhost",
                VirtualHost = "/",
                RoutingKeyOrQueueName = "MyTestQueue"
            };
            return configSender;
        }

        private static IBrokerConfigConsumers GetBrokerConfigConsumers()
        {
            IBrokerConfigConsumers configConsumers = new BrokerConfigConsumers()
            {
                Username = "test",
                Password = "123",
                Host = "localhost",
                VirtualHost = "/",
                RoutingKeyOrQueueName = "MyTestQueue",
                ShuffleHostList = false,
                Port = 0,
                TotalInstances = 3
            };
            return configConsumers;
        }

        private static async Task ReceiveMessageError(string routingKeyOrQueueName, string consumerTag, string exchange, string message, string errorMessage)
        {
            await Console.Out.WriteLineAsync($"Error: '{consumerTag}' | {errorMessage}");
        }

        private static async Task ShutdownConsumer(string consumerTag, ushort errorCode, string shutdownInitiator, string errorMessage)
        {
            await Console.Out.WriteLineAsync($"Shutdown '{consumerTag}' | {errorCode} | {shutdownInitiator} | {errorMessage}");
        }

        private static async Task<Constants.MessageProcessInstruction> ReceiveMessage(string routingKeyOrQueueName, string consumerTag, long firstErrorTimestamp, string exchange, string message)
        {
            await Console.Out.WriteLineAsync($"Message received from '{consumerTag}': {message}");
            return Constants.MessageProcessInstruction.OK;
        }
    }

Release Notes


     Release Notes for current version:

     - Update RabbitMQ.Client to latest version
     - Changed namespace for IRabbitMqConsumerService and IRabbitMqSenderService
     - Changed behavior for StopConsumers(consumerTag) - now stops all consumers with tag starting with 'consumerTag'
     - Added limiter feature for processing messages.

   

NuGet packages

This package is not used by any NuGet packages.

GitHub repositories

This package is not used by any popular GitHub repositories.

Version History

Version Downloads Last updated
2.3.0 157 9/4/2020
2.2.1 181 6/6/2020
2.2.0-beta.1 118 5/24/2020
2.2.0-beta 66 5/15/2020
2.1.1 107 4/14/2020
2.1.0 127 12/19/2019
2.0.0 119 11/16/2019
1.0.1 145 6/30/2019
1.0.0 239 4/23/2019