Changes but failing

This commit is contained in:
2026-03-23 03:13:08 +01:00
parent 27ec670e3a
commit 4ec4b50c40
16 changed files with 362 additions and 160 deletions
@@ -2,8 +2,8 @@
{
public record UpdateTrackPictureEvent(
Guid TrackId,
string BucketName,
string OriginalPictureKeyName,
string SourceBucketName,
string SourceKeyName,
string SmallPictureKeyName,
int SmallPictureWidth,
int SmallPictureHeight,
@@ -42,6 +42,14 @@ namespace Musify.Application.Tracks.Handler
return Result.Error($"Failed to upload picture for track {request.Title}");
}
var audioKeyName = Path.Combine(trackConfiguration.Routes.Audios, $"{Guid.NewGuid() }.{request.Audio.FileType}");
var audioUploadResult = await UploadFile(request.Audio, audioKeyName, cancellationToken);
if (!audioUploadResult.IsSuccess)
{
logger.LogError("Failed to upload audio for track {TrackTitle}", request.Title);
return Result.Error($"Failed to upload audio for track {request.Title}");
}
try
{
await PublishUpdateEvent(originalPictureKeyName, track, cancellationToken);
@@ -52,14 +60,6 @@ namespace Musify.Application.Tracks.Handler
return Result.Error($"Failed to publish resize picture event for Track with id {track.Id}.");
}
var audioKeyName = Path.Combine(trackConfiguration.Routes.Audios, $"{Guid.NewGuid() }.{request.Audio.FileType}");
var audioUploadResult = await UploadFile(request.Audio, audioKeyName, cancellationToken);
if (!audioUploadResult.IsSuccess)
{
logger.LogError("Failed to upload audio for track {TrackTitle}", request.Title);
return Result.Error($"Failed to upload audio for track {request.Title}");
}
try
{
await PublishTranscodeEvent(track.Id, audioUploadResult, cancellationToken);
@@ -0,0 +1,8 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments
{
public record ResizePictureLocalArguments(
string SourceFilePath,
string DestinationFilePath,
int Width,
int Height);
}
@@ -1,17 +1,38 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.Messaging.Activities.Arguments;
using Musify.Infrastructure.Messaging.Activities.Logs;
namespace Musify.Infrastructure.Messaging.Activities
{
public class DownloadFileFromBucketActivity(IStorageHandler storageHandler, ILogger<DownloadFileFromBucketActivity> logger)
public class DownloadFileFromBucketActivity(
IStorageHandler storageHandler,
ILogger<DownloadFileFromBucketActivity> logger,
IProcessTrackingStore processTrackingStore,
AudioTranscoderConfiguration audioTranscoderConfiguration)
: IExecuteActivity<DownloadFileFromBucketArguments>
{
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);
try
{
var getFile = await storageHandler.GetFileAsync(
executeContext.Arguments.BucketName,
@@ -20,22 +41,36 @@ namespace Musify.Infrastructure.Messaging.Activities
if (!getFile.IsSuccess)
{
var error = string.Join("; ", getFile.Errors);
logger.LogError("Failed to download file from bucket. Bucket: {BucketName}, Key: {KeyName}",
var errorMessage = string.Join("; ", getFile.Errors);
logger.LogError("Failed to download file from bucket. Bucket: {BucketName}, Key: {KeyName}, Errors: {Errors}",
executeContext.Arguments.BucketName,
executeContext.Arguments.KeyName);
throw new Exception(error);
executeContext.Arguments.KeyName,
errorMessage);
throw new Exception(errorMessage);
}
using var fileStream = getFile.Value;
if (fileStream.CanSeek)
fileStream.Position = 0;
var fileName = Path.GetFileName(executeContext.Arguments.KeyName);
using var destinationStream = File.Create($"D:\\tempsFilesDev\\{fileName}");
var destinationPath = Path.Combine(
audioTranscoderConfiguration.Routes.WorkingDirectory,
Path.GetFileName(executeContext.Arguments.KeyName));
Directory.CreateDirectory(Path.GetDirectoryName(destinationPath)!);
using var destinationStream = File.Create(destinationPath);
await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Error downloading file from bucket");
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
}
}
@@ -39,9 +39,8 @@ namespace Musify.Infrastructure.Messaging.Activities
if (!removeFile.IsSuccess)
{
string errorMessage = string.Join(";", removeFile.Errors);
logger.LogWarning(
"Failed to remove file {KeyName} from bucket {BucketName}: {ErrorMessage}",
var errorMessage = string.Join("; ", removeFile.Errors);
logger.LogWarning("Failed to remove file {KeyName} from bucket {BucketName}. Errors: {Errors}",
executeContext.Arguments.KeyName,
executeContext.Arguments.BucketName,
errorMessage);
@@ -52,6 +51,9 @@ namespace Musify.Infrastructure.Messaging.Activities
}
catch (Exception exception)
{
logger.LogError(exception, "Error removing file {KeyName} from bucket {BucketName}",
executeContext.Arguments.KeyName,
executeContext.Arguments.BucketName);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
@@ -1,4 +1,5 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
@@ -6,14 +7,14 @@ using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Activities
{
public class ResizePictureActivity(
IStorageHandler storageHandler,
IPictureHandler pictureHandler,
ILogger<ResizePictureActivity> logger,
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<ResizePictureArgument>
: IExecuteActivity<ResizePictureLocalArguments>
{
public const string ExecuteEndpointName = "Resize-Picture";
public async Task<ExecutionResult> Execute(ExecuteContext<ResizePictureArgument> executeContext)
public async Task<ExecutionResult> Execute(ExecuteContext<ResizePictureLocalArguments> executeContext)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
@@ -31,47 +32,54 @@ namespace Musify.Infrastructure.Messaging.Activities
try
{
var file = await storageHandler.GetFileAsync(
executeContext.Arguments.OriginalBucketName,
executeContext.Arguments.OriginalKeyName,
executeContext.CancellationToken);
if (!file.IsSuccess)
if (!File.Exists(executeContext.Arguments.SourceFilePath))
{
string errorMessages = string.Join(", ", file.Errors);
throw new Exception(errorMessages);
logger.LogError("Source file not found at {SourceFilePath}",
executeContext.Arguments.SourceFilePath);
throw new FileNotFoundException($"Source file not found: {executeContext.Arguments.SourceFilePath}");
}
await using var fileStream = File.OpenRead(executeContext.Arguments.SourceFilePath);
var resizedPicture = await pictureHandler.ResizePictureAsync(
file.Value,
fileStream,
executeContext.Arguments.Width,
executeContext.Arguments.Height,
executeContext.CancellationToken);
if (!resizedPicture.IsSuccess)
{
string errorMessage = string.Join(", ", resizedPicture.Errors);
var errorMessage = string.Join("; ", resizedPicture.Errors);
logger.LogError("Failed to resize picture {SourceFilePath} to {Width}x{Height}. Errors: {Errors}",
executeContext.Arguments.SourceFilePath,
executeContext.Arguments.Width,
executeContext.Arguments.Height,
errorMessage);
throw new Exception(errorMessage);
}
var uploadFile = await storageHandler.UploadFileAsync(
resizedPicture.Value,
pictureHandler.ContentType,
executeContext.Arguments.DestinationBucketName,
executeContext.Arguments.DestinationKeyName,
executeContext.CancellationToken);
if (!uploadFile.IsSuccess)
var destinationDirectory = Path.GetDirectoryName(executeContext.Arguments.DestinationFilePath);
if (!string.IsNullOrEmpty(destinationDirectory))
{
var errorMessage = string.Join(", ", uploadFile.Errors);
throw new Exception(errorMessage);
Directory.CreateDirectory(destinationDirectory);
}
await using var destinationStream = File.Create(executeContext.Arguments.DestinationFilePath);
await resizedPicture.Value.CopyToAsync(destinationStream, executeContext.CancellationToken);
logger.LogInformation("Picture resized successfully from {SourceFilePath} to {DestinationFilePath} ( {Width}x{Height} )",
executeContext.Arguments.SourceFilePath,
executeContext.Arguments.DestinationFilePath,
executeContext.Arguments.Width,
executeContext.Arguments.Height);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Error resizing picture {SourceFilePath}",
executeContext.Arguments.SourceFilePath);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
@@ -1,26 +1,43 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.Messaging.Activities.Arguments;
using Musify.Infrastructure.Messaging.Activities.Logs;
namespace Musify.Infrastructure.Messaging.Activities
{
public class TranscodeDashAudioActivity(IAudioTranscoder audioTranscoder, ILogger<TranscodeDashAudioActivity> logger,
public class TranscodeDashAudioActivity(
IAudioTranscoder audioTranscoder,
ILogger<TranscodeDashAudioActivity> logger,
IProcessTrackingStore processTrackingStore,
AudioTranscoderConfiguration audioTranscoderConfiguration)
: IExecuteActivity<TranscodeDashAudioArguments>
{
public const string ExecuteEndpointName = "Transcode-Dash-Audio";
public async Task<ExecutionResult> Execute(ExecuteContext<TranscodeDashAudioArguments> executeContext)
{
var processId = await processTrackingStore.GetOrCreateProcessAsync(
"RoutingSlip",
executeContext.CorrelationId ?? executeContext.TrackingNumber,
executeContext.ConversationId,
executeContext.MessageId,
executeContext.CancellationToken);
var stepId = await processTrackingStore.StartStepAsync(
processId,
nameof(TranscodeDashAudioActivity),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
try
{
var workingDirectory = Path.Combine(
audioTranscoderConfiguration.Routes.WorkingDirectory,
executeContext.Arguments.DestinationFolderName);
try
{
await using var fileStream = File.OpenRead(executeContext.Arguments.SourceFilePath);
var result = await audioTranscoder.TranscodeToDashAsync(
@@ -31,17 +48,20 @@ namespace Musify.Infrastructure.Messaging.Activities
if (!result.IsSuccess)
{
var errors = string.Join("; ", result.Errors);
logger.LogError("transcoding failed for file {SourceFilePath}. Error: {ErrorMessage}",
executeContext.Arguments.SourceFilePath, errors);
throw new Exception(errors);
var errorMessage = string.Join("; ", result.Errors);
logger.LogError("Transcoding failed for file {SourceFilePath}. Errors: {ErrorMessage}",
executeContext.Arguments.SourceFilePath, errorMessage);
throw new Exception(errorMessage);
}
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Error during transcoding activity.");
logger.LogError(exception, "Error during transcoding activity for file {SourceFilePath}",
executeContext.Arguments.SourceFilePath);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
@@ -1,16 +1,36 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Activities
{
public class TransferFilesToBucket(IStorageHandler storageHandler, ILogger<TransferFilesToBucket> logger)
public class TransferFilesToBucket(
IStorageHandler storageHandler,
ILogger<TransferFilesToBucket> logger,
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<TransferFilesToBucketArguments>
{
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(TransferFilesToBucket),
ProcessStepComponentType.Activity,
0,
executeContext.CancellationToken);
try
{
var transferFilesResult = await storageHandler.TransferFilesAsync(
executeContext.Arguments.FolderPath,
@@ -20,12 +40,26 @@ namespace Musify.Infrastructure.Messaging.Activities
if (!transferFilesResult.IsSuccess)
{
var errors = string.Join("; ", transferFilesResult.Errors);
logger.LogError($"An error occurred while transferring files to bucket: {errors}");
throw new Exception(errors);
var errorMessage = string.Join("; ", transferFilesResult.Errors);
logger.LogError("Failed to transfer files from {FolderPath} to bucket {DestinationBucketName}, key {DestinationKeyName}. Errors: {Errors}",
executeContext.Arguments.FolderPath,
executeContext.Arguments.DestinationBucketName,
executeContext.Arguments.DestinationKeyName,
errorMessage);
throw new Exception(errorMessage);
}
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Error transferring files from {FolderPath} to bucket {DestinationBucketName}",
executeContext.Arguments.FolderPath,
executeContext.Arguments.DestinationBucketName);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
}
}
}
@@ -1,11 +1,15 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Activities
{
public class UpdatePlayListPictureActivity(IDatabase database, IProcessTrackingStore processTrackingStore)
public class UpdatePlayListPictureActivity(
IDatabase database,
ILogger<UpdatePlayListPictureActivity> logger,
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<UpdatePlayListPictureArguments>
{
public const string ExecuteEndpointName = "Update-PlayList-Picture";
@@ -28,10 +32,16 @@ namespace Musify.Infrastructure.Messaging.Activities
try
{
var playList = await database.PlayLists.FindAsync(executeContext.Arguments.PlayListId, executeContext.CancellationToken);
var playList = await database.PlayLists.FindAsync(
new object[] { executeContext.Arguments.PlayListId },
cancellationToken: executeContext.CancellationToken);
if (playList is null)
throw new Exception($"PlayList with id {executeContext.Arguments.PlayListId} not found.");
{
logger.LogError("PlayList with id {PlayListId} not found",
executeContext.Arguments.PlayListId);
throw new InvalidOperationException($"PlayList with id {executeContext.Arguments.PlayListId} not found");
}
playList.SmallPictureKeyName = executeContext.Arguments.SmallPictureKeyName;
playList.MediumPictureKeyName = executeContext.Arguments.MediumPictureKeyName;
@@ -40,11 +50,16 @@ namespace Musify.Infrastructure.Messaging.Activities
database.PlayLists.Update(playList);
await database.SaveChangesAsync(executeContext.CancellationToken);
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, "Error updating PlayList {PlayListId} pictures",
executeContext.Arguments.PlayListId);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
@@ -1,11 +1,15 @@
using MassTransit;
using Microsoft.Extensions.Logging;
using Musify.Application.Contracts.Infrastructure;
using Musify.Domain.Entities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Activities
{
public class UpdateTrackPictureActivity(IDatabase database, IProcessTrackingStore processTrackingStore)
public class UpdateTrackPictureActivity(
IDatabase database,
ILogger<UpdateTrackPictureActivity> logger,
IProcessTrackingStore processTrackingStore)
: IExecuteActivity<UpdateTrackPictureArguments>
{
public const string ExecuteEndpointName = "Update-Track-Picture";
@@ -28,10 +32,16 @@ namespace Musify.Infrastructure.Messaging.Activities
try
{
var track = await database.Tracks.FindAsync(executeContext.Arguments.TrackId, executeContext.CancellationToken);
var track = await database.Tracks.FindAsync(
new object[] { executeContext.Arguments.TrackId },
cancellationToken: executeContext.CancellationToken);
if (track is null)
throw new Exception($"Track with id {executeContext.Arguments.TrackId} not found.");
{
logger.LogError("Track with id {TrackId} not found",
executeContext.Arguments.TrackId);
throw new InvalidOperationException($"Track with id {executeContext.Arguments.TrackId} not found");
}
track.SmallPictureKeyName = executeContext.Arguments.SmallPictureKeyName;
track.MediumPictureKeyName = executeContext.Arguments.MediumPictureKeyName;
@@ -40,11 +50,16 @@ namespace Musify.Infrastructure.Messaging.Activities
database.Tracks.Update(track);
await database.SaveChangesAsync(executeContext.CancellationToken);
logger.LogInformation("Updated Track {TrackId} pictures",
executeContext.Arguments.TrackId);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed();
}
catch (Exception exception)
{
logger.LogError(exception, "Error updating Track {TrackId} pictures",
executeContext.Arguments.TrackId);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw;
}
@@ -0,0 +1,7 @@
namespace Musify.Infrastructure.Messaging.Consumers
{
internal static class MessagingHelper
{
internal static Uri BuildExecuteActivityUri(string endpointName) => new($"queue:{endpointName}_execute");
}
}
@@ -16,7 +16,7 @@ namespace Musify.Infrastructure.Messaging.Consumers
routingSlipBuilder.AddActivity(
"RemoveFile",
BuildExecuteUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
MessagingHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
consumeContext.Message.BucketName,
consumeContext.Message.KeyName));
@@ -24,7 +24,5 @@ namespace Musify.Infrastructure.Messaging.Consumers
var routingSlip = routingSlipBuilder.Build();
await bus.Execute(routingSlip);
}
static Uri BuildExecuteUri(string endpointName) => new($"queue:{endpointName}_execute");
}
}
@@ -6,7 +6,9 @@ using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Consumers
{
public class TranscodeAudioFromTrackConsumer(IBus bus, AudioTranscoderConfiguration audioTranscoderConfiguration)
public class TranscodeAudioFromTrackConsumer(
IBus bus,
AudioTranscoderConfiguration audioTranscoderConfiguration)
: IConsumer<TranscodeAudioFromTrackEvent>
{
public const string QueueName = "Transcode-Audio-From-Track";
@@ -14,41 +16,35 @@ namespace Musify.Infrastructure.Messaging.Consumers
public async Task Consume(ConsumeContext<TranscodeAudioFromTrackEvent> consumeContext)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
var fileName = Path.GetFileName(consumeContext.Message.SourceKeyName);
var folderName = Guid.NewGuid().ToString();
var workingDirectory = Path.Combine("D:\\tempsFilesDev\\");
var workingDirectory = Path.Combine(audioTranscoderConfiguration.Routes.WorkingDirectory, folderName);
routingSlipBuilder.AddActivity(
"DownloadFile",
BuildExecuteUri(DownloadFileFromBucketActivity.ExecuteEndpointName),
MessagingHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName),
new DownloadFileFromBucketArguments(
consumeContext.Message.SourceBucketName,
consumeContext.Message.SourceKeyName,
workingDirectory)
);
var fileName = Path.GetFileName(consumeContext.Message.SourceKeyName);
workingDirectory));
routingSlipBuilder.AddActivity(
"TranscodeAudio",
BuildExecuteUri(TranscodeDashAudioActivity.ExecuteEndpointName),
MessagingHelper.BuildExecuteActivityUri(TranscodeDashAudioActivity.ExecuteEndpointName),
new TranscodeDashAudioArguments(
$"D:\\tempsFilesDev\\{fileName}",
"D:\\tempsFilesDev\\")
);
Path.Combine(workingDirectory, fileName),
folderName));
routingSlipBuilder.AddActivity(
"UploadFile",
BuildExecuteUri(TransferFilesToBucket.ExecuteEndpointName),
MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName),
new TransferFilesToBucketArguments(
workingDirectory,
consumeContext.Message.DestinationBucketName,
consumeContext.Message.DestinationKeyName)
);
consumeContext.Message.DestinationKeyName));
var routingSlip = routingSlipBuilder.Build();
await bus.Execute(routingSlip);
}
static Uri BuildExecuteUri(string endpointName) => new($"queue:{endpointName}_execute");
}
}
@@ -1,11 +1,14 @@
using MassTransit;
using Musify.Application.Events;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.Messaging.Activities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Consumers
{
public class UpdatePlayListPictureConsumer(IBus bus)
public class UpdatePlayListPictureConsumer(
IBus bus,
AudioTranscoderConfiguration audioTranscoderConfiguration)
: IConsumer<UpdatePlayListPictureEvent>
{
public const string QueueName = "Update-PlayList-Picture";
@@ -13,6 +16,8 @@ namespace Musify.Infrastructure.Messaging.Consumers
public async Task Consume(ConsumeContext<UpdatePlayListPictureEvent> consumeContext)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
var folderName = Guid.NewGuid().ToString();
var workingDirectory = Path.Combine(audioTranscoderConfiguration.Routes.WorkingDirectory, folderName);
AddResizeActivity(
routingSlipBuilder,
@@ -21,7 +26,8 @@ namespace Musify.Infrastructure.Messaging.Consumers
consumeContext.Message.SmallPictureWidth,
consumeContext.Message.SmallPictureHeight,
consumeContext.Message.OriginalPictureKeyName,
consumeContext.Message.BucketName);
consumeContext.Message.BucketName,
workingDirectory);
AddResizeActivity(
routingSlipBuilder,
@@ -30,7 +36,8 @@ namespace Musify.Infrastructure.Messaging.Consumers
consumeContext.Message.MediumPictureWidth,
consumeContext.Message.MediumPictureHeight,
consumeContext.Message.OriginalPictureKeyName,
consumeContext.Message.BucketName);
consumeContext.Message.BucketName,
workingDirectory);
AddResizeActivity(
routingSlipBuilder,
@@ -39,19 +46,27 @@ namespace Musify.Infrastructure.Messaging.Consumers
consumeContext.Message.LargePictureWidth,
consumeContext.Message.LargePictureHeight,
consumeContext.Message.OriginalPictureKeyName,
consumeContext.Message.BucketName);
consumeContext.Message.BucketName,
workingDirectory);
routingSlipBuilder.AddActivity(
"DeleteOriginal",
BuildExecuteUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
MessagingHelper.BuildExecuteActivityUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
consumeContext.Message.BucketName,
consumeContext.Message.OriginalPictureKeyName)
);
consumeContext.Message.OriginalPictureKeyName));
routingSlipBuilder.AddActivity(
"UploadFiles",
MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName),
new TransferFilesToBucketArguments(
workingDirectory,
consumeContext.Message.BucketName,
folderName));
routingSlipBuilder.AddActivity(
"UpdatePlayListPicture",
BuildExecuteUri(UpdatePlayListPictureActivity.ExecuteEndpointName),
MessagingHelper.BuildExecuteActivityUri(UpdatePlayListPictureActivity.ExecuteEndpointName),
new UpdatePlayListPictureArguments(
consumeContext.Message.PlayListId,
consumeContext.Message.SmallPictureKeyName,
@@ -69,20 +84,32 @@ namespace Musify.Infrastructure.Messaging.Consumers
int width,
int height,
string originalPictureKeyName,
string bucketName)
string bucketName,
string workingDirectory)
{
var fileName = Path.GetFileName(originalPictureKeyName);
var sourceFilePath = Path.Combine(workingDirectory, fileName);
var destinationFileName = Path.GetFileNameWithoutExtension(destinationKeyName) + Path.GetExtension(fileName);
var destinationFilePath = Path.Combine(workingDirectory, destinationFileName);
// First: Download original file
routingSlipBuilder.AddActivity(
activityName,
BuildExecuteUri(ResizePictureActivity.ExecuteEndpointName),
new ResizePictureArgument(
$"{activityName}_Download",
MessagingHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName),
new DownloadFileFromBucketArguments(
bucketName,
originalPictureKeyName,
bucketName,
destinationKeyName,
workingDirectory));
// Second: Resize picture
routingSlipBuilder.AddActivity(
activityName,
MessagingHelper.BuildExecuteActivityUri(ResizePictureActivity.ExecuteEndpointName),
new ResizePictureLocalArguments(
sourceFilePath,
destinationFilePath,
width,
height));
}
static Uri BuildExecuteUri(string endpointName) => new($"queue:{endpointName}_execute");
}
}
@@ -1,18 +1,31 @@
using MassTransit;
using Musify.Application.Events;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.Messaging.Activities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
namespace Musify.Infrastructure.Messaging.Consumers
{
public class UpdateTrackPictureConsumer(IBus bus)
public class UpdateTrackPictureConsumer(
IBus bus,
AudioTranscoderConfiguration audioTranscoderConfiguration)
: IConsumer<UpdateTrackPictureEvent>
{
public const string QueueName = "update-track-picture";
public const string QueueName = "Update-Track-Picture";
public async Task Consume(ConsumeContext<UpdateTrackPictureEvent> consumeContext)
{
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
var folderName = Guid.NewGuid().ToString();
var workingDirectory = Path.Combine(audioTranscoderConfiguration.Routes.WorkingDirectory, folderName);
routingSlipBuilder.AddActivity(
"DownloadFile",
MessagingHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName),
new DownloadFileFromBucketArguments(
consumeContext.Message.SourceBucketName,
consumeContext.Message.SourceKeyName,
workingDirectory));
AddResizeActivity(
routingSlipBuilder,
@@ -20,38 +33,41 @@ namespace Musify.Infrastructure.Messaging.Consumers
consumeContext.Message.SmallPictureKeyName,
consumeContext.Message.SmallPictureWidth,
consumeContext.Message.SmallPictureHeight,
consumeContext.Message.OriginalPictureKeyName,
consumeContext.Message.BucketName);
consumeContext.Message.SourceKeyName,
consumeContext.Message.SourceBucketName,
workingDirectory);
AddResizeActivity(
routingSlipBuilder,
"ResizeMedium",
consumeContext.Message.SmallPictureKeyName,
consumeContext.Message.MediumPictureKeyName,
consumeContext.Message.MediumPictureWidth,
consumeContext.Message.MediumPictureHeight,
consumeContext.Message.OriginalPictureKeyName,
consumeContext.Message.BucketName);
consumeContext.Message.SourceKeyName,
consumeContext.Message.SourceBucketName,
workingDirectory);
AddResizeActivity(
routingSlipBuilder,
"ResizeMedium",
consumeContext.Message.SmallPictureKeyName,
"ResizeLarge",
consumeContext.Message.LargePictureKeyName,
consumeContext.Message.LargePictureWidth,
consumeContext.Message.LargePictureHeight,
consumeContext.Message.OriginalPictureKeyName,
consumeContext.Message.BucketName);
consumeContext.Message.SourceKeyName,
consumeContext.Message.SourceBucketName,
workingDirectory);
routingSlipBuilder.AddActivity(
"DeleteOriginal",
BuildExecuteUri(RemoveFileFromBucketActivity.ExecuteEndpointName),
new RemoveFileFromBucketArguments(
consumeContext.Message.BucketName,
consumeContext.Message.OriginalPictureKeyName)
);
"UploadFiles",
MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName),
new TransferFilesToBucketArguments(
workingDirectory,
consumeContext.Message.SourceBucketName,
folderName));
routingSlipBuilder.AddActivity(
"UpdateTrackPictureActivity",
BuildExecuteUri(UpdateTrackPictureActivity.ExecuteEndpointName),
"UpdateTrackPicture",
MessagingHelper.BuildExecuteActivityUri(UpdateTrackPictureActivity.ExecuteEndpointName),
new UpdateTrackPictureArguments(
consumeContext.Message.TrackId,
consumeContext.Message.SmallPictureKeyName,
@@ -69,20 +85,32 @@ namespace Musify.Infrastructure.Messaging.Consumers
int width,
int height,
string originalPictureKeyName,
string bucketName)
string bucketName,
string workingDirectory)
{
var fileName = Path.GetFileName(originalPictureKeyName);
var sourceFilePath = Path.Combine(workingDirectory, fileName);
var destinationFileName = Path.GetFileNameWithoutExtension(destinationKeyName) + Path.GetExtension(fileName);
var destinationFilePath = Path.Combine(workingDirectory, destinationFileName);
// First: Download original file
routingSlipBuilder.AddActivity(
activityName,
BuildExecuteUri(ResizePictureActivity.ExecuteEndpointName),
new ResizePictureArgument(
$"{activityName}_Download",
MessagingHelper.BuildExecuteActivityUri(DownloadFileFromBucketActivity.ExecuteEndpointName),
new DownloadFileFromBucketArguments(
bucketName,
originalPictureKeyName,
bucketName,
destinationKeyName,
workingDirectory));
// Second: Resize picture
routingSlipBuilder.AddActivity(
activityName,
MessagingHelper.BuildExecuteActivityUri(ResizePictureActivity.ExecuteEndpointName),
new ResizePictureLocalArguments(
sourceFilePath,
destinationFilePath,
width,
height));
}
static Uri BuildExecuteUri(string endpointName) => new($"queue:{endpointName}_execute");
}
}
@@ -3,7 +3,6 @@ using Microsoft.Extensions.DependencyInjection;
using Musify.Infrastructure.Configuration;
using Musify.Infrastructure.Messaging.Activities;
using Musify.Infrastructure.Messaging.Activities.Arguments;
using Musify.Infrastructure.Messaging.Activities.Logs;
using Musify.Infrastructure.Messaging.Consumers;
using Musify.Infrastructure.Messaging.Filters;
@@ -19,7 +18,7 @@ namespace Musify.Infrastructure.Messaging
ConfigureRabbitMqHost(
busFactoryConfigurator,
busRegistrationContext.GetRequiredService<MessagingConfiguration>()));
} );
});
return serviceDescriptors;
}
@@ -28,17 +27,20 @@ namespace Musify.Infrastructure.Messaging
{
serviceDescriptors.AddMassTransit(options =>
{
// Register consumers
options.AddConsumer<UpdatePlayListPictureConsumer>();
options.AddConsumer<RemoveFileConsumer>();
options.AddConsumer<TranscodeAudioFromTrackConsumer>();
options.AddConsumer<UpdateTrackPictureConsumer>();
options.AddExecuteActivity<ResizePictureActivity, ResizePictureArgument>();
// Register execute activities
options.AddExecuteActivity<ResizePictureActivity, ResizePictureLocalArguments>();
options.AddExecuteActivity<RemoveFileFromBucketActivity, RemoveFileFromBucketArguments>();
options.AddExecuteActivity<UpdatePlayListPictureActivity, UpdatePlayListPictureArguments>();
options.AddExecuteActivity<UpdateTrackPictureActivity, UpdateTrackPictureArguments>();
options.AddExecuteActivity<DownloadFileFromBucketActivity, DownloadFileFromBucketArguments>();
options.AddExecuteActivity<TranscodeDashAudioActivity, TranscodeDashAudioArguments>();
options.AddExecuteActivity<TransferFilesToBucket, TransferFilesToBucketArguments>();
options.UsingRabbitMq((busRegistrationContext, busFactoryConfigurator) =>
{
@@ -48,27 +50,29 @@ namespace Musify.Infrastructure.Messaging
busFactoryConfigurator.UseConsumeFilter(typeof(ProcessTrackingConsumeFilter<>), busRegistrationContext);
// Consumer endpoints
busFactoryConfigurator.ReceiveEndpoint(UpdatePlayListPictureConsumer.QueueName, endpointConfigurator =>
{
endpointConfigurator.ConfigureConsumer<UpdatePlayListPictureConsumer>(busRegistrationContext);
} );
});
busFactoryConfigurator.ReceiveEndpoint(RemoveFileConsumer.QueueName, endpointConfigurator =>
{
endpointConfigurator.ConfigureConsumer<RemoveFileConsumer>(busRegistrationContext);
} );
});
busFactoryConfigurator.ReceiveEndpoint(TranscodeAudioFromTrackConsumer.QueueName, endpointConfigurator =>
{
endpointConfigurator.ConfigureConsumer<TranscodeAudioFromTrackConsumer>(busRegistrationContext);
} );
});
busFactoryConfigurator.ReceiveEndpoint(UpdateTrackPictureConsumer.QueueName, endpointConfigurator =>
{
endpointConfigurator.ConfigureConsumer<UpdateTrackPictureConsumer>(busRegistrationContext);
} );
});
ConfigureExecuteActivityEndpoint<ResizePictureActivity, ResizePictureArgument>(
// Activity endpoints
ConfigureExecuteActivityEndpoint<ResizePictureActivity, ResizePictureLocalArguments>(
busFactoryConfigurator,
busRegistrationContext,
ResizePictureActivity.ExecuteEndpointName);
@@ -97,8 +101,13 @@ namespace Musify.Infrastructure.Messaging
busFactoryConfigurator,
busRegistrationContext,
TranscodeDashAudioActivity.ExecuteEndpointName);
} );
} );
ConfigureExecuteActivityEndpoint<TransferFilesToBucket, TransferFilesToBucketArguments>(
busFactoryConfigurator,
busRegistrationContext,
TransferFilesToBucket.ExecuteEndpointName);
});
});
return serviceDescriptors;
}
@@ -111,7 +120,7 @@ namespace Musify.Infrastructure.Messaging
{
options.Username(massTransitConfiguration.Username);
options.Password(massTransitConfiguration.Password);
} );
});
}
static void ConfigureExecuteActivityEndpoint<TActivity, TArguments>(
@@ -124,7 +133,7 @@ namespace Musify.Infrastructure.Messaging
busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_execute", endpointConfigurator =>
{
endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(busRegistrationContext);
} );
});
}
static void ConfigureActivityEndpoints<TActivity, TArguments, TLog>(
@@ -138,12 +147,12 @@ namespace Musify.Infrastructure.Messaging
busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_execute", endpointConfigurator =>
{
endpointConfigurator.ExecuteActivityHost<TActivity, TArguments>(busRegistrationContext);
} );
});
busFactoryConfigurator.ReceiveEndpoint($"{endpointName}_compensate", endpointConfigurator =>
{
endpointConfigurator.CompensateActivityHost<TActivity, TLog>(busRegistrationContext);
} );
});
}
}
}