Sindika.AspNet.Connection 1.6.3

Sindika.AspNet.Connection

Sindika.AspNet.Connection is a .NET library designed to simplify and standardize connection management for various services such as Citus, Kafka, PostgreSQL, RabbitMQ, and Redis. It provides robust tools to configure, validate, and manage these connections efficiently.

Features

• Citus Connection Management:

  • Simplified connection setup for Citus DB.
  • Health checks to ensure availability and performance.
  • Configurable settings for database tuning.

• Kafka Integration:

  • Streamlined connection handling for Kafka brokers.
  • Support for producer and consumer setup.

• PostgreSQL Support:

  • Easy-to-configure connection for PostgreSQL databases.
  • Built-in health checks and validation.

• RabbitMQ Support:

  • Efficient connection management for RabbitMQ messaging.
  • Support for queues and message validation.

• Redis Integration:

  • Configurable connection settings for Redis caching and data storage.
  • Supports connection pooling and health checks.

Usage

Citus Connection

1. Configuration: Use CitusSettings.cs to define the necessary settings for connecting to the Citus database.

   var settings = new CitusSettings
   {
       Host = "localhost",
       Port = 5432,
       Database = "my_database",
       Username = "user",
       Password = "password"
   };
   

2. Connection Manager: Use CitusConnectionManager to establish and manage the connection.

   var manager = new CitusConnectionManager(settings);
   var connection = manager.GetConnection();
   

3. Health Check: Ensure the connection is healthy using CitusHealthChecker.

   var healthChecker = new CitusHealthChecker();
   bool isHealthy = healthChecker.CheckHealth(connection);
   

RabbitMQ

1. RabbitMQ Configuration in appsettings.json

To connect the application with RabbitMQ, add the following configuration in appsettings.json or appsettings.Development.json:

"RabbitmqSettings": {
    "HostName": "rabbitmq-host",
    "Port": 5672,
    "UserName": "rabbitmq-username",
    "Password": "rabbitmq-password"
}

Note:

• Replace rabbitmq-host, rabbitmq-username, rabbitmq-password with the appropriate RabbitMQ address.
• Do not store sensitive credentials in the code. Use environment variables or secret manager if needed.

2. Register RabbitMQ in Program.cs

Add the following code in Program.cs to configure RabbitMQ:

using Sindika.AspNet.Connection.Rabbitmq;

var builder = WebApplication.CreateBuilder(args);

var Configuration = builder.Configuration;
Configuration.AddEnvironmentVariables();

// Register RabbitMQ Extensions
builder.Services.AddRabbitMQ(Configuration);

3. Using RabbitMQ Service

3.1 Publishing Messages

Example of publishing messages to RabbitMQ:

public class OrderService
{
    private readonly RabbitmqService _rabbitmqService;

    public OrderService(RabbitmqService rabbitmqService)
    {
        _rabbitmqService = rabbitmqService;
    }

    public async Task SendMessageAsync()
    {
        var exchangeConfig = new RabbitmqExchangeConfig
        {
            Name = "order-exchange",
            Type = ExchangeType.Direct,
            Durable = true,
            AutoDelete = false
        };

        var routingKey = "order.created";
        var message = "New order placed";

        await _rabbitmqService.PublishMessageAsync(exchangeConfig, routingKey, message);
    }
}

3.2 Consuming Messages (Background Service)

For continuous message consumption, use a BackgroundService implementation:

public class OrderConsumerService : BackgroundService
{
    private readonly RabbitmqService _rabbitmqService;
    private readonly ILogger<OrderConsumerService> _logger;
    private string? _consumerTag;

    public OrderConsumerService(
        RabbitmqService rabbitmqService,
        ILogger<OrderConsumerService> logger)
    {
        _rabbitmqService = rabbitmqService;
        _logger = logger;
    }

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        try
        {
            var exchangeConfig = new RabbitmqExchangeConfig
            {
                Name = "order-exchange",
                Type = ExchangeType.Direct,
                Durable = true,
                AutoDelete = false
            };

            var queueConfig = new RabbitmqQueueConfig
            {
                Name = "order-queue",
                Durable = true,
                Exclusive = false,
                AutoDelete = false
            };

            var queueBindConfig = new RabbitmqQueueBindConfig
            {
                Queue = "order-queue",
                Exchange = "order-exchange",
                RoutingKey = "order.created"
            };

            // Start consuming messages
            _consumerTag = await _rabbitmqService.ConsumeMessageAsync(
                exchangeConfig,
                queueConfig,
                queueBindConfig,
                async (message) =>
                {
                    _logger.LogInformation("Received message: {Message}", message);

                    // Process your message here
                    await ProcessOrderAsync(message);
                },
                stoppingToken
            );

            _logger.LogInformation("Consumer started with tag: {ConsumerTag}", _consumerTag);

            // Keep the service running
            await Task.Delay(Timeout.Infinite, stoppingToken);
        }
        catch (OperationCanceledException)
        {
            _logger.LogInformation("Consumer service is stopping...");
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "Error in consumer service");
            throw;
        }
    }

    public override async Task StopAsync(CancellationToken cancellationToken)
    {
        if (!string.IsNullOrEmpty(_consumerTag))
        {
            await _rabbitmqService.StopConsumingAsync(_consumerTag, cancellationToken);
            _logger.LogInformation("Consumer stopped");
        }

        await base.StopAsync(cancellationToken);
    }

    private async Task ProcessOrderAsync(string message)
    {
        // Your business logic here
        _logger.LogInformation("Processing order: {Message}", message);
        await Task.CompletedTask;
    }
}

Register the background service in Program.cs:

builder.Services.AddHostedService<OrderConsumerService>();

3.3 Getting Single Message (Pull-based)

For scenarios where you need to fetch a single message on-demand:

public class OrderProcessingService
{
    private readonly RabbitmqService _rabbitmqService;

    public OrderProcessingService(RabbitmqService rabbitmqService)
    {
        _rabbitmqService = rabbitmqService;
    }

    public async Task<string> GetNextOrderAsync()
    {
        var exchangeConfig = new RabbitmqExchangeConfig
        {
            Name = "order-exchange",
            Type = ExchangeType.Direct,
            Durable = true
        };

        var queueConfig = new RabbitmqQueueConfig
        {
            Name = "order-queue",
            Durable = true,
            Exclusive = false,
            AutoDelete = false
        };

        var queueBindConfig = new RabbitmqQueueBindConfig
        {
            Queue = "order-queue",
            Exchange = "order-exchange",
            RoutingKey = "order.created"
        };

        // Get single message (returns empty string if no message available)
        var message = await _rabbitmqService.GetMessageAsync(
            exchangeConfig,
            queueConfig,
            queueBindConfig,
            autoAck: false // Set to true for auto-acknowledge
        );

        return message;
    }
}

3.4 Queue Management

Delete a queue when needed:

public async Task DeleteOrderQueueAsync()
{
    await _rabbitmqService.DeleteQueueAsync("order-queue");
}

4. RabbitMQ Configuration Models

RabbitmqExchangeConfig

public class RabbitmqExchangeConfig
{
    public string Name { get; set; }           // Exchange name
    public string Type { get; set; }           // Exchange type: "direct", "fanout", "topic", "headers"
    public bool Durable { get; set; }          // Survive broker restart
    public bool AutoDelete { get; set; }       // Delete when no longer in use
    public IDictionary<string, object>? Arguments { get; set; }
    public bool NoWait { get; set; }
    public CancellationToken CancellationToken { get; set; }
}

RabbitmqQueueConfig

public class RabbitmqQueueConfig
{
    public string Name { get; set; }           // Queue name
    public bool Durable { get; set; }          // Survive broker restart
    public bool Exclusive { get; set; }        // Used by only one connection
    public bool AutoDelete { get; set; }       // Delete when consumer cancels
    public IDictionary<string, object>? Arguments { get; set; }
    public bool NoWait { get; set; }
    public CancellationToken CancellationToken { get; set; }
}

RabbitmqQueueBindConfig

public class RabbitmqQueueBindConfig
{
    public string Queue { get; set; }          // Queue name to bind
    public string Exchange { get; set; }       // Exchange name to bind to
    public string RoutingKey { get; set; }     // Routing key for binding
    public IDictionary<string, object>? Arguments { get; set; }
    public bool NoWait { get; set; }
    public CancellationToken CancellationToken { get; set; }
}

5. Best Practices

• Use BackgroundService for Consumers: Always implement consumers as BackgroundService or IHostedService for proper lifecycle management
• Handle Cancellation Tokens: Pass CancellationToken to enable graceful shutdown
• Error Handling: Consumer automatically handles errors with Nack and requeue mechanism
• Auto-Recovery: Connection includes automatic recovery enabled by default
• Durable Queues: Use Durable = true for queues that should survive broker restarts
• Manual Acknowledgment: Use autoAck: false and manually acknowledge messages after processing to ensure reliability

Future Plans

• Add implementation for Kafka connection management.
• Extend PostgreSQL, RabbitMQ, and Redis connection support.
• Provide detailed examples for each service.
• Include performance optimization tools.