feat(application): implement MediatR pipeline behaviors for DEX job stages
- Add MainQueryExecutionBehavior - Execute main SQL query via ISQLExecutor - Populate command.MainQueryResults - Throw JobSqlException on failure - Add CheckQueryExecutionBehavior - Execute check SQL query via ISQLExecutor - Validate ErrorAction.SkipInsert for zero results - Throw JobSqlException on failure/validation error - Add ReCRequestExecutionBehavior - Execute ReC API requests for each main query result - Batch ID tracking, error handling - Throw JobHttpException on API failures - Conditional compilation for MediatR signature differences - net480: Handle(request, cancellationToken, next) - net8.0: Handle(request, next, cancellationToken) - Refactor TriggeringDEXJobCommand handler: delegate all logic to behaviors
This commit is contained in:
@@ -1,15 +1,5 @@
|
||||
using ECMJobRunner.Application.Common.Constants;
|
||||
using ECMJobRunner.Application.Common.Dtos;
|
||||
using ECMJobRunner.Application.Common.Exceptions;
|
||||
using ECMJobRunner.Application.Common.Interfaces;
|
||||
using ECMJobRunner.Application.Common.Options;
|
||||
using ECMJobRunner.Domain.Entities;
|
||||
using ECMJobRunner.Domain.Entities;
|
||||
using MediatR;
|
||||
using Microsoft.Extensions.Options;
|
||||
using ReC.Client;
|
||||
using ReC.Client.Api;
|
||||
using System;
|
||||
using System.Text.RegularExpressions;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
|
||||
@@ -17,7 +7,10 @@ namespace ECMJobRunner.Application.DEXJob
|
||||
{
|
||||
/// <summary>
|
||||
/// Command to trigger a single DEX job execution
|
||||
/// Executes main query, check query, and ReC request in sequence
|
||||
/// Execution logic is handled by pipeline behaviors:
|
||||
/// 1. MainQueryExecutionBehavior - executes main SQL query
|
||||
/// 2. CheckQueryExecutionBehavior - validates with check SQL query
|
||||
/// 3. ReCRequestExecutionBehavior - invokes ReC HTTP request
|
||||
/// </summary>
|
||||
public class TriggeringDEXJobCommand : IRequest<Unit>
|
||||
{
|
||||
@@ -34,98 +27,25 @@ namespace ECMJobRunner.Application.DEXJob
|
||||
|
||||
/// <summary>
|
||||
/// Handler for TriggeringDEXJobCommand
|
||||
/// Orchestrates execution of main query, check query, and ReC HTTP request
|
||||
/// All execution logic is delegated to pipeline behaviors
|
||||
/// This handler simply returns completion after behaviors execute
|
||||
/// </summary>
|
||||
public class TriggeringDEXJobCommandHandler(
|
||||
ISQLExecutor executor,
|
||||
IOptions<DexJobOptions> options,
|
||||
ReCClient reCClient)
|
||||
: IRequestHandler<TriggeringDEXJobCommand, Unit>
|
||||
public class TriggeringDEXJobCommandHandler : IRequestHandler<TriggeringDEXJobCommand, Unit>
|
||||
{
|
||||
private readonly DexJobOptions _options = options.Value;
|
||||
|
||||
/// <summary>
|
||||
/// Handles the TriggeringDEXJobCommand
|
||||
/// Executes main query, check query, and ReC request according to configuration
|
||||
/// Returns immediately as behaviors perform all work
|
||||
/// </summary>
|
||||
/// <param name="request">The command containing job and batch ID</param>
|
||||
/// <param name="cancel">Cancellation token</param>
|
||||
/// <param name="cancellationToken">Cancellation token</param>
|
||||
/// <returns>Unit value indicating completion</returns>
|
||||
public async Task<Unit> Handle(TriggeringDEXJobCommand request, CancellationToken cancel)
|
||||
public Task<Unit> Handle(TriggeringDEXJobCommand request, CancellationToken cancellationToken)
|
||||
{
|
||||
#region Main Query
|
||||
if (!string.IsNullOrWhiteSpace(request.Job.SqlMainQuery))
|
||||
{
|
||||
var sqlMainQuery = Regex.Replace(
|
||||
request.Job.SqlMainQuery!,
|
||||
_options.Placeholders.BatchId.Pattern,
|
||||
request.BatchId,
|
||||
_options.Placeholders.BatchId.RegexOptions);
|
||||
|
||||
try
|
||||
{
|
||||
var result = await executor.ExecuteQueryAsync<MainQueryResult>(sqlMainQuery, cancel);
|
||||
if (_options.Error.MainQuery.OnUnexpectedResult == ErrorAction.Stop)
|
||||
{
|
||||
if (result is null)
|
||||
throw new DEXJobException("SQL Main Query", request.BatchId, sqlMainQuery, "Main Query returned nothing.");
|
||||
else if (result.ReturnValue is not null)
|
||||
throw new DEXJobException("SQL Main Query", request.BatchId, sqlMainQuery, $"The query unexpectedly returned the value {result.ReturnValue}. The expected value was null.");
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
if (_options.Error.MainQuery.OnExecution == ErrorAction.Stop)
|
||||
throw new DEXJobException("SQL Main Query", request.BatchId, sqlMainQuery, ex);
|
||||
}
|
||||
}
|
||||
else if (_options.Error.MainQuery.IfNullOrWhiteSpace == ErrorAction.Stop)
|
||||
{
|
||||
throw new DEXJobException("SQL Main Query", request.BatchId, null, "SQL Main Query is null or empty");
|
||||
}
|
||||
#endregion
|
||||
|
||||
#region Check Query
|
||||
if (!string.IsNullOrWhiteSpace(request.Job.SqlCheckQuery))
|
||||
{
|
||||
var sqlCheckQuery = Regex.Replace(
|
||||
request.Job.SqlCheckQuery!,
|
||||
_options.Placeholders.BatchId.Pattern,
|
||||
request.BatchId,
|
||||
_options.Placeholders.BatchId.RegexOptions);
|
||||
|
||||
try
|
||||
{
|
||||
var result = await executor.ExecuteQueryAsync<CheckQueryResult>(sqlCheckQuery, cancel);
|
||||
if (_options.Error.CheckQuery.OnUnexpectedResult == ErrorAction.Stop)
|
||||
{
|
||||
if (result is null)
|
||||
throw new DEXJobException("SQL Check Query", request.BatchId, sqlCheckQuery, "Check Query returned nothing.");
|
||||
else if (result.ReturnValue <= 0)
|
||||
throw new DEXJobException("SQL Check Query", request.BatchId, sqlCheckQuery, $"The query unexpectedly returned the value {result.ReturnValue}. The expected value was any value greater than 0.");
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
if (_options.Error.CheckQuery.OnExecution == ErrorAction.Stop)
|
||||
throw new DEXJobException("SQL Check Query", request.BatchId, sqlCheckQuery, ex);
|
||||
}
|
||||
}
|
||||
else if (_options.Error.CheckQuery.IfNullOrWhiteSpace == ErrorAction.Stop)
|
||||
{
|
||||
throw new DEXJobException("SQL Check Query", request.BatchId, null, "SQL Check Query is null or empty");
|
||||
}
|
||||
#endregion
|
||||
|
||||
#region ReC Request
|
||||
long profileId = request.Job.ProfileId;
|
||||
await reCClient.RecActions.InvokeAsync(profileId, new InvokeReferences()
|
||||
{
|
||||
BatchId = request.BatchId, // Fixed: Use request.BatchId instead of profileId.ToString()
|
||||
}, cancel);
|
||||
#endregion
|
||||
|
||||
return Unit.Value; // Fixed: Added missing return statement
|
||||
// All execution logic is handled by pipeline behaviors:
|
||||
// - MainQueryExecutionBehavior
|
||||
// - CheckQueryExecutionBehavior
|
||||
// - ReCRequestExecutionBehavior
|
||||
return Task.FromResult(Unit.Value);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user