src/Transports/MassTransit.Azure.ServiceBus.Core/Testing/AzureFunctionsTestExtensions.cs - develop - vshmidt/masstransit

src/Transports/MassTransit.Azure.ServiceBus.Core/Testing/AzureFunctionsTestExtensions.cs

54 строки2 KB

Chris Patterson

Simple clean up for Azure Function configuration extension and methods

25 июн 2025, 19:03

25 июн 2025, 19:039376d1b

namespace MassTransit.Testing;

using System;
using System.Reflection;
using System.Threading.Tasks;
using Azure.Core.Amqp;
using Azure.Messaging.ServiceBus;
using AzureServiceBusTransport;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.DependencyInjection.Extensions;
using Serialization;


public static class AzureFunctionsTestExtensions
{
    public static IBusRegistrationConfigurator AddAzureFunctionsTestComponents(this IBusRegistrationConfigurator configurator)
    {
        configurator.TryAddSingleton<IAsyncBusHandle, AsyncBusHandle>();
        configurator.TryAddSingleton<IMessageReceiver, MessageReceiver>();

        return configurator;
    }

    /// <summary>
    /// Handle the Azure Service Bus message using the specified consumer
    /// </summary>
    /// <param name="harness"></param>
    /// <param name="message"></param>
    /// <typeparam name="TConsumer"></typeparam>
    public static Task HandleConsumer<TConsumer>(this ITestHarness harness, object message)
        where TConsumer : class, IConsumer
    {
        var body = SystemTextJsonMessageSerializer.Instance.SerializeObject(message);

        var messageBody = new AmqpMessageBody([new BinaryData(body.GetBytes()).ToMemory()]);
        var annotatedMessage = new AmqpAnnotatedMessage(messageBody)
        {
            Header = { DeliveryCount = 1 },
            Properties =
            {
                MessageId = new AmqpMessageId(NewId.NextGuid().ToString()),
                ContentType = SystemTextJsonRawMessageSerializer.JsonContentType.MediaType
            }
        };

        var receivedMessage = (ServiceBusReceivedMessage)typeof(ServiceBusReceivedMessage).GetConstructor(
            BindingFlags.NonPublic | BindingFlags.Instance, null, [typeof(AmqpAnnotatedMessage)], null).Invoke([annotatedMessage]);

        var receiver = harness.Scope.ServiceProvider.GetRequiredService<IMessageReceiver>();
        var formatter = harness.Scope.ServiceProvider.GetService<IEndpointNameFormatter>() ?? DefaultEndpointNameFormatter.Instance;

        return receiver.HandleConsumer<TConsumer>(formatter.Consumer<TConsumer>(), receivedMessage, harness.CancellationToken);
    }
}