IReceiverClient : why does the RegisterMessageHandler not contain the client ?

[StefH]

Actual Behavior

Currently I'm using the code as:

public void Register()
{
    ISubscriptionClient client = _factory.Create();
    var messageHandlerOptions = new MessageHandlerOptions(ExceptionReceivedHandler)
    {
        MaxConcurrentCalls = 1,
        AutoComplete = false
    };

    client.RegisterMessageHandler(ProcessMessagesAsync, messageHandlerOptions);
}

private async Task ProcessMessagesAsync(Message message, CancellationToken cancellationToken)
{
    try
    {
        // Do something with message ...

        // Complete the message --> note that I need a reference to the client here !
        await client.CompleteAsync(message.SystemProperties.LockToken);
    }
    catch
    {
        if (!client.IsClosedOrClosing)
        {
            await client.AbandonAsync(message.SystemProperties.LockToken);
        }
    }
}

Expected Behavior

public void Register()
{
    // same as above...
}

private async Task ProcessMessagesAsync(ISubscriptionClient client, Message message, CancellationToken cancellationToken)
{
    try
    {
        // Do something with message ...

        // Complete the message --> client is passed to this callback so no need to keep a global-reference somewhere !
        await client.CompleteAsync(message.SystemProperties.LockToken);
    }
    catch
    {
        if (!client.IsClosedOrClosing)
        {
            await client.AbandonAsync(message.SystemProperties.LockToken);
        }
    }
}

[SeanFeldman]

@StefH could you please let know why us it an issue in your case to the the client?
You're in charge of creating and disposing it, aren't you?

[StefH]

I want to only keep a reference to the client in that Register method, and not define a class variable to hold the client.

[SeanFeldman]

That method is a callback registration method. If you need to terminate callback infinite running, you would need to do it outside anyways and you would need the reference to the client to invoke close method. Which means you'll need an external reference anyways.

[StefH]

I currently solve this by registering on the cancellationToken:

using MessageReceiverAsAService.Lib.Interfaces;
using Microsoft.Azure.ServiceBus;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json;
using System;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using TopicsExampleNewWithDI.Interfaces;
using TopicsExampleNewWithDI.Models;

namespace MessageReceiverAsAService.Lib.Implementations
{
    public class MessageHandlerService : IMessageHandlerService
    {
        private readonly ILogger<MessageHandlerService> _logger;
        private readonly ISubscriptionClientFactory _factory;
        private readonly IBinarySerializer _serializer;

        public MessageHandlerService(ILogger<MessageHandlerService> logger, ISubscriptionClientFactory factory, IBinarySerializer serializer)
        {
            _logger = logger;
            _factory = factory;
            _serializer = serializer;
        }

        public Task StartListeningAsync(CancellationToken stoppingToken)
        {
            _logger.LogInformation("StartListeningAsync");

            ISubscriptionClient client = _factory.Create();

            stoppingToken.Register(async () =>
            {
                _logger.LogInformation("StopListeningAsync");
                if (!client.IsClosedOrClosing)
                {
                    await client.CloseAsync();
                }
            });

            var messageHandlerOptions = new MessageHandlerOptions(ExceptionReceivedHandler)
            {
                MaxConcurrentCalls = 1,

                AutoComplete = false
            };

            _logger.LogInformation("Waiting for messages...");

            // Register the function that processes messages.
            client.RegisterMessageHandler((message, cancellationToken) => ProcessMessagesAsync(client, message, cancellationToken), messageHandlerOptions);

            return Task.CompletedTask;
        }

        private async Task ProcessMessagesAsync(ISubscriptionClient client, Message message, CancellationToken cancellationToken)
        {
            try
            {
                // process message ...
                
                await Task.Delay(TimeSpan.FromSeconds(1), cancellationToken); // simulate delay

                await client.CompleteAsync(message.SystemProperties.LockToken);
            }
            catch
            {
                if (!client.IsClosedOrClosing)
                {
                    await client.AbandonAsync(message.SystemProperties.LockToken);
                }
            }
        }

        private Task ExceptionReceivedHandler(ExceptionReceivedEventArgs exceptionReceivedEventArgs)
        {
            var context = exceptionReceivedEventArgs.ExceptionReceivedContext;
            _logger.LogError(exceptionReceivedEventArgs.Exception, $"Endpoint: {context.Endpoint} | Entity Path: {context.EntityPath} | Executing Action: {context.Action}");

            return Task.CompletedTask;
        }
    }
}

Full example project can be found at:
https://github.com/StefH/dotnetcore-windows-service/tree/master/MessageReceiverAsAService

[SeanFeldman]

Clever! 🙂

If MessageHandlerService has a field of SubscriptionClient, when it's disposed, the client will be disposed as well, wouldn't it? What would be the value of having the client passed into the user callback aside from removing the need in the field variable?

Saying that, it would be possible to pass the client into the user callback, but this would be a breaking change and would need to wait for the next major.

[SeanFeldman]

Actually this could be a minor with a follow up major to pull the new signatures into the contracts.

@axisc thoughts?

[axisc]

There may be value in passing the client reference, primarily to decouple the handling logic from the actual clients.

It may still be a flavor of what you have done in your sample code @StefH , so thanks for that. Will check on this

[SeanFeldman]

@axisc, I've made a suggestion (@nemakam can elaborate on details). If that works, could PR it.

[axisc]

@SeanFeldman , I synced with @nemakam on this.

The idea of passing the IReceiverClient reference as an argument to RegisterMessageHandler() may be a good idea.

In the future versions (as a breaking change), I would also recommend abstracting handling logic to it's own class (i.e. designing with SOLID principles in mind).