src/Transports/MassTransit.Azure.ServiceBus.Core/Configuration/ServiceBusMessageSchedulerBusExtensions.cs - develop - vshmidt/masstransit

src/Transports/MassTransit.Azure.ServiceBus.Core/Configuration/ServiceBusMessageSchedulerBusExtensions.cs

67 строк3 KB

Chris Patterson

AddSqlMessageScheduler() method created so that scheduled messages can be canceled

17 май 2024, 17:42

17 май 2024, 17:4240775e9

namespace MassTransit
{
    using DependencyInjection;
    using Microsoft.Extensions.DependencyInjection;
    using Microsoft.Extensions.DependencyInjection.Extensions;
    using Scheduling;


    public static class ServiceBusMessageSchedulerBusExtensions
    {
        /// <summary>
        /// Create a message scheduler that uses the Azure Service Bus ScheduleEnqueueTimeUtc property to
        /// schedule messages.
        /// NOTE that this should only be used to schedule messages outside of a message consumer. Consumers should
        /// use the ScheduleSend extensions on ConsumeContext.
        /// </summary>
        /// <param name="bus"></param>
        /// <returns></returns>
        public static IMessageScheduler CreateServiceBusMessageScheduler(this IBus bus)
        {
            return new MessageScheduler(new ServiceBusScheduleMessageProvider(bus), bus.Topology);
        }

        /// <summary>
        /// Create a message scheduler that uses the Azure Service Bus ScheduleEnqueueTimeUtc property to
        /// schedule messages.
        /// NOTE that this should only be used to schedule messages outside of a message consumer. Consumers should
        /// use the ScheduleSend extensions on ConsumeContext.
        /// </summary>
        /// <param name="sendEndpointProvider"></param>
        /// <param name="busTopology"></param>
        /// <returns></returns>
        public static IMessageScheduler CreateServiceBusMessageScheduler(this ISendEndpointProvider sendEndpointProvider, IBusTopology busTopology)
        {
            return new MessageScheduler(new ServiceBusScheduleMessageProvider(sendEndpointProvider), busTopology);
        }

        /// <summary>
        /// Add an <see cref="IMessageScheduler" /> to the container that uses the Azure message enqueue time to schedule messages.
        /// </summary>
        /// <param name="configurator"></param>
        public static void AddServiceBusMessageScheduler(this IRegistrationConfigurator configurator)
        {
            configurator.TryAddScoped(provider =>
            {
                var bus = provider.GetRequiredService<IBus>();
                var sendEndpointProvider = provider.GetRequiredService<ISendEndpointProvider>();
                return sendEndpointProvider.CreateServiceBusMessageScheduler(bus.Topology);
            });
        }

        /// <summary>
        /// Add an <see cref="IMessageScheduler" /> to the container that uses the Azure message enqueue time to schedule messages.
        /// </summary>
        /// <param name="configurator"></param>
        public static void AddServiceBusMessageScheduler<TBus>(this IBusRegistrationConfigurator<TBus> configurator)
            where TBus : class, IBus
        {
            configurator.TryAddScoped(provider =>
            {
                var bus = provider.GetRequiredService<TBus>();
                var sendEndpointProvider = provider.GetRequiredService<ISendEndpointProvider>();
                return Bind<TBus>.Create(sendEndpointProvider.CreateServiceBusMessageScheduler(bus.Topology));
            });
        }
    }
}