SmartTender.MessageBroker 1.0.6

SmartMessageBroker NuGet Package

Overview

This NuGet package provides services for working with RabbitMQ message broker within the internal affairs of Smarttender. It offers functionality for message publishing and consuming, allowing efficient communication between different parts of your system via RabbitMQ queues.

Installation

To use this package, you can install it via NuGet Package Manager or directly add it to your project's dependencies in the .csproj file.

dotnet add package Smarttender.MessageBroker

Usage

Dependency Injection methods in Smarttender.MessageBroker.AspNetCore package

public static class SmartMessageBrokerServicesDi
{
    // Adding required services to the service collection
    public static IServiceCollection AddSmartMessageBroker(this IServiceCollection services, string connectionString, Action<BrokerConfig> configureReceivers = null);

    // Starting consuming queues
    public static IApplicationBuilder UseSmartMessageBrokerConsumers(this IApplicationBuilder app);
}

DI Registration Examples

using SmartMessageBroker;
using Smarttender.MessageBroker.Producer.Consumers;

var builder = WebApplication.CreateBuilder(args);

// Add services to the container.

builder.Services.AddControllers();
builder.Services.AddSmartMessageBroker("amqps://your-rabbitmq-url",
	cfg =>
	{
		cfg.AddConsumer<NewTenderBidMessageConsumer, NewTenderBidMessage>();
		cfg.AddConsumer<FeatureUsingMessageConsumer, FeatureUsingMessage>();
	});
var app = builder.Build();

// Configure the HTTP request pipeline.

app.UseHttpsRedirection();
app.UseSmartMessageBrokerConsumers();

app.MapControllers();

app.Run();

Replace "amqps://your-rabbitmq-url" with your actual RabbitMQ connection string.

Publisher

Service that provides you ability to send message to particular queue

public interface IMessagePublisher
{
    void Publish<T>(T message, Queue queue)
        where T: BasicMessage;
}

Publisher usage

using Microsoft.AspNetCore.Mvc;
using Smarttender.MessageBroker.Abstracts;
using Smarttender.MessageBroker.Models;


[ApiController]
[Route("api/[controller]")]
public class MessageController : ControllerBase
{
	private readonly IMessagePublisher _messagePublisher;

	public MessageController(IMessagePublisher messagePublisher)
	{
		_messagePublisher = messagePublisher ?? throw new ArgumentNullException(nameof(messagePublisher));
	}
	[HttpPost("string")]
	public IActionResult ProduceStringData()
	{
		try
		{
			var fakeMessage = new StringDataMessage
			{
				Data = Faker.Name.FullName(),
				Edrpou = Faker.RandomNumber.Next(10000000, 99999999).ToString(),
				UserLogin = Faker.Internet.UserName(),
				Segment = Faker.Lorem.GetFirstWord(),
				FeatureRole = Faker.Lorem.GetFirstWord(),
				Datestamp = DateTime.Now,
				Tags = new string[] { Faker.Lorem.GetFirstWord(), Faker.Lorem.GetFirstWord() }
			};

			// Here you can send the mock message using your IMessagePublisher implementation
			_messagePublisher.Publish(fakeMessage, Queue.Common);

			Console.WriteLine($"Produced mock message: {fakeMessage.Data}");
			return Ok();
		}
		catch (Exception ex)
		{
			Console.WriteLine($"Error producing mock message: {ex.Message}");
			return StatusCode(500, "An error occurred while producing the mock message.");
		}
	}
	[HttpPost]
	public IActionResult Produce()
	{
		try
		{
			var fakeMessage = new FeatureUsingMessage
			{
				FeatureName = Faker.Name.FullName(),
				Edrpou = Faker.RandomNumber.Next(10000000, 99999999).ToString(),
				UserLogin = Faker.Internet.UserName(),
				Segment = Faker.Lorem.GetFirstWord(),
				FeatureRole = Faker.Lorem.GetFirstWord(),
				Datestamp = DateTime.Now,
				Tags = new string[] { Faker.Lorem.GetFirstWord(), Faker.Lorem.GetFirstWord() }
			};

			// Here you can send the mock message using your IMessagePublisher implementation
			_messagePublisher.Publish(fakeMessage, Queue.Common);

			Console.WriteLine($"Produced mock message: {fakeMessage.FeatureName}");
			return Ok();
		}
		catch (Exception ex)
		{
			Console.WriteLine($"Error producing mock message: {ex.Message}");
			return StatusCode(500, "An error occurred while producing the mock message.");
		}
	}
}

Consumer

Class that implements basic async consumers:

	public abstract class BasicAsyncConsumer<T> : IAsyncMessageConsumer<T>
		where T : BasicMessage
	{
		public virtual string Filter => typeof(T).Name;
		public virtual int Ttl => 4 * 60 * 60 * 1000; // базовий час життя 4 години
		public abstract Queue Queue { get; }
		public abstract Task<MessageDeliveryStatus> OnMessage(T message);
	}

and registered in DI like in "DI Registration Examples" blok shown is represents a consumer responsible for processing messages of type <T> synchronously/asynchronously. When a message of this type is received from the RabbitMQ queue, the OnMessage method is invoked.

the message delivery acknowledgment is decided by result of OnMessagge method, that returns MessageDeliveryStatus

	public enum MessageDeliveryStatus
	{
		Acknowledged = 0,
		NotAcknowledged = 1,
	}

If OnMessagge returns MessageDeliveryStatus.Acknowledged that means message will be considered as delivered successfuly otherwise if MessageDeliveryStatus.NotAcknowledged being returned, that means message is not accepted by consumer end requeued again

Consumer example

	public class FeatureUsingMessageConsumer : BasicAsyncConsumer<FeatureUsingMessage>
	{
		public override Task<MessageDeliveryStatus> OnMessage(FeatureUsingMessage message)
		{
			// Implement the logic to process the received message
			Console.WriteLine($"Received FeatureUsingMessage: {message.FeatureName}");
			return Task.FromResult(MessageDeliveryStatus.Acknowledged);
		}
	}

Target frameworks

net462;
netstandard2.0;
netstandard2.1

Dependencies

• RabbitMQ.Client

Routing Scheme Overview

Consumer Registration

1. Queue Declaration: For each consumer in the consumers collection, a queue is declared with a name constructed based on _queue, the executing assembly's name, and the consumer's message type.
2. Queue Binding: The declared queue is bound to a common exchange (Exchange.Common) with a routing key constructed from the consumer's message type and _queue.
3. Consumer Creation: An AsyncEventingBasicConsumer is created for each consumer, and its Received event is handled. When a message is received on the queue, the consumer's OnMessage method is invoked asynchronously.

Message Publication

1. Routing Key: When publishing a message, the routing key is constructed from the message's type and the target queue (queue). This routing key is used to determine how the message should be routed within RabbitMQ.
2. Exchange: The message is published to the common exchange (Exchange.Common). RabbitMQ uses the exchange to route messages to queues based on routing keys.
3. Message Body: The message is serialized to JSON and converted to a byte array before being published to the exchange.

Routing Scheme Details

• Exchange: A common exchange (Exchange.Common) is used for message routing.
• Queues: Each consumer has its own queue declared with a name constructed based on _queue, the executing assembly's name, and the consumer's message type.
• Routing Keys: Messages are published with routing keys constructed from the message type and the target queue. Consumers bind their queues to the exchange using routing keys based on their filter and _queue.