IMessageReceiver on ServiceBusTrigger

[AmyJeanes]

I understand IMessageReceiver is no longer bindable with dotnet-isolated but what is the recommended way to achieve the functions it previously provided now? For example, manual completion of the message, receiving deferred messages, deferring messages etc. I cannot see any way to do this now.

[fabiocav]

Those scenarios are not currently supported. We'll continue to close those gaps as we enhance the binding model in the host and adopt that in the worker, but in the meantime, to use bindings for those scenarios, you still need to use the in-proc model.

[AmyJeanes]

That is a shame, but thank you for the quick response. Is this planned for .NET 5 or is this more like something on the roadmap for .NET 6? Just would like to get an idea of a timeline as this is blocking our upgrade to .NET 5

[timgabrhel]

Another vote here. Our .NET 5 upgrade was dependent on the new isolated hosting model, but I'm now stuck for this same reason. We have two functions that are dependent on being able to complete/abandon + update message Properties on messages.

[fabiocav]

To provide an update and answer the previous question, this requires enhancements to the host and SDK so the Worker can leverage the new capabilities once introduced, and given the timelines, this will land in-proc in .NET 6 prior to being done in the worker.

A lot of activity will be happening in the coming weeks to make a .NET 6 preview available in a new major version of Functions.

[MatheusXavier]

I have a service where I can complete the messages, so I don't need the IMessegaReceiver, but I need the lockToken, Are there some way to get that? On documentation they talk about the MessageReceiver so I used this property with string type but I received a json with empty TokenProvider:

  "RegisteredPlugins": [
    
  ],
  "ReceiveMode": 0,
  "PrefetchCount": 100,
  "LastPeekedSequenceNumber": 0,
  "Path": "desk4me-monitoring-statuses-local-development",
  "OperationTimeout": "00:01:00",
  "ServiceBusConnection": {
    "Endpoint": "sb://desk4me-dev.servicebus.windows.net",
    "OperationTimeout": "00:01:00",
    "RetryPolicy": {
      "MinimalBackoff": "00:00:00",
      "MaximumBackoff": "00:00:30",
      "DeltaBackoff": "00:00:03",
      "MaxRetryCount": 5,
      "IsServerBusy": false,
      "ServerBusyExceptionMessage": null
    },
    "TransportType": 0,
    "TokenProvider": {
      
    },
    "IsClosedOrClosing": false
  },
  "IsClosedOrClosing": false,
  "OwnsConnection": true,
  "ClientId": "MessageReceiver1desk4me-monitoring-statuses-local-development",
  "RetryPolicy": {
    "MinimalBackoff": "00:00:00",
    "MaximumBackoff": "00:00:30",
    "DeltaBackoff": "00:00:03",
    "MaxRetryCount": 5,
    "IsServerBusy": false,
    "ServerBusyExceptionMessage": null
  }
}

[timgabrhel]

@MatheusXavier The lockToken can be found in the binding metadata off of the FunctionContext passed into the function. I tried to init my own SB connection and mark the message as complete with the lock token, but service bus rejects it as the message was received by another MessageReceiver. Curious if you have an alternative to make it work.

[MatheusXavier]

@timgabrhel I tried your solution, so I could get lockToken from BindingContext:
var lockToken = context.BindingContext.BindingData.GetValueOrDefault("LockToken").ToString();
I discovered that if you add a parameter on your function called string lockToken you could get it too, but when I tried to complete the message I got the same error:

{"The lock supplied is invalid. Either the lock expired, or the message has already been removed from the queue, or was received by a different receiver instance."}

I built the Microsoft.Azure.ServiceBus.QueueClient with all my Service Bus and Queue connection settings and then called the CompleteAsync but it didn't work. If I get this QueueClient and register a message handler, and inside it I use the CompleteAsync It works fine.

[timgabrhel]

@MatheusXavier That's great to hear! Any chance you can share a code snippet around the message handler & completing the message?

[MatheusXavier]

Sure! I use two differents queues on my application, the first one I use to handle events from my domain, on application startup I register a message handler that receives events, it is working perfectly, this is the code:

        public void StartConsumer()
        {
            _serviceBusConnection.GetQueue().RegisterMessageHandler(
                async (message, token) =>
                {
                    var messageData = Encoding.UTF8.GetString(message.Body);

                    // Complete the message so that it is not received again.
                    if (await ProcessEventAsync(message.Label, messageData))
                    {
                        await _serviceBusConnection.GetQueue().CompleteAsync(message.SystemProperties.LockToken);
                    }
                },
                new MessageHandlerOptions(ExceptionReceivedHandler)
                {
                    MaxConcurrentCalls = 10,
                    AutoComplete = false,
                });
        }

The second queue I use to process some data, so I use an Azure Function with Service Bus Trigger, this is the code on dotnet 3.1 version:

    public class Processor
    {
        private readonly IMediator _mediator;

        public Processor(IMediator mediator)
        {
            _mediator = mediator;
        }

        [FunctionName("processor")]
        public async Task Run(
            [ServiceBusTrigger("%QueueName%", Connection = "ServiceBusConnection")]
            Message message,
            ILogger log,
            MessageReceiver messageReceiver)
        {
            try
            {
                var jsonData = Encoding.UTF8.GetString(message.Body);
                var settings = new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All };
                var deserialized = JsonConvert.DeserializeObject(jsonData, settings);

                await _mediator.Send(deserialized);
                await messageReceiver.CompleteAsync(message.SystemProperties.LockToken);
            }
            catch (Exception ex)
            {
                log.LogError(ex, $"[Processor] - An error with message: {ex.Message} occurred while " +
                    $"handling message with Id: '{message.MessageId}' and label: '{message.Label}' on {ex.StackTrace}");
            }
        }
    }

After the migration to dotnet 5 we don't have the MessageReceiver anymore, so I tried to use my _serviceBusConnection instance to complete the messages like the first one sample. This is the code:

    public class Processor
    {
        private readonly IMediator _mediator;
        private readonly IServiceBusPersisterConnection _serviceBus;

        public Processor(
            IMediator mediator,
            IServiceBusPersisterConnection serviceBus)
        {
            _mediator = mediator;
            _serviceBus = serviceBus;
        }

        [Function("processor")]
        public async Task Run(
            [ServiceBusTrigger("%QueueName%", Connection = "serviceBusConnection")] string message,
            string messageReceiver,
            string systemProperties,
            string messageId,
            string lockToken,
            FunctionContext context)
        {
            try
            {
                var settings = new JsonSerializerSettings { TypeNameHandling = TypeNameHandling.All };
                var deserialized = JsonConvert.DeserializeObject(message, settings);

                await _mediator.Send(deserialized);
                await _serviceBus.GetQueue().CompleteAsync(lockToken);
            }
            catch (Exception ex)
            {
                var log = context.GetLogger<Processor>();

                log.LogError(ex, $"[Processor] - An error with message: {ex.Message} occurred while " +
                    $"handling message with Id: '{messageId}' on {ex.StackTrace}");
            }
        }
    }

But I'm gettins this error on complete:

{"The lock supplied is invalid. Either the lock expired, or the message has already been removed from the queue, or was received by a different receiver instance."}

Any ideia to solve that?

[timgabrhel]

@MatheusXavier My apologies, I misunderstood and thought you meant you somehow were able to register a new message handler within the function to complete the message. Looks like we're on hold until this support is enabled by the functions team.