Cubivue.EventBus 1.0.1

Event Bus Implementation , Its generic nuget package that handle Topic ,Queues and Subscription automatically.

#Create Rules dynamically

//Get Azure Settings from App Settings and assign it to Class , you can inject in program.cs

Need to add this section in your appsettings

"AzureServiceBusSettings": {
"ConnectionString": "your ServiceBusConnection",
"SubscriptionClientName": "Your Topic subscription if you are using topics , not needed in case of queue",
"TopicOrQueueName": " your Topic or Queue name",
"IsTopic":  true for topic use and false fro queue use

}

You need to create AzureServiceBusConfiguration in your own project and AzureServiceBusConfigurationValidation for to validate settings , all are mandatory in our case

 public class AzureServiceBusConfiguration : IAzureServiceBusConfiguration
{
    public string ConnectionString { get; set; }
    public string SubscriptionClientName { get; set; }
    public string TopicOrQueueName {get; set;  }
    public bool IsTopic {  get; set; }
}

public class AzureServiceBusConfigurationValidation : IValidateOptions<AzureServiceBusConfiguration>
{
    public ValidateOptionsResult Validate(string name, AzureServiceBusConfiguration options)
    {
        if (string.IsNullOrEmpty(options.ConnectionString))
        {
            return ValidateOptionsResult.Fail($"{nameof(options.ConnectionString)} configuration parameter for the Azure Service Bus is required");
        }

        if (string.IsNullOrEmpty(options.SubscriptionClientName))
        {
            return ValidateOptionsResult.Fail($"{nameof(options.SubscriptionClientName)} configuration parameter for the Azure Service Bus is required");
        }

        return ValidateOptionsResult.Success;
    }
}

builder.Services.Configure<AzureServiceBusConfiguration>(builder.Configuration.GetSection("AzureServiceBusSettings"));
builder.Services.AddSingleton<IValidateOptions<AzureServiceBusConfiguration>, AzureServiceBusConfigurationValidation>();
builder.Services.AddSingleton<IAzureServiceBusConfiguration, AzureServiceBusConfiguration>(sp =>
{
    var azureServiceBusConfiguration = sp.GetRequiredService<IOptions<AzureServiceBusConfiguration>>().Value;
    return azureServiceBusConfiguration;
});


// We are mamanging subscriptionions in our memory , its contains method of subscribe , unsubscribe and many more. if we add any subscription in or code it will
automatically add rules to services bus topic and on unsubscribe remove its rules from topic

builder.Services.AddSingleton<IEventBusSubscriptionsManager, InMemoryEventBusSubscriptionsManager>();

// Subscriber Handlers, That will handle your events , it will recieve its own specific messages in case of topics and in case of queues this would manage all message
builder.Services.AddTransient<IIntegrationEventHandler<YourIntegrationEvent>, YourIntegrationEventHandler>();

all events will be ingerited from integration event , you need to create events in your own project that will inherit from event bus integration event

  public class YourIntegrationEvent : IntegrationEvent
{
    public long UserId { get; private set; }

    public YourIntegrationEvent(long userId)
    {
        UserId = userId;
    }
}

Handler will inherit from IIntegrationEventHandler<T>

public class YourIntegrationEventHandler : IIntegrationEventHandler<YourIntegrationEvent>
{
    private readonly ILogger<YourIntegrationEventHandler> _logger;

    public YourIntegrationEventHandler(ILogger<YourIntegrationEventHandler> logger)
    {
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
    }

    public async Task HandleAsync(YourIntegrationEvent @event)
    {
        _logger.LogInformation("Handling integration event: {IntegrationEventId} - ({@IntegrationEvent})", @event.Id, @event);

    }
}


// This is responsible for managing all connections with your Azure clud topics , queues , creating it sender 
it take Ilogger of microsoft extension and IAzureServiceBusConfiguration in its constructor which we have created and injected above

 public ServiceBusConnectionManagementService(ILogger<ServiceBusConnectionManagementService> logger, IAzureServiceBusConfiguration azureServiceBusConfiguration)

 you need to inject this in your code
builder.Services.AddSingleton<IServiceBusConnectionManagementService, ServiceBusConnectionManagementService>();



// This is main Event Bus code which is responsible to publish and subscribe events , it contain multiple methods. By default it will create topic , queue and its subscription
if not already exist. 


 public AzureServiceBusEventBus(IServiceBusConnectionManagementService serviceBusConnectionManagementService, IEventBusSubscriptionsManager subscriptionManager,
 IServiceProvider serviceProvider, ILogger<AzureServiceBusEventBus> logger)

 Add following code in your project , in below example one subscriber of topic is also listnening to its events , and co-relation rule will be automatically created accordingly 
 in case of queue it will treated as only handler.

 Setup method is to register subscription handlers , it needs if we have any subscription to handle 

 If we only need to handle publish events then we dont need to have setup Async method and Subscribe Aysnc methods.
   

builder.Services.AddSingleton<IEventBus, AzureServiceBusEventBus>(sp =>
{
    var serviceBusConnectionManagementService = sp.GetRequiredService<IServiceBusConnectionManagementService>();
    var logger = sp.GetRequiredService<ILogger<AzureServiceBusEventBus>>();
    var eventBusSubcriptionsManager = sp.GetRequiredService<IEventBusSubscriptionsManager>();

    var eventBus = new AzureServiceBusEventBus(serviceBusConnectionManagementService, eventBusSubcriptionsManager, sp, logger);
    eventBus.SetupAsync().GetAwaiter().GetResult();

    eventBus.SubscribeAsync<YourIntegrationEvent, IIntegrationEventHandler<YourIntegrationEvent>>()
                            .GetAwaiter().GetResult();

    return eventBus;
});

Now you are ready to use Cubivue Event Bus 
use event bus publish method to publish any event , all other stuff will be handled automatically , if subscriber is also registered in same project then it event will
hit your project handler as we have YourIntegrationEvent in our example.

Below code and approach is not mandatory , event bus is already setup for you, 

For to use this solution with logging in database for all your publish events and Then you also need to add our Cubivue.EventLog Nuget
Then use event publishing event in hybrid way , its description is given below

Create interface in your project

public interface IYourPortalIntegrationEventService
{
    Task PublishEventsThroughEventBusAsync(IntegrationEvent @event);
    Task AddAndSaveEventAsync(IntegrationEvent @event);
}

Implement it in your Application like

public class YourPortalIntegrationEventService : IYourPortalIntegrationEventService
{
    private readonly ApplicationDbContext _portalDbContext;
    private readonly IEventBus _eventBus;
    private readonly IEventLogService _eventLogService;
    private readonly ILogger<YourPortalIntegrationEventService> _logger;
    private readonly Func<DbConnection, IEventLogService> _integrationEventLogServiceFactory;

    public YourPortalIntegrationEventService(ApplicationDbContext portalDbContext, Func<DbConnection, IEventLogService> integrationEventLogServiceFactory,
                                 IEventBus eventBus,
                                 ILogger<YourPortalIntegrationEventService> logger)
    {
        _portalDbContext = portalDbContext ?? throw new ArgumentNullException(nameof(portalDbContext));
        _integrationEventLogServiceFactory = integrationEventLogServiceFactory ?? throw new ArgumentNullException(nameof(integrationEventLogServiceFactory));
        _eventBus = eventBus ?? throw new ArgumentNullException(nameof(eventBus));
        _eventLogService = _integrationEventLogServiceFactory(_portalDbContext.Database.GetDbConnection());
        _logger = logger ?? throw new ArgumentNullException(nameof(logger));
    }

    public async Task AddAndSaveEventAsync(IntegrationEvent @event)
    {
        await ResilientTransaction.CreateNew(_portalDbContext).ExecuteAsync(async () =>
        {
            await _portalDbContext.SaveChangesAsync();
            await _eventLogService.SaveEventAsync(@event, _portalDbContext.Database.CurrentTransaction);
        });
    }

    public async Task PublishEventsThroughEventBusAsync(IntegrationEvent @event)
    {
        try
        {
            await _eventLogService.MarkEventAsInProgressAsync(@event.Id);
            await _eventBus.PublishAsync(@event);
            await _eventLogService.MarkEventAsPublishedAsync(@event.Id);
        }
        catch (Exception ex)
        {
            _logger.LogError(ex, "ERROR publishing integration event: '{IntegrationEventId}'", @event.Id);

            await _eventLogService.MarkEventAsFailedAsync(@event.Id);
        }
    }
}




from you handler or any where else call like that, Inject above interface in your class or handler IYourPortalIntegrationEventService


   var yourIntegrationEvent = new YourIntegrationEvent(request.UserId);
    await _carrierPortalIntegrationEventService.AddAndSaveEventAsync(yourIntegrationEvent);
    await _carrierPortalIntegrationEventService.PublishEventsThroughEventBusAsync(yourIntegrationEvent);


    Reolve its dependency in you program.cs class

    Add eventdbcontext  , and put its migration in your actual Application Db context project . You need create and run migration for Event Log Db, in same project where 
    your existing DB migration exists. kindly make sure to create separate folder for EventLog in migrations

       builder.Services.AddDbContext<EventLogContext>(options =>
{
    options.UseSqlServer(connectionString,
                         sqlServerOptionsAction: sqlOptions =>
                         {
                             sqlOptions.MigrationsAssembly(typeof(ApplicationDbContext).GetTypeInfo().Assembly.GetName().Name);
                             sqlOptions.EnableRetryOnFailure(10, TimeSpan.FromSeconds(30), null);
                         });
});


then inject its used service

 builder.Services.AddTransient<IYourPortalIntegrationEventService, YourPortalIntegrationEventService>();


// For to Log events
builder.Services.AddTransient<Func<DbConnection, IEventLogService>>(
               sp => (DbConnection connection) => new EventLogService(connection));


              

This software is licensed under the Cubivue License.

Synergy-IT 2023.