Scheduling messages with Azure Service Bus #7302

rogercprops created 9 months ago

ABP Framework version: 7.2.2
UI Type: N/A
Database System: EF Core (SQL Server)
Tiered (for MVC) or Auth Server Separated (for Angular): yes
Exception message and full stack trace: N/A
Steps to reproduce the issue: N/A

We have a number of use cases where it would be ideal to utilize the Azure Service Bus message scheduling feature https://learn.microsoft.com/en-us/azure/service-bus-messaging/message-sequencing, but I can't find anything in the documentation, samples, blog posts or anywhere that shows how to do that.

For example when a client schedules an appointment we'd like to send reminders via SMS and/or email at various intervals (e.g. one week prior, 2 days prior to scheduled date).

Is this possible using AzureDistributedEventBus and if so, what are the detailed steps to implement?

Thank you.

liangshiwei created 9 months ago

Support Team Fullstack Developer

Hi,

You need to override the PublishAsync method of AzureDistributedEventBus

For example:

public interface IEventDataHasScheduledEnqueueTime
{
    DateTimeOffset ScheduledEnqueueTime { get; set; }
}

[Dependency(ReplaceServices = true)]
[ExposeServices(typeof(IDistributedEventBus), typeof(AzureDistributedEventBus))]
public class MyAzureDistributedEventBus : AzureDistributedEventBus
{
    public MyAzureDistributedEventBus(IServiceScopeFactory serviceScopeFactory, ICurrentTenant currentTenant, IUnitOfWorkManager unitOfWorkManager, IOptions<AbpDistributedEventBusOptions> abpDistributedEventBusOptions, IGuidGenerator guidGenerator, IClock clock, IOptions<AbpAzureEventBusOptions> abpAzureEventBusOptions, IAzureServiceBusSerializer serializer, IAzureServiceBusMessageConsumerFactory messageConsumerFactory, IPublisherPool publisherPool, IEventHandlerInvoker eventHandlerInvoker, ILocalEventBus localEventBus, ICorrelationIdProvider correlationIdProvider) : base(serviceScopeFactory, currentTenant, unitOfWorkManager, abpDistributedEventBusOptions, guidGenerator, clock, abpAzureEventBusOptions, serializer, messageConsumerFactory, publisherPool, eventHandlerInvoker, localEventBus, correlationIdProvider)
    {
    }

    protected async override Task PublishAsync(string eventName, object eventData)
    {
        var body = Serializer.Serialize(eventData);
        var message = new ServiceBusMessage(body)
        {
            Subject = eventName
        };
        
        if (eventData is IEventDataHasScheduledEnqueueTime eventDataHasScheduledEnqueueTime)
        {
            message.ScheduledEnqueueTime = eventDataHasScheduledEnqueueTime.ScheduledEnqueueTime;
        }

        if (message.MessageId.IsNullOrWhiteSpace())
        {
            message.MessageId = GuidGenerator.Create().ToString("N");
        }

        message.CorrelationId = CorrelationIdProvider.Get();

        var publisher = await PublisherPool.GetAsync(
            Options.TopicName,
            Options.ConnectionName);

        await publisher.SendMessageAsync(message);
    }
}