0
\$\begingroup\$

I'm using azure's service bus in order to make orders using a IOrderLogicService interface. The Logger.InsertLog is irrelevant for this thread.

private readonly IOrderLogicService _orderLogicService;
private readonly AppSettingsConfig _config;
private IQueueClient _queueClient;

public ServiceBusQueueHandler(AppSettingsConfig config, IOrderLogicService orderLogicService)
{
    _config = config;
    _orderLogicService = orderLogicService;

    _queueClient = new QueueClient(_config.connectionString, _config.QueueName);
}

public void RegisterServiceBus()
{
    RegisterMessageHandler().GetAwaiter().GetResult();
}

private  async Task RegisterMessageHandler(int tries = 1)
{
    try
    {
        if (tries != 3)
        {
            _queueClient.RegisterMessageHandler(ProcessMessagesAsync, CreateMessageHandlerOptions());                    
        }
    }
    catch(NullReferenceException ex)
    {
        Logger.InsertLog($"Faild connection service bus queue, _queueClient variable is null",E_LogType.Error ,ex);
    }
    catch(Exception e)
    {
        Logger.InsertLog($"Faild connection service bus queue, this is try number : {tries} , try again!", E_LogType.Warning, e);
        tries++;
        await RegisterMessageHandler(tries);

        if (tries == 3)
        {
            await _queueClient.CloseAsync();
            Logger.InsertLog($"QueueClient closed , try number {tries}", E_LogType.Error, e);
        }
    }

}

private MessageHandlerOptions CreateMessageHandlerOptions()
{
    return  new MessageHandlerOptions(ExceptionReceivedHandler)
    {
        MaxConcurrentCalls = _config.Azure.MaxConcurrentCalls,
        MaxAutoRenewDuration = TimeSpan.FromSeconds(10),
        AutoComplete = false
    };
}

private async Task ProcessMessagesAsync(Message message, CancellationToken token)
{
    if (message != null)
    {
        await ProccesMessage(message);              
    }
}

private async Task ProccesMessage(Message message)
{
    Logger.InsertLog($"Recive Message - MessageId: {message.MessageId}", E_LogType.Info);
    string messagebody = Encoding.UTF8.GetString(message.Body);
    OrderRequestDTO orderMessage = Serialization.DeserializeFromJson<OrderRequestDTO>(messagebody);

    if(!string.IsNullOrWhiteSpace(messagebody))
    {
        Logger.InsertLog(messagebody, E_LogType.Info, null, null);
    }
    else
    {
        Logger.InsertLog($"Empty Message", E_LogType.Error, null, null);
    }
    await _queueClient.CompleteAsync(message.SystemProperties.LockToken);
    await _orderLogicService.DoOrder(orderMessage);
}

private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
{
    StringBuilder str = new StringBuilder();
    var context = exceptionReceivedEventArgs.ExceptionReceivedContext;
    str.Append($"Message handler encountered an exception {exceptionReceivedEventArgs.Exception}.");
    str.Append("Exception context for troubleshooting:");
    str.Append($"- Endpoint: {context.Endpoint}");
    str.Append($"- Entity Path: {context.EntityPath}");
    str.Append($"- Executing Action: {context.Action}");

    Logger.InsertLog($"ServiceBus - Exception - ReceivedHandler {str}",E_LogType.Error, exceptionReceivedEventArgs.Exception);

    return Task.CompletedTask;
}

startup.cs:

public void ConfigureServices(IServiceCollection services)
{
    //...
    services.AddTransient<IServiceBusQueueHandler, ServiceBusQueueHandler>();
}

public void Configure(IApplicationBuilder app, IHostingEnvironment env)
{
    //Apply when we have connection string
    var serviceBus = app.ApplicationServices.CreateScope().ServiceProvider.GetService<IServiceBusQueueHandler>();
    serviceBus.RegisterServiceBus();
}    
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.