Remove processingStatus
This commit is contained in:
@@ -1,7 +1,6 @@
|
||||
using MassTransit;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Musify.Application.Contracts.Infrastructure;
|
||||
using Musify.Domain.Entities;
|
||||
using Musify.Infrastructure.MassTransit.Activities.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Activities.Logs;
|
||||
|
||||
@@ -9,28 +8,13 @@ namespace Musify.Infrastructure.MassTransit.Activities.Files
|
||||
{
|
||||
internal class DownloadFileFromBucketActivity(
|
||||
IStorageService storageHandler,
|
||||
ILogger<DownloadFileFromBucketActivity> logger,
|
||||
IProcessTrackingStore processTrackingStore)
|
||||
ILogger<DownloadFileFromBucketActivity> logger)
|
||||
: IActivity<DownloadFileFromBucketArguments, DownloadFileFromBucketLog>
|
||||
{
|
||||
public const string ExecuteEndpointName = "download-file-from-bucket";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<DownloadFileFromBucketArguments> executeContext)
|
||||
{
|
||||
var processId = await processTrackingStore.GetOrCreateProcessAsync(
|
||||
"RoutingSlip",
|
||||
executeContext.CorrelationId ?? executeContext.TrackingNumber,
|
||||
executeContext.ConversationId,
|
||||
executeContext.MessageId,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var stepId = await processTrackingStore.StartStepAsync(
|
||||
processId,
|
||||
nameof(DownloadFileFromBucketActivity),
|
||||
ProcessStepComponentType.Activity,
|
||||
0,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var destinationPath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(destinationPath, nameof(destinationPath));
|
||||
|
||||
@@ -50,8 +34,6 @@ namespace Musify.Infrastructure.MassTransit.Activities.Files
|
||||
|
||||
using var destinationStream = File.Create(destinationPath);
|
||||
await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken);
|
||||
|
||||
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
|
||||
return executeContext.Completed(new DownloadFileFromBucketLog(
|
||||
executeContext.Arguments.Bucket,
|
||||
executeContext.Arguments.Key,
|
||||
@@ -60,7 +42,6 @@ namespace Musify.Infrastructure.MassTransit.Activities.Files
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Failed to download file from bucket");
|
||||
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
using MassTransit;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Musify.Application.Contracts.Infrastructure;
|
||||
using Musify.Domain.Entities;
|
||||
using Musify.Infrastructure.MassTransit.Activities.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Activities.Logs;
|
||||
|
||||
@@ -9,28 +8,13 @@ namespace Musify.Infrastructure.MassTransit.Activities
|
||||
{
|
||||
public class TransferFilesToBucketActivity(
|
||||
IStorageService storageHandler,
|
||||
ILogger<TransferFilesToBucketActivity> logger,
|
||||
IProcessTrackingStore processTrackingStore)
|
||||
ILogger<TransferFilesToBucketActivity> logger)
|
||||
: IActivity<TransferFilesToBucketArguments, TransferFilesToBucketLog>
|
||||
{
|
||||
public const string ExecuteEndpointName = "transfer-files-to-bucket";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<TransferFilesToBucketArguments> executeContext)
|
||||
{
|
||||
var processId = await processTrackingStore.GetOrCreateProcessAsync(
|
||||
"RoutingSlip",
|
||||
executeContext.CorrelationId ?? executeContext.TrackingNumber,
|
||||
executeContext.ConversationId,
|
||||
executeContext.MessageId,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var stepId = await processTrackingStore.StartStepAsync(
|
||||
processId,
|
||||
nameof(TransferFilesToBucketActivity),
|
||||
ProcessStepComponentType.Activity,
|
||||
0,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var folderPath = executeContext.GetVariable<string>(executeContext.Arguments.SourceDirectoryVariable);
|
||||
ArgumentNullException.ThrowIfNull(folderPath, nameof(folderPath));
|
||||
var destinationKey = executeContext.Arguments.DestinationKey;
|
||||
@@ -47,8 +31,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
destinationKey,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
|
||||
return executeContext.Completed();
|
||||
}
|
||||
catch (Exception exception)
|
||||
@@ -56,7 +38,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
|
||||
logger.LogError(exception, "Failed to transfer files from {FolderPath} to {DestinationBucket}",
|
||||
folderPath,
|
||||
executeContext.Arguments.DestinationBucket);
|
||||
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,35 +2,19 @@ using MassTransit;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using MimeMapping;
|
||||
using Musify.Application.Contracts.Infrastructure;
|
||||
using Musify.Domain.Entities;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Logs;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities.Files
|
||||
{
|
||||
public class UploadFileToBucketActivity(IStorageService storageHandler,
|
||||
ILogger<UploadFileToBucketActivity> logger,
|
||||
IProcessTrackingStore processTrackingStore)
|
||||
ILogger<UploadFileToBucketActivity> logger)
|
||||
: IActivity<UploadFileToBucketArguments, UploadFileToBucketLog>
|
||||
{
|
||||
public const string ExecuteEndpointName = "upload-file-to-bucket";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<UploadFileToBucketArguments> executeContext)
|
||||
{
|
||||
var processId = await processTrackingStore.GetOrCreateProcessAsync(
|
||||
"RoutingSlip",
|
||||
executeContext.CorrelationId ?? executeContext.TrackingNumber,
|
||||
executeContext.ConversationId,
|
||||
executeContext.MessageId,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var stepId = await processTrackingStore.StartStepAsync(
|
||||
processId,
|
||||
nameof(UploadFileToBucketActivity),
|
||||
ProcessStepComponentType.Activity,
|
||||
0,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.FilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(sourceFilePath, nameof(sourceFilePath));
|
||||
var fileName = Path.GetFileName(sourceFilePath);
|
||||
@@ -42,16 +26,13 @@ namespace Musify.Infrastructure.MassTransit.Activities.Files
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
Path.Combine(executeContext.Arguments.DestinationRoute, fileName),
|
||||
executeContext.CancellationToken);
|
||||
|
||||
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
|
||||
return executeContext.CompletedWithVariables(new UploadFileToBucketLog(
|
||||
executeContext.Arguments.DestinationBucket,
|
||||
executeContext.Arguments.DestinationRoute));
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Failed to upload file to bucket for process {ProcessId} step {StepId}", processId, stepId);
|
||||
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
|
||||
logger.LogError(exception, "Failed to upload file to bucket");
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
+1
-21
@@ -1,14 +1,11 @@
|
||||
using MassTransit;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Musify.Application.Contracts.Infrastructure;
|
||||
using Musify.Domain.Entities;
|
||||
using Musify.Infrastructure.MassTransit.Activities.Arguments;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Activities
|
||||
{
|
||||
public class GeneratePictureWorkflowPathsActivity(
|
||||
ILogger<GeneratePictureWorkflowPathsActivity> logger,
|
||||
IProcessTrackingStore processTrackingStore)
|
||||
ILogger<GeneratePictureWorkflowPathsActivity> logger)
|
||||
: IExecuteActivity<GeneratePictureWorkflowPathsArguments>
|
||||
{
|
||||
public const string ExecuteEndpointName = "generate-picture-workflow-paths";
|
||||
@@ -17,20 +14,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<GeneratePictureWorkflowPathsArguments> executeContext)
|
||||
{
|
||||
var processId = await processTrackingStore.GetOrCreateProcessAsync(
|
||||
"RoutingSlip",
|
||||
executeContext.CorrelationId ?? executeContext.TrackingNumber,
|
||||
executeContext.ConversationId,
|
||||
executeContext.MessageId,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var stepId = await processTrackingStore.StartStepAsync(
|
||||
processId,
|
||||
nameof(GeneratePictureWorkflowPathsActivity),
|
||||
ProcessStepComponentType.Activity,
|
||||
0,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
try
|
||||
{
|
||||
var destinationFolderName = Guid.NewGuid().ToString();
|
||||
@@ -49,8 +32,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
|
||||
|
||||
logger.LogDebug("Generated picture workflow paths in {WorkingDirectory}",
|
||||
workingDirectory);
|
||||
|
||||
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
|
||||
return executeContext.CompletedWithVariables(new Dictionary<string, object>
|
||||
{
|
||||
[RoutingSlipVariableNames.Workflow.TemporalDirectory] = workingDirectory,
|
||||
@@ -65,7 +46,6 @@ namespace Musify.Infrastructure.MassTransit.Activities
|
||||
{
|
||||
logger.LogError(exception, "Failed to generate picture workflow paths for {SourceKey}",
|
||||
executeContext.Arguments.SourceKey);
|
||||
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
using MassTransit;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Musify.Application.Contracts.Infrastructure;
|
||||
using Musify.Domain.Entities;
|
||||
using Musify.Infrastructure.MassTransit.Activities.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Activities.Logs;
|
||||
using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
|
||||
@@ -10,28 +9,13 @@ namespace Musify.Infrastructure.MassTransit.Activities
|
||||
{
|
||||
public class ResizePictureActivity(
|
||||
IPictureService pictureHandler,
|
||||
ILogger<ResizePictureActivity> logger,
|
||||
IProcessTrackingStore processTrackingStore)
|
||||
ILogger<ResizePictureActivity> logger)
|
||||
: IActivity<ResizePictureLocalArguments, ResizePictureLog>
|
||||
{
|
||||
public const string ExecuteEndpointName = "resize-picture";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<ResizePictureLocalArguments> executeContext)
|
||||
{
|
||||
var processId = await processTrackingStore.GetOrCreateProcessAsync(
|
||||
"RoutingSlip",
|
||||
executeContext.CorrelationId ?? executeContext.TrackingNumber,
|
||||
executeContext.ConversationId,
|
||||
executeContext.MessageId,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var stepId = await processTrackingStore.StartStepAsync(
|
||||
processId,
|
||||
nameof(ResizePictureActivity),
|
||||
ProcessStepComponentType.Activity,
|
||||
0,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var sourceFilePath = executeContext.GetVariable<string>(executeContext.Arguments.SourceFilePathVariable);
|
||||
ArgumentNullException.ThrowIfNull(sourceFilePath, nameof(sourceFilePath));
|
||||
var destinationFilePath = executeContext.GetVariable<string>(executeContext.Arguments.DestinationFilePathVariable);
|
||||
@@ -63,14 +47,11 @@ namespace Musify.Infrastructure.MassTransit.Activities
|
||||
logger.LogDebug("Resized picture from {SourceFilePath} to {DestinationFilePath}",
|
||||
sourceFilePath,
|
||||
destinationFilePath);
|
||||
|
||||
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
|
||||
return executeContext.Completed(new ResizePictureLog(destinationFilePath));
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Failed to resize picture {SourceFilePath}", sourceFilePath);
|
||||
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
using MassTransit;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Musify.Application.Contracts.Infrastructure;
|
||||
using Musify.Domain.Entities;
|
||||
using Musify.Infrastructure.MassTransit.Activities.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Activities.Logs;
|
||||
|
||||
@@ -9,28 +8,13 @@ namespace Musify.Infrastructure.MassTransit.Activities
|
||||
{
|
||||
public class UpdatePlayListPictureActivity(
|
||||
IDatabase database,
|
||||
ILogger<UpdatePlayListPictureActivity> logger,
|
||||
IProcessTrackingStore processTrackingStore)
|
||||
ILogger<UpdatePlayListPictureActivity> logger)
|
||||
: IActivity<UpdatePlayListPictureArguments, UpdatePlayListPictureLog>
|
||||
{
|
||||
public const string ExecuteEndpointName = "update-playlist-picture";
|
||||
|
||||
public async Task<ExecutionResult> Execute(ExecuteContext<UpdatePlayListPictureArguments> executeContext)
|
||||
{
|
||||
var processId = await processTrackingStore.GetOrCreateProcessAsync(
|
||||
"RoutingSlip",
|
||||
executeContext.CorrelationId ?? executeContext.TrackingNumber,
|
||||
executeContext.ConversationId,
|
||||
executeContext.MessageId,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var stepId = await processTrackingStore.StartStepAsync(
|
||||
processId,
|
||||
nameof(UpdatePlayListPictureActivity),
|
||||
ProcessStepComponentType.Activity,
|
||||
0,
|
||||
executeContext.CancellationToken);
|
||||
|
||||
var smallResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.SmallPictureVariable);
|
||||
ArgumentNullException.ThrowIfNull(smallResizedVariable, nameof(smallResizedVariable));
|
||||
var mediumResizedVariable = executeContext.GetVariable<string>(executeContext.Arguments.MediumPictureVariable);
|
||||
@@ -68,15 +52,12 @@ namespace Musify.Infrastructure.MassTransit.Activities
|
||||
|
||||
logger.LogInformation("Updated playlist {PlayListId} pictures",
|
||||
executeContext.Arguments.PlayListId);
|
||||
|
||||
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
|
||||
return executeContext.Completed();
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
logger.LogError(exception, "Failed to update playlist {PlayListId} pictures",
|
||||
executeContext.Arguments.PlayListId);
|
||||
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -11,7 +11,6 @@ using Musify.Infrastructure.MassTransit.Activities.Files;
|
||||
using Musify.Infrastructure.MassTransit.Activities.Logs;
|
||||
using Musify.Infrastructure.MassTransit.Arguments;
|
||||
using Musify.Infrastructure.MassTransit.Consumers;
|
||||
using Musify.Infrastructure.MassTransit.Filters;
|
||||
using Musify.Infrastructure.MassTransit.Logs;
|
||||
using Musify.Infrastructure.MassTransit.RoutingSlip.Builders;
|
||||
using Musify.Infrastructure.MassTransit.Activities.Tracks;
|
||||
@@ -92,8 +91,6 @@ namespace Musify.Infrastructure.MassTransit
|
||||
busFactoryConfigurator,
|
||||
busRegistrationContext.GetRequiredService<MassTransitConfiguration>());
|
||||
|
||||
busFactoryConfigurator.UseConsumeFilter(typeof(ProcessTrackingConsumeFilter<>), busRegistrationContext);
|
||||
|
||||
busFactoryConfigurator.ReceiveEndpoint(UpdatePlayListPictureConsumer.QueueName, endpointConfigurator =>
|
||||
endpointConfigurator.ConfigureConsumer<UpdatePlayListPictureConsumer>(busRegistrationContext));
|
||||
|
||||
|
||||
@@ -1,48 +0,0 @@
|
||||
using MassTransit;
|
||||
using Musify.Application.Contracts.Infrastructure;
|
||||
using Musify.Domain.Entities;
|
||||
|
||||
namespace Musify.Infrastructure.MassTransit.Filters
|
||||
{
|
||||
public class ProcessTrackingConsumeFilter<TMessage>(IProcessTrackingStore processTrackingStore)
|
||||
: IFilter<ConsumeContext<TMessage>>
|
||||
where TMessage : class
|
||||
{
|
||||
public async Task Send(ConsumeContext<TMessage> context, IPipe<ConsumeContext<TMessage>> next)
|
||||
{
|
||||
var processId = await processTrackingStore.GetOrCreateProcessAsync(
|
||||
typeof(TMessage).Name,
|
||||
context.CorrelationId ?? context.MessageId,
|
||||
context.ConversationId,
|
||||
context.MessageId,
|
||||
context.CancellationToken);
|
||||
|
||||
var stepId = await processTrackingStore.StartStepAsync(
|
||||
processId,
|
||||
$"{typeof(TMessage).Name}.Consume",
|
||||
ProcessStepComponentType.Consumer,
|
||||
0,
|
||||
context.CancellationToken);
|
||||
|
||||
try
|
||||
{
|
||||
await next.Send(context);
|
||||
await processTrackingStore.CompleteStepAsync(processId, stepId, context.CancellationToken);
|
||||
}
|
||||
catch (Exception exception)
|
||||
{
|
||||
await processTrackingStore.FailStepAsync(
|
||||
processId,
|
||||
stepId,
|
||||
exception.Message,
|
||||
context.CancellationToken);
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
public void Probe(ProbeContext context)
|
||||
{
|
||||
context.CreateFilterScope(nameof(ProcessTrackingConsumeFilter<>));
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user