SmartTender.MessageBroker 2.0.5

SmartMessageBroker NuGet Package

Installation

To install the SmartMessageBroker NuGet package, run the following command in the Package Manager Console:

Install-Package SmartMessageBroker

Dependency Injection Setup

.NET Framework using Ninject

1. Install the Ninject package if not already installed:

Install-Package Ninject

2. Register the SmartMessageBroker services in your Ninject module:

using Ninject;
using Ninject.Modules;
using SmartTender.MessageBroker;

public class DiModule : NinjectModule
{
    public override void Load()
    {
        this.AddSmartMessageBroker(config =>
        {
            // Configure your BrokerConfig here
        });
    }
}

3. Use the SmartMessageBroker consumers in your application:

using Ninject;
using SmartTender.MessageBroker;

public class Program
{
    public static void Main()
    {
        var kernel = new StandardKernel(new DiModule());
        kernel.UseSmartMessageBrokerConsumers();
    }
}

Portable using IServiceCollection

1. Register the SmartMessageBroker services in your Startup.cs:

using Microsoft.Extensions.DependencyInjection;
using SmartTender.MessageBroker;

public class Startup
{
    public void ConfigureServices(IServiceCollection services)
    {
        services.AddSmartMessageBroker(config =>
        {
            // Configure your BrokerConfig here
        });
    }

    public void Configure(IApplicationBuilder app, IHostingEnvironment env)
    {
        // Other configurations...
    }
}

Static Containers using Ninject (Used for IT-Enterprise)

1. Register the SmartMessageBroker services in your Ninject module:

using Ninject;
using Ninject.Modules;
using SmartTender.MessageBroker;

public class DiModule : NinjectModule
{
    public override void Load()
    {
        this.AddSmartMessageBroker(config =>
        {
            // Configure your BrokerConfig here
        });
    }
}

2. Use the SmartMessageBroker publisher and consumers in your application:

using Ninject;
using SmartTender.MessageBroker.ITEnterprise;

public class Program
{
    public static void Main()
    {
        var kernel = new StandardKernel(new DiModule());

        using (var publisher = new SmartMessageBrokerPublisher(config =>
        {
            // Configure your BrokerConfig here
        }))
        {
            // Use publisher to publish messages
        }

        using (var dispatcher = new SmartMessageBrokerConsumersDispatcher(config =>
        {
            // Configure your BrokerConfig here
        }))
        {
            dispatcher.StartConsuming();
        }
    }
}

Available Configurations

When configuring the SmartMessageBroker, you can set the following options in the BrokerConfig:

• SendTimeout: TimeSpan? - The timeout for sending messages.
• RecieveTimeout: TimeSpan? - The timeout for receiving messages.
• ProducerConnectionString: string - The connection string for the message producer. Required: If not set, there will be a runtime exception.
• ConsumerConnectionString: string - The connection string for the message consumer. Required: If not set, there will be a runtime exception.
• ApplicationName: string - The name of the application. Important: This acts as the scope of your queues and is used to differentiate different apps in one broker server.
• Mode: QueueListeningMode - The mode for queue listening. Important: This determines how queues are listened to, either OnePerApp or OnePerInstance.
• ConnectRetriesCount: int - The number of retries for connection attempts.
• ConnectRetryDelayMs: int - The delay in milliseconds between connection retries.

Logging Functions

• LogExceptionsTo(Action<Exception, string> action): Sets the action to log exceptions.
• LogTo(Action<string> action): Sets the action to log general messages.

Service Lifetimes

• Singleton: Main services are registered as singleton and will have a single instance throughout the application's lifetime.
• Scoped: Consumers are registered as scoped and will have a single instance per request or scope. Each message from the broker will be processed in a separate scope.

Consumer Registration Example

To register a consumer, use the AddConsumer method:

config.AddConsumer<MyConsumer, MyMessage>();

Profile Registration Example

To register profiles, use the AddFromProfile or AddFromProfiles methods:

config.AddFromProfile(new MyBrokerProfile());
config.AddFromProfiles();

Example configuration:

services.AddSmartMessageBroker(config =>
{
    config.SendTimeout = TimeSpan.FromSeconds(30);
    config.RecieveTimeout = TimeSpan.FromSeconds(30);
    config.ProducerConnectionString = "your-producer-connection-string"; // Required
    config.ConsumerConnectionString = "your-consumer-connection-string"; // Required
    config.ApplicationName = "YourApplicationName"; // Important
    config.Mode = QueueListeningMode.OnePerApp; // Important
    config.ConnectRetriesCount = 3;
    config.ConnectRetryDelayMs = 2000;
    config.LogExceptionsTo((ex, context) => 
    {
        // Log exception
    });
    config.LogTo(message => 
    {
        // Log debug message
    });
    config.AddConsumer<MyConsumer, MyMessage>(); // Register consumer
    config.AddFromProfile(new MyBrokerProfile()); // Register profile
});

Message Publication

To publish messages, use the IMessagePublisher interface. Here is an example:

using SmartTender.MessageBroker.Models; // Import the Queue enum

public class MyService
{
    private readonly IMessagePublisher _publisher;

    public MyService(IMessagePublisher publisher)
    {
        _publisher = publisher;
    }

    public async Task PublishMessageAsync()
    {
        var message = new MyMessage
        {
            // Set message properties
        };
        var queue = Queue.Common; // Use the Queue enum

        await _publisher.PublishAsync(message, queue);
    }
}

public enum Queue
{
	Common,
	Commercial,
	ProzorroSale,
	ProzorroPurchase,
	Signy,
}

Message Consuming

To implement a consumer, create a class that inherits from BasicAsyncConsumer<T>, where T is the type of message the consumer will handle. Override the OnMessage method to define the logic for processing the message.

Here is an example:

using SmartTender.MessageBroker.Abstracts;
using SmartTender.MessageBroker.Models;
using System;
using System.Threading.Tasks;

public class MyMessageConsumer : BasicAsyncConsumer<MyMessage>
{
    public override Queue Queue => Queue.Common;

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

To register the consumer, use the AddConsumer method in the configuration:

services.AddSmartMessageBroker(config =>
{
    // ...existing configuration...
    config.AddConsumer<MyMessageConsumer, MyMessage>();
});

This will ensure that the MyMessageConsumer is registered and ready to consume messages of type MyMessage from the specified queue.

BasicAsyncConsumer

The BasicAsyncConsumer<T> class is an abstract base class for implementing message consumers. It provides several properties and methods that can be overridden to customize the behavior of the consumer.

Properties

• Queue: Specifies the queue from which the consumer will receive messages. This property must be overridden.
• Filter: Specifies a filter for the messages. The default value is the name of the message type. Based on this field inner routing of messages is working. Route is strign key that consist from Filter and Queue.
• Ttl: Specifies the time-to-live for the messages. The default value is 4 hours.
• Mode: Specifies the queue listening mode. The default value is QueueListeningMode.OnePerApp. QueueListeningMode.OnePerInstance - used to create queue per one instance of application, this queue will be deleted in process of graceful shutdown of app.
• RetryCount: Specifies the number of retry attempts for message processing. The default value is 5.
• AckOnError: Specifies the acknowledgment status when an error occurs. The default value is MessageDeliveryStatus.NotAcknowledgedRequeue.

Methods

• OnMessage(T message): This method must be overridden to define the logic for processing the received message.
• BeforeMessageConsuming(MessageConsumingStartedEventArgs<T> args): This method can be overridden to define logic that runs before message consuming starts.
• AfterMessageConsumed(MessageConsumingEndedEventArgs<T> args): This method can be overridden to define logic that runs after message consuming ends.

Here is an example of a consumer implementation:

using SmartTender.MessageBroker.Abstracts;
using SmartTender.MessageBroker.Models;
using System;
using System.Threading.Tasks;

public class MyMessageConsumer : BasicAsyncConsumer<MyMessage>
{
    public override Queue Queue => Queue.Common;

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

To register the consumer, use the AddConsumer method in the configuration:

services.AddSmartMessageBroker(config =>
{
    // ...existing configuration...
    config.AddConsumer<MyMessageConsumer, MyMessage>();
});

This will ensure that the MyMessageConsumer is registered and ready to consume messages of type MyMessage from the specified queue.

How Consuming Works

When a message is published to a queue, the SmartMessageBroker framework routes the message to the appropriate consumer based on the queue and filter settings. The consumer then processes the message and returns a MessageDeliveryStatus indicating the result of the processing.

MessageDeliveryStatus

The MessageDeliveryStatus enum defines the possible outcomes of message processing:

• Acknowledged: The message was successfully processed and can be removed from the queue.
• NotAcknowledgedRequeue: The message was not processed successfully and should be requeued for another attempt.
• NotAcknowledged: The message was not processed successfully and should be silently dropped.

Here is an example of how to handle different MessageDeliveryStatus values in a consumer:

using SmartTender.MessageBroker.Abstracts;
using SmartTender.MessageBroker.Models;
using System;
using System.Threading.Tasks;

public class MyMessageConsumer : BasicAsyncConsumer<MyMessage>
{
    public override Queue Queue => Queue.Common;

    public override async Task<MessageDeliveryStatus> OnMessage(MyMessage message)
    {
        try
        {
            // Implement the logic to process the received message
            Console.WriteLine($"Received MyMessage: {message.SomeProperty}");

            // Simulate message processing
            bool isProcessed = ProcessMessage(message);

            if (isProcessed)
            {
                return MessageDeliveryStatus.Acknowledged;
            }
            else
            {
                return MessageDeliveryStatus.NotAcknowledgedRequeue;
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"Error processing message: {ex.Message}");
            return MessageDeliveryStatus.NotAcknowledged;
        }
    }

    private bool ProcessMessage(MyMessage message)
    {
        // Simulate message processing logic
        return true; // Return true if processed successfully, false otherwise
    }
}

In this example, the OnMessage method processes the message and returns the appropriate MessageDeliveryStatus based on the outcome of the processing. If an exception occurs, the message is moved to the dead-letter queue.

To register the consumer, use the AddConsumer method in the configuration:

services.AddSmartMessageBroker(config =>
{
    // ...existing configuration...
    config.AddConsumer<MyMessageConsumer, MyMessage>();
});

This will ensure that the MyMessageConsumer is registered and ready to consume messages of type MyMessage from the specified queue.

Internals of Consuming

When a message is received by the consumer, it goes through several steps to ensure proper handling and logging:

1. Deserialization: The message is deserialized using the configured serializer.
2. Language Handling: The message's language is set based on the message properties.
3. BeforeMessageConsuming: The user defined consumer BeforeMessageConsuming method is called to perform any pre-processing logic.
4. OnMessage: The user defined consumer OnMessage method is called to process the message and return a MessageDeliveryStatus.
5. AfterMessageConsumed: The user defined consumer AfterMessageConsumed method is called to perform any post-processing logic.
6. Logging: The start and end of message processing are logged.

Handling MessageDeliveryStatus

The MessageDeliveryStatus returned from the OnMessage method determines the next steps:

• Acknowledged: The message is acknowledged and removed from the queue.
• NotAcknowledgedRequeue: The message is requeued for another attempt. The retry count is incremented, and an exponential backoff delay is applied.
• NotAcknowledged: The message is silently dropped.

Here is a high-level description of how the internals of consumer handles the MessageDeliveryStatus:

1. Acknowledged: The message is acknowledged using the BasicAck method, which removes it from the queue.
2. NotAcknowledged: The message is negatively acknowledged using the BasicNack method without requeueing, effectively dropping the message.
3. NotAcknowledgedRequeue: The message is negatively acknowledged using the BasicNack method with requeueing. The retry count is incremented, and an exponential backoff delay is applied before re-publishing the message to the retry queue.

Logging and Error Handling

The consumers internals also includes logging and error handling mechanisms to ensure that any issues during message processing are properly logged and handled.

Here is a high-level description of how the exception handling method works:

1. Deserialization: The message is deserialized to obtain the original message object.
2. AfterMessageConsumed: The AfterMessageConsumed method is called to perform any necessary cleanup.
3. Logging: The exception is logged using the LogEnd method, which records the details of the exception and the message processing status.

By following these steps, the consumers internals methods ensures that messages are properly processed, logged, and handled, even in the case of errors or exceptions.