Use variables

This commit is contained in:
2026-03-23 16:28:02 +01:00
parent 773a4e727c
commit 0b471db6aa
8 changed files with 35 additions and 28 deletions
@@ -1,4 +1,4 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments namespace Musify.Infrastructure.Messaging.Activities.Arguments
{ {
public record TranscodeDashAudioArguments(string SourceFilePath, string DestinationFolderName); public record TranscodeDashAudioArguments();
} }
@@ -1,7 +1,6 @@
namespace Musify.Infrastructure.Messaging.Activities.Arguments namespace Musify.Infrastructure.Messaging.Activities.Arguments
{ {
public record TransferFilesToBucketArguments( public record TransferFilesToBucketArguments(
string FolderPath,
string DestinationBucketName, string DestinationBucketName,
string DestinationKeyName); string DestinationKeyName);
} }
@@ -53,17 +53,23 @@ namespace Musify.Infrastructure.Messaging.Activities
if (fileStream.CanSeek) if (fileStream.CanSeek)
fileStream.Position = 0; fileStream.Position = 0;
var destinationPath = Path.Combine( var fileName = Path.GetFileName(executeContext.Arguments.KeyName);
audioTranscoderConfiguration.Routes.WorkingDirectory, var folderName = Guid.NewGuid().ToString();
Path.GetFileName(executeContext.Arguments.KeyName)); var workingDirectory = Path.Combine(executeContext.Arguments.DestinationPath, folderName);
var destinationPath = Path.Combine(workingDirectory, fileName);
Directory.CreateDirectory(Path.GetDirectoryName(destinationPath)!); Directory.CreateDirectory(workingDirectory);
using var destinationStream = File.Create(destinationPath); using var destinationStream = File.Create(destinationPath);
await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken); await fileStream.CopyToAsync(destinationStream, executeContext.CancellationToken);
await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken); await processTrackingStore.CompleteStepAsync(processId, stepId, executeContext.CancellationToken);
return executeContext.Completed(); return executeContext.CompletedWithVariables(new
{
SourceFilePath = destinationPath,
WorkingDirectory = workingDirectory,
DestinationFolderName = folderName
} );
} }
catch (Exception exception) catch (Exception exception)
{ {
@@ -32,17 +32,21 @@ namespace Musify.Infrastructure.Messaging.Activities
0, 0,
executeContext.CancellationToken); executeContext.CancellationToken);
var sourceFilePath = executeContext.GetVariable<string>("SourceFilePath");
var destinationFolderName = executeContext.GetVariable<string>("DestinationFolderName");
if (string.IsNullOrWhiteSpace(sourceFilePath) || string.IsNullOrWhiteSpace(destinationFolderName))
{
throw new InvalidOperationException("Transcode activity requires SourceFilePath and DestinationFolderName variables.");
}
try try
{ {
var workingDirectory = Path.Combine( await using var fileStream = File.OpenRead(sourceFilePath);
audioTranscoderConfiguration.Routes.WorkingDirectory,
executeContext.Arguments.DestinationFolderName);
await using var fileStream = File.OpenRead(executeContext.Arguments.SourceFilePath);
var result = await audioTranscoder.TranscodeToDashAsync( var result = await audioTranscoder.TranscodeToDashAsync(
fileStream, fileStream,
executeContext.Arguments.DestinationFolderName, destinationFolderName,
audioTranscoderConfiguration.TranscodingTimeout, audioTranscoderConfiguration.TranscodingTimeout,
executeContext.CancellationToken); executeContext.CancellationToken);
@@ -50,7 +54,7 @@ namespace Musify.Infrastructure.Messaging.Activities
{ {
var errorMessage = string.Join("; ", result.Errors); var errorMessage = string.Join("; ", result.Errors);
logger.LogError("Transcoding failed for file {SourceFilePath}. Errors: {ErrorMessage}", logger.LogError("Transcoding failed for file {SourceFilePath}. Errors: {ErrorMessage}",
executeContext.Arguments.SourceFilePath, errorMessage); sourceFilePath, errorMessage);
throw new Exception(errorMessage); throw new Exception(errorMessage);
} }
@@ -60,7 +64,7 @@ namespace Musify.Infrastructure.Messaging.Activities
catch (Exception exception) catch (Exception exception)
{ {
logger.LogError(exception, "Error during transcoding activity for file {SourceFilePath}", logger.LogError(exception, "Error during transcoding activity for file {SourceFilePath}",
executeContext.Arguments.SourceFilePath); sourceFilePath);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken); await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw; throw;
} }
@@ -30,10 +30,16 @@ namespace Musify.Infrastructure.Messaging.Activities
0, 0,
executeContext.CancellationToken); executeContext.CancellationToken);
var folderPath = executeContext.GetVariable<string>("WorkingDirectory");
if (string.IsNullOrWhiteSpace(folderPath))
{
throw new InvalidOperationException("Transfer activity requires WorkingDirectory variable.");
}
try try
{ {
var transferFilesResult = await storageHandler.TransferFilesAsync( var transferFilesResult = await storageHandler.TransferFilesAsync(
executeContext.Arguments.FolderPath, folderPath,
executeContext.Arguments.DestinationBucketName, executeContext.Arguments.DestinationBucketName,
executeContext.Arguments.DestinationKeyName, executeContext.Arguments.DestinationKeyName,
executeContext.CancellationToken); executeContext.CancellationToken);
@@ -42,7 +48,7 @@ namespace Musify.Infrastructure.Messaging.Activities
{ {
var errorMessage = string.Join("; ", transferFilesResult.Errors); var errorMessage = string.Join("; ", transferFilesResult.Errors);
logger.LogError("Failed to transfer files from {FolderPath} to bucket {DestinationBucketName}, key {DestinationKeyName}. Errors: {Errors}", logger.LogError("Failed to transfer files from {FolderPath} to bucket {DestinationBucketName}, key {DestinationKeyName}. Errors: {Errors}",
executeContext.Arguments.FolderPath, folderPath,
executeContext.Arguments.DestinationBucketName, executeContext.Arguments.DestinationBucketName,
executeContext.Arguments.DestinationKeyName, executeContext.Arguments.DestinationKeyName,
errorMessage); errorMessage);
@@ -55,7 +61,7 @@ namespace Musify.Infrastructure.Messaging.Activities
catch (Exception exception) catch (Exception exception)
{ {
logger.LogError(exception, "Error transferring files from {FolderPath} to bucket {DestinationBucketName}", logger.LogError(exception, "Error transferring files from {FolderPath} to bucket {DestinationBucketName}",
executeContext.Arguments.FolderPath, folderPath,
executeContext.Arguments.DestinationBucketName); executeContext.Arguments.DestinationBucketName);
await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken); await processTrackingStore.FailStepAsync(processId, stepId, exception.Message, executeContext.CancellationToken);
throw; throw;
@@ -16,9 +16,6 @@ namespace Musify.Infrastructure.Messaging.Consumers
public async Task Consume(ConsumeContext<TranscodeAudioFromTrackEvent> consumeContext) public async Task Consume(ConsumeContext<TranscodeAudioFromTrackEvent> consumeContext)
{ {
var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid()); var routingSlipBuilder = new RoutingSlipBuilder(NewId.NextGuid());
var fileName = Path.GetFileName(consumeContext.Message.SourceKeyName);
var folderName = Guid.NewGuid().ToString();
var workingDirectory = Path.Combine(audioTranscoderConfiguration.Routes.WorkingDirectory, folderName);
routingSlipBuilder.AddActivity( routingSlipBuilder.AddActivity(
"DownloadFile", "DownloadFile",
@@ -26,20 +23,17 @@ namespace Musify.Infrastructure.Messaging.Consumers
new DownloadFileFromBucketArguments( new DownloadFileFromBucketArguments(
consumeContext.Message.SourceBucketName, consumeContext.Message.SourceBucketName,
consumeContext.Message.SourceKeyName, consumeContext.Message.SourceKeyName,
workingDirectory)); audioTranscoderConfiguration.Routes.WorkingDirectory));
routingSlipBuilder.AddActivity( routingSlipBuilder.AddActivity(
"TranscodeAudio", "TranscodeAudio",
MessagingHelper.BuildExecuteActivityUri(TranscodeDashAudioActivity.ExecuteEndpointName), MessagingHelper.BuildExecuteActivityUri(TranscodeDashAudioActivity.ExecuteEndpointName),
new TranscodeDashAudioArguments( new TranscodeDashAudioArguments());
Path.Combine(workingDirectory, fileName),
folderName));
routingSlipBuilder.AddActivity( routingSlipBuilder.AddActivity(
"UploadFile", "UploadFile",
MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName), MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName),
new TransferFilesToBucketArguments( new TransferFilesToBucketArguments(
workingDirectory,
consumeContext.Message.DestinationBucketName, consumeContext.Message.DestinationBucketName,
consumeContext.Message.DestinationKeyName)); consumeContext.Message.DestinationKeyName));
@@ -68,7 +68,6 @@ namespace Musify.Infrastructure.Messaging.Consumers
"UploadFiles", "UploadFiles",
MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName), MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName),
new TransferFilesToBucketArguments( new TransferFilesToBucketArguments(
workingDirectory,
consumeContext.Message.SourceBucketName, consumeContext.Message.SourceBucketName,
folderName)); folderName));
@@ -68,7 +68,6 @@ namespace Musify.Infrastructure.Messaging.Consumers
"UploadFiles", "UploadFiles",
MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName), MessagingHelper.BuildExecuteActivityUri(TransferFilesToBucket.ExecuteEndpointName),
new TransferFilesToBucketArguments( new TransferFilesToBucketArguments(
workingDirectory,
consumeContext.Message.SourceBucketName, consumeContext.Message.SourceBucketName,
folderName)); folderName));