Admin
b3ff1b6516
Fix (recoverability bug): - TransferFilesToBucketActivity returned Completed() with no log, so its compensation could never delete the transferred files (orphaned objects on fault). IStorageService.TransferFilesAsync now returns the real uploaded keys; the activity logs them so Compensate can remove them. Also drops the Path.Combine key recomputation (produced backslash keys on Windows). Consistency / DRY: - Normalize all courier activities to internal + folder-based namespaces (several were public and in the flat .Activities namespace) - Extract the duplicated exponential retry into UseStandardRetry - Unify endpoint-name logic into a single EndpointHelper (remove the parallel MassTransitEndpointNames helper) Build: 0 warnings, 0 errors.
72 lines
3.3 KiB
C#
72 lines
3.3 KiB
C#
using MassTransit;
|
|
using Musify.Infrastructure.MassTransit.Consumers;
|
|
using Musify.Infrastructure.Persistence;
|
|
|
|
namespace Musify.Infrastructure.MassTransit
|
|
{
|
|
public static partial class MassTransitDependencyInjection
|
|
{
|
|
static void UseStandardRetry(IReceiveEndpointConfigurator endpointConfigurator) =>
|
|
endpointConfigurator.UseMessageRetry(retryConfigurator =>
|
|
retryConfigurator.Exponential(
|
|
retryLimit: 5,
|
|
minInterval: TimeSpan.FromSeconds(1),
|
|
maxInterval: TimeSpan.FromSeconds(30),
|
|
intervalDelta: TimeSpan.FromSeconds(2)));
|
|
|
|
static void ConfigureConsumerEndpoint<TConsumer>(
|
|
IRabbitMqBusFactoryConfigurator busFactoryConfigurator,
|
|
IBusRegistrationContext busRegistrationContext,
|
|
string queueName)
|
|
where TConsumer : class, IConsumer
|
|
{
|
|
busFactoryConfigurator.ReceiveEndpoint(queueName, endpointConfigurator =>
|
|
{
|
|
UseStandardRetry(endpointConfigurator);
|
|
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
|
|
endpointConfigurator.ConfigureConsumer<TConsumer>(busRegistrationContext);
|
|
});
|
|
}
|
|
|
|
static void ConfigureExecuteActivityEndpoint<TActivity, TArguments>(
|
|
IRabbitMqBusFactoryConfigurator busFactoryConfigurator,
|
|
IBusRegistrationContext busRegistrationContext,
|
|
string endpointName)
|
|
where TActivity : class, IExecuteActivity<TArguments>
|
|
where TArguments : class
|
|
{
|
|
busFactoryConfigurator.ReceiveEndpoint(EndpointHelper.ExecuteQueueName(endpointName), endpointConfigurator =>
|
|
{
|
|
UseStandardRetry(endpointConfigurator);
|
|
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
|
|
endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(busRegistrationContext);
|
|
});
|
|
}
|
|
|
|
static void ConfigureActivityEndpoint<TActivity, TArguments, TLog>(
|
|
IRabbitMqBusFactoryConfigurator busFactoryConfigurator,
|
|
IBusRegistrationContext busRegistrationContext,
|
|
string endpointName)
|
|
where TActivity : class, IActivity<TArguments, TLog>
|
|
where TArguments : class
|
|
where TLog : class
|
|
{
|
|
busFactoryConfigurator.ReceiveEndpoint(EndpointHelper.ExecuteQueueName(endpointName), endpointConfigurator =>
|
|
{
|
|
UseStandardRetry(endpointConfigurator);
|
|
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
|
|
endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(
|
|
EndpointHelper.BuildCompensateActivityUri(endpointName),
|
|
busRegistrationContext);
|
|
});
|
|
|
|
busFactoryConfigurator.ReceiveEndpoint(EndpointHelper.CompensateQueueName(endpointName), endpointConfigurator =>
|
|
{
|
|
UseStandardRetry(endpointConfigurator);
|
|
endpointConfigurator.UseEntityFrameworkOutbox<Database>(busRegistrationContext);
|
|
endpointConfigurator.CompensateActivityHost<TActivity, TLog>(busRegistrationContext);
|
|
});
|
|
}
|
|
}
|
|
}
|