diff --git a/ECMJobRunner.Application/Behaviors/CheckQueryExecutionBehavior.cs b/ECMJobRunner.Application/Behaviors/CheckQueryExecutionBehavior.cs
new file mode 100644
index 0000000..14a8083
--- /dev/null
+++ b/ECMJobRunner.Application/Behaviors/CheckQueryExecutionBehavior.cs
@@ -0,0 +1,116 @@
+using ECMJobRunner.Application.Common.Constants;
+using ECMJobRunner.Application.Common.Dtos;
+using ECMJobRunner.Application.Common.Exceptions;
+using ECMJobRunner.Domain.Interfaces;
+using ECMJobRunner.Application.Common.Options;
+using ECMJobRunner.Application.DEXJob;
+using MediatR;
+using Microsoft.Extensions.Options;
+using System;
+using System.Text.RegularExpressions;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace ECMJobRunner.Application.Behaviors
+{
+ ///
+ /// Pipeline behavior that executes the check SQL query for DEX jobs
+ /// Runs after MainQueryExecutionBehavior and validates query results
+ ///
+ /// The request type
+ /// The response type
+ public class CheckQueryExecutionBehavior : IPipelineBehavior
+ where TRequest : notnull
+ {
+ private readonly ISQLExecutor _executor;
+ private readonly DexJobOptions _options;
+
+ ///
+ /// Initializes a new instance of CheckQueryExecutionBehavior
+ ///
+ /// SQL executor for query execution
+ /// DEX job configuration options
+ public CheckQueryExecutionBehavior(ISQLExecutor executor, IOptions options)
+ {
+ _executor = executor;
+ _options = options.Value;
+ }
+
+ ///
+ /// Handles the pipeline behavior
+ /// Executes check query if request is TriggeringDEXJobCommand
+ ///
+#if NET48
+ public async Task Handle(TRequest request, CancellationToken cancellationToken, RequestHandlerDelegate next)
+#else
+ public async Task Handle(TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken)
+#endif
+ {
+ if (request is TriggeringDEXJobCommand command)
+ {
+ await ExecuteCheckQueryAsync(command, cancellationToken);
+ }
+
+ return await next();
+ }
+
+ private async Task ExecuteCheckQueryAsync(TriggeringDEXJobCommand command, CancellationToken cancel)
+ {
+ if (!string.IsNullOrWhiteSpace(command.Job.SqlCheckQuery))
+ {
+ var sqlCheckQuery = Regex.Replace(
+ command.Job.SqlCheckQuery!,
+ _options.Placeholders.BatchId.Pattern,
+ command.BatchId,
+ _options.Placeholders.BatchId.RegexOptions);
+
+ try
+ {
+ var result = await _executor.ExecuteQueryAsync(sqlCheckQuery, cancel);
+ if (_options.Error.CheckQuery.OnUnexpectedResult == ErrorAction.Stop)
+ {
+ if (result is null)
+ throw new JobSqlException(
+ jobName: "Triggering DEX",
+ processName: "Check Query",
+ batchId: command.BatchId,
+ reason: "Check Query returned nothing.",
+ query: sqlCheckQuery,
+ innerException: null);
+ else if (result.ReturnValue <= 0)
+ throw new JobSqlException(
+ jobName: "Triggering DEX",
+ processName: "Check Query",
+ batchId: command.BatchId,
+ reason: $"The query unexpectedly returned the value {result.ReturnValue}. The expected value was any value greater than 0.",
+ query: sqlCheckQuery,
+ innerException: null);
+ }
+ }
+ catch (Exception ex)
+ {
+ if (_options.Error.CheckQuery.OnExecution == ErrorAction.Stop)
+ throw new JobSqlException(
+ jobName: "Triggering DEX",
+ processName: "Check Query",
+ batchId: command.BatchId,
+ reason: null,
+ query: sqlCheckQuery,
+ innerException: ex
+ );
+ }
+ }
+ else if (_options.Error.CheckQuery.IfNullOrWhiteSpace == ErrorAction.Stop)
+ {
+ throw new JobSqlException(
+ jobName:"Triggering DEX",
+ processName:"Check Query",
+ batchId:command.BatchId,
+ reason: "SQL Check Query is null or empty",
+ query: null,
+ innerException: null
+ );
+ }
+ }
+ }
+}
diff --git a/ECMJobRunner.Application/Behaviors/MainQueryExecutionBehavior.cs b/ECMJobRunner.Application/Behaviors/MainQueryExecutionBehavior.cs
new file mode 100644
index 0000000..c6b12e6
--- /dev/null
+++ b/ECMJobRunner.Application/Behaviors/MainQueryExecutionBehavior.cs
@@ -0,0 +1,113 @@
+using ECMJobRunner.Application.Common.Constants;
+using ECMJobRunner.Application.Common.Dtos;
+using ECMJobRunner.Application.Common.Exceptions;
+using ECMJobRunner.Domain.Interfaces;
+using ECMJobRunner.Application.Common.Options;
+using ECMJobRunner.Application.DEXJob;
+using MediatR;
+using Microsoft.Extensions.Options;
+using System;
+using System.Text.RegularExpressions;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace ECMJobRunner.Application.Behaviors
+{
+ ///
+ /// Pipeline behavior that executes the main SQL query for DEX jobs
+ /// Runs before the command handler and validates query results
+ ///
+ /// The request type
+ /// The response type
+ public class MainQueryExecutionBehavior : IPipelineBehavior
+ where TRequest : notnull
+ {
+ private readonly ISQLExecutor _executor;
+ private readonly DexJobOptions _options;
+
+ ///
+ /// Initializes a new instance of MainQueryExecutionBehavior
+ ///
+ /// SQL executor for query execution
+ /// DEX job configuration options
+ public MainQueryExecutionBehavior(ISQLExecutor executor, IOptions options)
+ {
+ _executor = executor;
+ _options = options.Value;
+ }
+
+ ///
+ /// Handles the pipeline behavior
+ /// Executes main query if request is TriggeringDEXJobCommand
+ ///
+#if NET48
+ public async Task Handle(TRequest request, CancellationToken cancellationToken, RequestHandlerDelegate next)
+#else
+ public async Task Handle(TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken)
+#endif
+ {
+ if (request is TriggeringDEXJobCommand command)
+ {
+ await ExecuteMainQueryAsync(command, cancellationToken);
+ }
+
+ return await next();
+ }
+
+ private async Task ExecuteMainQueryAsync(TriggeringDEXJobCommand command, CancellationToken cancel)
+ {
+ if (!string.IsNullOrWhiteSpace(command.Job.SqlMainQuery))
+ {
+ var sqlMainQuery = Regex.Replace(
+ command.Job.SqlMainQuery!,
+ _options.Placeholders.BatchId.Pattern,
+ command.BatchId,
+ _options.Placeholders.BatchId.RegexOptions);
+
+ try
+ {
+ var result = await _executor.ExecuteQueryAsync(sqlMainQuery, cancel);
+ if (_options.Error.MainQuery.OnUnexpectedResult == ErrorAction.Stop)
+ {
+ if (result is null)
+ throw new JobSqlException(
+ jobName: "Triggering DEX",
+ processName: "Main Query",
+ batchId: command.BatchId,
+ reason: "Main Query returned nothing.",
+ query: sqlMainQuery,
+ innerException: null);
+ else if (result.ReturnValue is not null)
+ throw new JobSqlException(
+ jobName: "Triggering DEX",
+ processName: "Main Query",
+ batchId: command.BatchId,
+ reason: $"The query unexpectedly returned the value {result.ReturnValue}. The expected value was null.",
+ query: sqlMainQuery, innerException: null);
+ }
+ }
+ catch (Exception ex)
+ {
+ if (_options.Error.MainQuery.OnExecution == ErrorAction.Stop)
+ throw new JobSqlException(
+ jobName: "Triggering DEX",
+ processName: "Main Query",
+ batchId: command.BatchId,
+ reason: null,
+ query: sqlMainQuery,
+ innerException: ex);
+ }
+ }
+ else if (_options.Error.MainQuery.IfNullOrWhiteSpace == ErrorAction.Stop)
+ {
+ throw new JobSqlException(
+ jobName: "Triggering DEX",
+ processName: "Main Query",
+ batchId: command.BatchId,
+ reason: "SQL Check Query is null or empty",
+ query: null,
+ innerException: null);
+ }
+ }
+ }
+}
diff --git a/ECMJobRunner.Application/Behaviors/ReCRequestExecutionBehavior.cs b/ECMJobRunner.Application/Behaviors/ReCRequestExecutionBehavior.cs
new file mode 100644
index 0000000..26decd3
--- /dev/null
+++ b/ECMJobRunner.Application/Behaviors/ReCRequestExecutionBehavior.cs
@@ -0,0 +1,81 @@
+using ECMJobRunner.Application.Common.Constants;
+using ECMJobRunner.Application.Common.Exceptions;
+using ECMJobRunner.Application.Common.Options;
+using ECMJobRunner.Application.DEXJob;
+using MediatR;
+using Microsoft.Extensions.Options;
+using ReC.Client;
+using ReC.Client.Api;
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace ECMJobRunner.Application.Behaviors
+{
+ ///
+ /// Pipeline behavior that sends ReC HTTP request for DEX jobs
+ /// Runs after CheckQueryExecutionBehavior and invokes ReC API
+ ///
+ /// The request type
+ /// The response type
+ public class ReCRequestExecutionBehavior : IPipelineBehavior
+ where TRequest : notnull
+ {
+ private readonly ReCClient _reCClient;
+ private readonly DexJobOptions _options;
+
+ ///
+ /// Initializes a new instance of ReCRequestExecutionBehavior
+ ///
+ /// ReC client for HTTP requests
+ /// DEX job configuration options
+ public ReCRequestExecutionBehavior(ReCClient reCClient, IOptions options)
+ {
+ _reCClient = reCClient;
+ _options = options.Value;
+ }
+
+ ///
+ /// Handles the pipeline behavior
+ /// Sends ReC request if request is TriggeringDEXJobCommand
+ ///
+#if NET48
+ public async Task Handle(TRequest request, CancellationToken cancellationToken, RequestHandlerDelegate next)
+#else
+ public async Task Handle(TRequest request, RequestHandlerDelegate next, CancellationToken cancellationToken)
+#endif
+ {
+ if (request is TriggeringDEXJobCommand command)
+ {
+ await SendReCRequestAsync(command, cancellationToken);
+ }
+
+ return await next();
+ }
+
+ private async Task SendReCRequestAsync(TriggeringDEXJobCommand command, CancellationToken cancel)
+ {
+ try
+ {
+ await _reCClient.RecActions.InvokeAsync(command.Job.ProfileId, new InvokeReferences()
+ {
+ BatchId = command.BatchId,
+ }, cancel);
+ }
+ catch (Exception ex)
+ {
+ if (_options.Error.ReCRequest.OnSending == ErrorAction.Stop)
+ {
+ throw new JobHttpException(
+ jobName: "Triggering DEX",
+ processName: "ReC Http Request",
+ batchId: command.BatchId,
+ reason: null,
+ clientLibrary: "ReC.Client",
+ clientMethod: "RecActions.InvokeAsync",
+ innerException: ex);
+ }
+ }
+ }
+ }
+}
diff --git a/ECMJobRunner.Application/DEXJob/TriggeringDEXJobCommand.cs b/ECMJobRunner.Application/DEXJob/TriggeringDEXJobCommand.cs
index 6ba5c1e..f28053c 100644
--- a/ECMJobRunner.Application/DEXJob/TriggeringDEXJobCommand.cs
+++ b/ECMJobRunner.Application/DEXJob/TriggeringDEXJobCommand.cs
@@ -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
{
///
/// 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
///
public class TriggeringDEXJobCommand : IRequest
{
@@ -34,98 +27,25 @@ namespace ECMJobRunner.Application.DEXJob
///
/// 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
///
- public class TriggeringDEXJobCommandHandler(
- ISQLExecutor executor,
- IOptions options,
- ReCClient reCClient)
- : IRequestHandler
+ public class TriggeringDEXJobCommandHandler : IRequestHandler
{
- private readonly DexJobOptions _options = options.Value;
-
///
/// Handles the TriggeringDEXJobCommand
- /// Executes main query, check query, and ReC request according to configuration
+ /// Returns immediately as behaviors perform all work
///
/// The command containing job and batch ID
- /// Cancellation token
+ /// Cancellation token
/// Unit value indicating completion
- public async Task Handle(TriggeringDEXJobCommand request, CancellationToken cancel)
+ public Task 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(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(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);
}
}
}