diff --git a/README.md b/README.md index 6178f605..78da6bd3 100644 --- a/README.md +++ b/README.md @@ -225,6 +225,7 @@ Subscribing is configured on job sources that support it with a `SUBSCRIBE` envi | Job source | Environment variable | |------------|-----------------------------------| +| ActiveMQ | `JOB_SOURCE__ACTIVEMQ__SUBSCRIBE` | | RabbitMQ | `JOB_SOURCE__RABBITMQ__SUBSCRIBE` | #### Notes on Implementing Other Subscribe Patterns diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Configuration/CoreConfigurationService.cs b/src/RedShirt.Example.JobWorker.Core/Services/Configuration/CoreConfigurationService.cs index d7e3e9f0..f50b46f7 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Configuration/CoreConfigurationService.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Configuration/CoreConfigurationService.cs @@ -9,7 +9,12 @@ namespace RedShirt.Example.JobWorker.Core.Services.Configuration; /// public interface ICoreConfigurationService { + /// + /// Maximum number of jobs the worker should hold in backlog. + /// Callers may assume the returned value is at least 1. + /// int GetBacklogSize(); + bool IsHaltOnFailure(); /// diff --git a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs index a41863be..456a7651 100644 --- a/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs +++ b/src/RedShirt.Example.JobWorker.Core/Services/Jobs/Subscriptions/JobSubscriberIntakeQueue.cs @@ -10,7 +10,9 @@ namespace RedShirt.Example.JobWorker.Core.Services.Jobs.Subscriptions; /// In-memory handoff queue between a subscription job source and . /// Subscription sources push batches via ; the subscriber manager drains them via /// and submits each batch through job intake. -/// This interface exists because the implementation of indirectly uses +/// This subscriber queue should not be used in a non subscriber context. If the configured +/// is not a subscriber, then this queue will not be read from. +/// This interface exists because indirectly uses /// as a dependency. Creating this queue was the most expedient way to avoid a circular loop. /// public interface IJobSubscriberIntakeQueue @@ -43,7 +45,9 @@ internal class JobSubscriberIntakeQueue : IJobSubscriberIntakeQueue private readonly ConcurrentQueue _jobs = new(); private bool _done; +#pragma warning disable S2325 private void Cancel() +#pragma warning restore S2325 { _done = true; _doNotWaitIfSetEvent.Set(); @@ -54,7 +58,9 @@ public JobSubscriberIntakeQueue(IExecutionEndArbiter executionEndArbiter) executionEndArbiter.AddOnStopCallback(_ => Cancel()); } +#pragma warning disable S2325 public void Load(IJobSourceResponse jobSourceResponse) +#pragma warning disable S2325 { _jobs.Enqueue(jobSourceResponse); _doNotWaitIfSetEvent.Set(); diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Configuration/ActiveMqConfigurationModel.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Configuration/ActiveMqConfigurationModel.cs new file mode 100644 index 00000000..5c5b3047 --- /dev/null +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Configuration/ActiveMqConfigurationModel.cs @@ -0,0 +1,6 @@ +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Configuration; + +public sealed class ActiveMqConfigurationModel +{ + public required string QueueName { get; init; } +} \ No newline at end of file diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Extensions/ServiceCollectionExtensions.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Extensions/ServiceCollectionExtensions.cs index 607edbcb..b757d5b4 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Extensions/ServiceCollectionExtensions.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Extensions/ServiceCollectionExtensions.cs @@ -1,6 +1,7 @@ using Microsoft.Extensions.Configuration; using Microsoft.Extensions.DependencyInjection; using RedShirt.Example.JobWorker.Core.Services.Abstractions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Configuration; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Factories; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; @@ -9,21 +10,48 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Extensions; public static class ServiceCollectionExtensions { + private const string ConfigurationSectionName = "JobSource:ActiveMq"; + public static IServiceCollection AddActiveMqJobManagement(this IServiceCollection services, IConfigurationRoot configuration) { + var section = configuration.GetSection(ConfigurationSectionName); + + // shorthand + var useSubscribe = section.Get()?.Subscribe == true; + + if (useSubscribe) + { + services.AddSingleton(); + } + else + { + services.AddSingleton(); + } + return services // Required - .AddSingleton() .AddSingleton() // Supporting - .Configure(configuration.GetSection("JobSource:ActiveMq")) - .Configure( - configuration.GetSection("JobSource:ActiveMq")) + .AddSingleton( + new ActiveMqSubscribeConfigurationService(useSubscribe)) + .Configure(section) + .Configure(section) .AddSingleton() .AddSingleton() .AddSingleton() .AddSingleton() - .AddSingleton(); + .AddSingleton() + .AddSingleton() + .AddSingleton(); + } + + private sealed class SubscribeConfigurationModel + { +#pragma warning disable S3459 +#pragma warning disable S1144 + public required bool Subscribe { get; init; } +#pragma warning disable S1144 +#pragma warning restore S3459 } } \ No newline at end of file diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Factories/ActiveMqConnectionFactory.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Factories/ActiveMqConnectionFactory.cs index 34ad3545..382b9ec7 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Factories/ActiveMqConnectionFactory.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Factories/ActiveMqConnectionFactory.cs @@ -4,15 +4,21 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Factories; internal interface IActiveMqConnectionFactory { - Task GetConnectionAsync(CancellationToken cancellationToken = default); + Task GetConnectionAsync( + bool forceNewSecretManagerPull = false, + CancellationToken cancellationToken = default); } internal class ActiveMqConnectionFactory(IInnerActiveMqConnectionFactory innerActiveMqConnectionFactory) : IActiveMqConnectionFactory { - public async Task GetConnectionAsync(CancellationToken cancellationToken = default) + public async Task GetConnectionAsync( + bool forceNewSecretManagerPull = false, + CancellationToken cancellationToken = default) { - var wrapper = await innerActiveMqConnectionFactory.GetConnectionFactoryWrapperAsync(cancellationToken); + var wrapper = await innerActiveMqConnectionFactory.GetConnectionFactoryWrapperAsync( + forceNewSecretManagerPull, + cancellationToken); return await wrapper.CreateConnectionAsync(cancellationToken); } diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Factories/InnerActiveMqConnectionFactory.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Factories/InnerActiveMqConnectionFactory.cs index 4b276983..a525e422 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Factories/InnerActiveMqConnectionFactory.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Factories/InnerActiveMqConnectionFactory.cs @@ -1,4 +1,5 @@ using Apache.NMS.ActiveMQ; +using RedShirt.Example.JobWorker.Core.Services.Configuration; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Wrappers; @@ -7,23 +8,68 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Factories; internal interface IInnerActiveMqConnectionFactory { Task GetConnectionFactoryWrapperAsync( + bool forceNewSecretManagerPull = false, CancellationToken cancellationToken = default); } internal class InnerActiveMqConnectionFactory( - IActiveMqServerConfigurationSource configurationSource) + IActiveMqServerConfigurationSource configurationSource, + IActiveMqSubscribeConfigurationService activeMqSubscribeConfigurationService, + ICoreConfigurationService coreConfigurationService) : IInnerActiveMqConnectionFactory { public async Task GetConnectionFactoryWrapperAsync( + bool forceNewSecretManagerPull = false, CancellationToken cancellationToken = default) { - var configuration = await configurationSource.GetConfigurationAsync(cancellationToken); - var connectionFactory = new ConnectionFactory(configuration.BrokerUri) + var configuration = await configurationSource.GetConfigurationAsync( + forceNewSecretManagerPull, + cancellationToken); + + // Subscription mode uses our own reconnect/resubscribe loop (see ActiveMqSubscribeJobSource), + // which can force a fresh secret-manager pull when credentials change on top of a connection interruption. + // NMS failover transport would reconnect with the original factory credentials and fight that mechanism — strip it. + var brokerUri = activeMqSubscribeConfigurationService.IsSubscription + ? StripFailoverUri(configuration.BrokerUri) + : configuration.BrokerUri; + + var connectionFactory = new ConnectionFactory(brokerUri) { UserName = configuration.User, Password = configuration.Password }; + if (activeMqSubscribeConfigurationService.IsSubscription) + { + // Cap unacked messages pushed to the consumer (listener / receive), analogous to RabbitMQ BasicQos. + // Only do this for subscriptions, as it's not guaranteed that a user + // would set a backlog size for batch-mode polling and I don't want to worry about the weird interaction. + connectionFactory.PrefetchPolicy.QueuePrefetch = coreConfigurationService.GetBacklogSize(); + } + return new ActiveMqConnectionWrapper(connectionFactory); } + + /// + /// Removes an NMS failover: wrapper from , leaving the nested + /// broker address (first composite URI when several are listed). Plain URIs are returned unchanged. + /// + internal static string StripFailoverUri(string brokerUri) + { + if (!brokerUri.StartsWith("failover:", StringComparison.OrdinalIgnoreCase)) + { + return brokerUri; + } + + var open = brokerUri.IndexOf('('); + var close = brokerUri.LastIndexOf(')'); + if (open < 0 || close <= open) + { + return brokerUri; + } + + var nested = brokerUri[(open + 1)..close]; + var comma = nested.IndexOf(','); + return comma < 0 ? nested : nested[..comma]; + } } \ No newline at end of file diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqConfigurationSource.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqConfigurationSource.cs index ce4f12cf..ae4c66ed 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqConfigurationSource.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqConfigurationSource.cs @@ -6,7 +6,9 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; internal interface IActiveMqServerConfigurationSource { - Task GetConfigurationAsync(CancellationToken cancellationToken = default); + Task GetConfigurationAsync( + bool forceNewSecretManagerPull = false, + CancellationToken cancellationToken = default); } internal class ActiveMqServerConfigurationSource( @@ -14,10 +16,12 @@ internal class ActiveMqServerConfigurationSource( IOptions options) : IActiveMqServerConfigurationSource { public async Task GetConfigurationAsync( + bool forceNewSecretManagerPull = false, CancellationToken cancellationToken = default) { var secrets = await secretManagerCacheService.GetSecretsAsync( [options.Value.UserPath, options.Value.PasswordPath], + force: forceNewSecretManagerPull, cancellationToken: cancellationToken); return new ActiveMqServerConfigurationModel diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqConsumerRetryWrapper.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqConsumerRetryWrapper.cs new file mode 100644 index 00000000..e53b7afd --- /dev/null +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqConsumerRetryWrapper.cs @@ -0,0 +1,128 @@ +using Apache.NMS; +using Microsoft.Extensions.Options; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Configuration; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Exceptions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Factories; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; + +internal interface IActiveMqConsumerRetryWrapper +{ + Task GetChannelAndDoActionWithRetryAsync(Func callback, + Action? onNewConnectionCallback = null, + Action? onNewMessageConsumerCallback = null, + CancellationToken cancellationToken = default); + + void ResetConsumer(); +} + +internal class ActiveMqConsumerRetryWrapper( + IActiveMqConnectionFactory connectionFactory, + IActiveMqRetryWrapperService retryWrapperService, + IActiveMqExceptionArbiterService exceptionArbiterService, + IOptions configuration) : IActiveMqConsumerRetryWrapper +{ + private IMessageConsumer? _messageConsumer; + + private async Task CallbackAsync(Func callback, + RetryState state, + Action? onNewConnectionCallback, + Action? onNewMessageConsumerCallback, + CancellationToken cancellationToken) + { + if (state.Exception is not null) + { + state.RetryNumber++; + ResetConsumer(); + // Future: Distinguish between exceptions, in the style of RabbitMQ + } + + // ReSharper disable once ReplaceWithSingleAssignment.False + var forceNewSecretManagerPull = false; + + // ReSharper disable once ConvertIfToOrExpression + if (state.Exception is NMSSecurityException securityException + && exceptionArbiterService.GetReport(securityException, state.RetryNumber) is {CouldBeTransient: true}) + { + forceNewSecretManagerPull = true; + } + + try + { + var consumer = await GetConsumerAsync(onNewConnectionCallback, onNewMessageConsumerCallback, + forceNewSecretManagerPull, cancellationToken); + await callback(consumer, cancellationToken); + } + catch (Exception e) + { + state.Exception = e; + throw; + } + } + + /// + /// Get a cached consumer or get a new one from the connection factory. + /// Confirming that the invocation of this method should be already covered by the retry wrapper service. + /// + /// + /// + /// + /// + /// + /// + private async Task GetConsumerAsync(Action? onNewConnectionCallback, + Action? onNewMessageConsumerCallback, bool forceNewSecretManagerPull, + CancellationToken cancellationToken) + { + if (_messageConsumer is not null) + { + return _messageConsumer; + } + + var connection = await connectionFactory.GetConnectionAsync( + forceNewSecretManagerPull, + cancellationToken); + onNewConnectionCallback?.Invoke(connection); + await connection.StartAsync(); + var session = await connection.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge); + var queue = await session.GetQueueAsync(configuration.Value.QueueName); + + if (queue is null) + { + throw new CouldNotLoadQueueException(); + } + + var consumer = await session.CreateConsumerAsync(queue); + onNewMessageConsumerCallback?.Invoke(consumer); + // Cache for later + _messageConsumer = consumer; + + return consumer; + } + + public void ResetConsumer() + { + _messageConsumer = null; + } + + public Task GetChannelAndDoActionWithRetryAsync(Func callback, + Action? onNewConnectionCallback = null, + Action? onNewMessageConsumerCallback = null, + CancellationToken cancellationToken = default) + { + return retryWrapperService.RunAsync( + (state, ct) => CallbackAsync(callback, state, onNewConnectionCallback, onNewMessageConsumerCallback, ct), + new RetryState + { + Exception = null, + RetryNumber = 0 + }, cancellationToken); + } + + private sealed class RetryState + { + public required Exception? Exception { get; set; } + public required int RetryNumber { get; set; } + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs index 7cf87d1d..31124003 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqJobSource.cs @@ -4,99 +4,19 @@ using RedShirt.Example.JobWorker.Core.Enums; using RedShirt.Example.JobWorker.Core.Models; using RedShirt.Example.JobWorker.Core.Services.Abstractions; -using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Exceptions; -using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Factories; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Configuration; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Models; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; internal class ActiveMqJobSource( - IActiveMqConnectionFactory connectionFactory, IActiveMqRetryWrapperService retryWrapperService, - IOptions configuration, + IActiveMqConsumerRetryWrapper consumerRetryWrapper, + IOptions configuration, ILogger logger) : IJobSource { - private IMessageConsumer? _messageConsumer; - - private async Task FetchJobsAsync(int batchSize, CancellationToken cancellationToken) - { - try - { - var consumer = await retryWrapperService.RunAsync(GetConsumerAsync, cancellationToken); - var getJobsResponseItems = new List(); - - while (getJobsResponseItems.Count < batchSize) - { - var result = - await retryWrapperService.RunAsync(_ => consumer.ReceiveAsync(TimeSpan.FromMilliseconds(100)), - cancellationToken); - - if (result is null) - // Nothing more to grab at the moment. - { - break; - } - - // Got a message, add it to return set. - getJobsResponseItems.Add(new ActiveMqRawJobModel - { - Message = result, - MessageId = result.NMSMessageId, // Not really used by this framework, but why not - CreatedAtUtc = DateTime.UtcNow - }); - } - - return new JobSourceResponse - { - Items = getJobsResponseItems - }; - } - catch - { - ResetConsumer(); - throw; - } - } - - /// - /// Get a cached consumer or get a new one from the connection factory. - /// Confirming that the invocation of this method should be already covered by the retry wrapper service. - /// - /// - /// - /// - private async Task GetConsumerAsync(CancellationToken cancellationToken) - { - if (_messageConsumer is not null) - { - return _messageConsumer; - } - - var connection = await connectionFactory.GetConnectionAsync(cancellationToken); - await connection.StartAsync(); - var session = await connection.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge); - var queue = await session.GetQueueAsync(configuration.Value.QueueName); - - if (queue is null) - { - throw new CouldNotLoadQueueException(); - } - - var consumer = await session.CreateConsumerAsync(queue); - - // Cache for later - _messageConsumer = consumer; - - return consumer; - } - - private void ResetConsumer() - { - _messageConsumer = null; - } - public int RecommendedHeartbeatIntervalSeconds => 0; public bool IsSubscriptionSource => false; @@ -116,6 +36,9 @@ public async Task AcknowledgeAsync(IRawJobModel message, CoreJobResult result, // Acknowledge whether successful, recoverable, or unrecoverable // (ActiveMQ client API has no direct dead-letter call here). + // Noting that it is very intentional that we use the base IActiveMqRetryWrapperService here. + // An exception here is not going to be anything that we could solve with a reconnect. + // In fact, it would only cause more problems. await retryWrapperService.RunAsync( _ => jobModel.Message.AcknowledgeAsync(), cancellationToken); @@ -128,7 +51,35 @@ public async Task GetJobsAsync(int batchSize, CancellationTo logger.LogTrace("Fetching up to {EffectiveBatchSize} messages from ActiveMQ Queue: {QueueName}", batchSize, configuration.Value.QueueName); - return await FetchJobsAsync(batchSize, cancellationToken); + var getJobsResponseItems = new List(); + + while (getJobsResponseItems.Count < batchSize) + { + IMessage? result = null; + + await consumerRetryWrapper.GetChannelAndDoActionWithRetryAsync( + async (consumer, _) => { result = await consumer.ReceiveAsync(TimeSpan.FromMilliseconds(100)); }, + cancellationToken: cancellationToken); + + if (result is null) + // Nothing more to grab at the moment. + { + break; + } + + // Got a message, add it to return set. + getJobsResponseItems.Add(new ActiveMqRawJobModel + { + Message = result, + MessageId = result.NMSMessageId, // Not really used by this framework, but why not + CreatedAtUtc = DateTime.UtcNow + }); + } + + return new JobSourceResponse + { + Items = getJobsResponseItems + }; } public Task HeartbeatAsync(IRawJobModel message, CancellationToken cancellationToken = default) @@ -143,9 +94,4 @@ public Task StartSubscriberAsync(CancellationToken cancellationToken = default) { throw new NotSupportedException(); } - - public sealed class ConfigurationModel - { - public required string QueueName { get; init; } - } } \ No newline at end of file diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqSubscribeConfigurationService.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqSubscribeConfigurationService.cs new file mode 100644 index 00000000..ad268f95 --- /dev/null +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqSubscribeConfigurationService.cs @@ -0,0 +1,15 @@ +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; + +/// +/// Indicates whether subscription mode is enabled. +/// Used by to prevent a circular loop. +/// +internal interface IActiveMqSubscribeConfigurationService +{ + bool IsSubscription { get; } +} + +internal class ActiveMqSubscribeConfigurationService(bool isSubscription) : IActiveMqSubscribeConfigurationService +{ + public bool IsSubscription => isSubscription; +} \ No newline at end of file diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqSubscribeJobSource.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqSubscribeJobSource.cs new file mode 100644 index 00000000..c398763d --- /dev/null +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqSubscribeJobSource.cs @@ -0,0 +1,322 @@ +using Apache.NMS; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using RedShirt.Example.JobWorker.Common.Services.Utility; +using RedShirt.Example.JobWorker.Core.Enums; +using RedShirt.Example.JobWorker.Core.Exceptions; +using RedShirt.Example.JobWorker.Core.Models; +using RedShirt.Example.JobWorker.Core.Services.Abstractions; +using RedShirt.Example.JobWorker.Core.Services.Configuration; +using RedShirt.Example.JobWorker.Core.Services.ExecutionState; +using RedShirt.Example.JobWorker.Core.Services.Jobs.Subscriptions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Configuration; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Models; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; + +#pragma warning disable S107 +internal class ActiveMqSubscribeJobSource( + IActiveMqRetryWrapperService retryWrapperService, + IActiveMqConsumerRetryWrapper consumerRetryWrapper, + ICoreConfigurationService coreConfigurationService, + IJobSubscriberIntakeQueue jobSubscriberIntakeQueue, + IExecutionEndArbiter executionEndArbiter, + ISleepService sleepService, + IActiveMqSubscribeExceptionArbiter subscribeExceptionArbiter, + IOptions configuration, + ILogger logger) + : IJobSource +#pragma warning restore S107 +{ + /// + /// Cancellation token provided when subscription started. + /// + private CancellationToken _cancellationToken; + + /// + /// Whether is currently running. + /// + private bool _subscribeLoopRunning; + + private Task OnReceivedAsync(IMessage message, CancellationToken cancellationToken) + { + try + { + _ = cancellationToken; + + logger.LogTrace("Received message {MessageId} from ActiveMQ Queue: {QueueName}", + message.NMSMessageId ?? "UNKNOWN", configuration.Value.QueueName); + + var job = new ActiveMqRawJobModel + { + Message = message, + MessageId = message.NMSMessageId ?? "UNKNOWN", + CreatedAtUtc = DateTime.UtcNow + }; + + jobSubscriberIntakeQueue.Load(new JobSourceResponse + { + Items = [job] + }); + return Task.CompletedTask; + } + catch (Exception exception) + { + return Task.FromException(exception); + } + } + + /// + /// Use a consumer to start an async listener. + /// Assumed to be invoked within a retry wrapper. + /// + /// + /// + private Task StartConsumerAsync(IMessageConsumer consumer, CancellationToken cancellationToken) + { + _ = cancellationToken; + + logger.LogTrace("Subscribing to ActiveMQ Queue: {QueueName}", configuration.Value.QueueName); + + consumer.AsyncListener += OnReceivedAsync; + + logger.LogTrace("Subscribed to ActiveMQ Queue: {QueueName}", configuration.Value.QueueName); + return Task.CompletedTask; + } + + private void OnConnectionResumed() + { + logger.LogInformation("ActiveMQ connection established"); + // Unlike RabbitMQ, no need to resubscribe - handled by underlying client library + } + + /// + /// Attempt to start the consumer, retrying according to transient / halt-on-failure configuration. + /// Only one invocation may run at a time; concurrent callers return immediately. + /// + /// + /// Verb used in error logs (e.g. "subscribing" or "re-subscribing"). + /// + /// + private async Task SubscribeWithRetryLoopAsync(string logVerb, CancellationToken cancellationToken) + { + // CompareExchange returns the prior value; true means another caller already holds the lock. + if (Interlocked.CompareExchange(ref _subscribeLoopRunning, true, false)) + { + return; + } + + try + { + var firstIteration = true; + while (true) + { + if (!firstIteration) + { + await sleepService.DelayAsync(TimeSpan.FromSeconds(1), cancellationToken); + } + + firstIteration = false; + + try + { + await GetConsumerAndDoActionWithRetryAsync(StartConsumerAsync, cancellationToken); + } + catch (OperationCanceledException e) when (e.CancellationToken.IsCancellationRequested) + { + // Pass + } +#pragma warning disable S2139 + // Misguided sonar warning + catch (Exception e) +#pragma warning restore S2139 + { + // Some variety of non-transient failure + logger.LogError(e, "Error {LogVerb} to ActiveMQ", logVerb); + + if (e is WorkerJobSourceException {CouldBeTransient: true} && + !coreConfigurationService.IsTreatingTransientExceptionAsFailure()) + { + // Transient: Retry and try again + continue; + } + + if (!coreConfigurationService.IsHaltOnFailure()) + { + // Not halting on failure, continue and try again + continue; + } + + // HaltOnFailure is true. + // Pass the exception up to one of our threads as opposed to an ActiveMQ-managed one + executionEndArbiter.Stop(e); + // Fall through to break out of loop + } + + break; + } + } + finally + { + Interlocked.Exchange(ref _subscribeLoopRunning, false); + } + } + + private Task GetConsumerAndDoActionWithRetryAsync(Func callback, + CancellationToken cancellationToken) + { + return consumerRetryWrapper.GetChannelAndDoActionWithRetryAsync(callback, OnNewConnection, + cancellationToken: cancellationToken); + } + + /// + /// Handle ActiveMQ exceptions. + /// Intended to handle network connection problems and initiate a reconnect. + /// + /// + private void OnException(Exception exception) + { + /* + * ExceptionListener is the reconnect signal when not using NMS failover. + * However, ExceptionListener casts a wider net that we need to filter out. + * + * During development this was originally done using the ActiveMQ library's built-in fail-over settings, + * which were enforced on the broker URI in the connection factory. However, this did not cover the niche + * case of what might happen if the connection was interrupted AND the credentials changed. + * Also explained in the connection factory. + */ + + if (subscribeExceptionArbiter.IsReasonToReconnect(exception) + || subscribeExceptionArbiter.IsReasonToStopIfHaltOnFailure(exception)) + { + /* + * Is an explicit reason to reconnect or another serious error. Funnel both through reconnection. + * + * If it is a known reason to reconnect, then reconnect is exactly what we'll do. + * If it is another error, then the reconnect serves a few different purposes: + * * The reconnect is aware of the established retry loop and the main exception arbiter, allowing both to weigh in. + * * If HaltOnFailure is false, then it allows the subscriber a chance to recover + * * If HaltOnFailure is true, then the established retry loop still stops the application + */ + + // We want to kick off a worker thread to reconnect and resubscribe. + // Not doing it here because we are not in an async method. + + // Avoid spawning another reconnect task while a subscribe loop is already in flight. + if (Volatile.Read(ref _subscribeLoopRunning)) + { + return; + } + + logger.LogWarning(exception, "ActiveMQ ExceptionListener problem, reconnecting"); + + consumerRetryWrapper.ResetConsumer(); + + _ = Task.Run(() => SubscribeWithRetryLoopAsync("re-subscribing", _cancellationToken), _cancellationToken); + return; + } + + if (subscribeExceptionArbiter.IsAccountedForAndLikelyTransientError(exception)) + { + // Is an expected transient error, not worth warning about + return; + } + + logger.LogWarning(exception, + "Unaccounted-for exception in {Name}. Classify via {IActiveMqSubscribeExceptionArbiter} methods", + nameof(ActiveMqSubscribeJobSource), + nameof(IActiveMqSubscribeExceptionArbiter)); + } + + private void OnNewConnection(IConnection connection) + { + // Deliberately using ExceptionListener instead of ConnectionInterruptedListener. + // ConnectionInterruptedListener is not used when not using fail-over + // (as is enforced in the factory for subscribe mode, see there for justification) + connection.ExceptionListener -= OnException; + connection.ExceptionListener += OnException; + connection.ConnectionResumedListener -= OnConnectionResumed; + connection.ConnectionResumedListener += OnConnectionResumed; + } + + private async Task WaitThenStopSubscriberAsync(CancellationToken cancellationToken = default) + { + _cancellationToken = cancellationToken; + + await executionEndArbiter.WaitForFinishedAsync(cancellationToken); + + try + { + await GetConsumerAndDoActionWithRetryAsync( + (consumer, _) => + { + consumer.AsyncListener -= OnReceivedAsync; + + return Task.CompletedTask; + }, + cancellationToken); + } + catch (Exception exception) + { + logger.LogError(exception, "Could not unsubscribe: {Message}", exception.Message); + // Not terribly concerned about any other exceptions because it's assumed in the shutdown period anyway. + // But just in case... + } + } + + /// + /// Acknowledge a message. + /// Same client-ack behaviour as . + /// + /// + /// + /// + public async Task AcknowledgeAsync(IRawJobModel message, CoreJobResult result, + CancellationToken cancellationToken = default) + { + if (message is not ActiveMqRawJobModel jobModel) + { + // Message did not originate from ActiveMQ, return + return; + } + + // Intentionally not using result + // The `_ = result;` phrasing prevents certain code analysis tools from flagging this as a potential issue + _ = result; + + // Acknowledge whether successful, recoverable, or unrecoverable + // (ActiveMQ client API has no direct dead-letter call here). + // Noting that it is very intentional that we use the base IActiveMqRetryWrapperService here. + // An exception here is not going to be anything that we could solve with a reconnect. + // In fact, it would only cause more problems. + await retryWrapperService.RunAsync( + _ => jobModel.Message.AcknowledgeAsync(), + cancellationToken); + } + + public Task GetJobsAsync(int batchSize, CancellationToken cancellationToken = default) + { + throw new NotSupportedException(); + } + + public int RecommendedHeartbeatIntervalSeconds => 0; + + public bool IsSubscriptionSource => true; + + public Task HeartbeatAsync(IRawJobModel message, CancellationToken cancellationToken = default) + { + /* + * Not necessary. Heartbeats are managed by the persistence of the IMessage / IConnection objects. + */ + return Task.CompletedTask; + } + + public async Task StartSubscriberAsync(CancellationToken cancellationToken = default) + { + // Kick off the task that shall watch for unsubscribes when the application stops + _ = Task.Run(() => WaitThenStopSubscriberAsync(cancellationToken), cancellationToken); + + await SubscribeWithRetryLoopAsync("subscribing", cancellationToken); + } +} \ No newline at end of file diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs index 569f8587..e255e5b5 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqExceptionArbiterService.cs @@ -15,7 +15,16 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; /// internal interface IActiveMqExceptionArbiterService { - ActiveMqExceptionArbiterReport GetReport(Exception exception); + /// + /// Get a judgement on an exception. + /// + /// + /// + /// Attempt number. First attempt number starts at 1. This arbiter's partner retry wrapper uses + /// Polly pipelines that are zero-based, but I find using non-zero-based more intuitive. + /// + /// + ActiveMqExceptionArbiterReport GetReport(Exception exception, int attemptNumber); } /// @@ -52,7 +61,22 @@ private static ActiveMqExceptionArbiterReport Handled( }; } - public ActiveMqExceptionArbiterReport GetReport(Exception exception) + /// + /// Handle the special case of an NMSSecurityException. + /// + /// + /// + /// + private static ActiveMqExceptionArbiterReport MapSecurityException(NMSSecurityException exception, + int attemptNumber) + { + // Not super-happy about judging off of a message, but Google says that the exit code property is ambiguous. + // That's a bit of an understatement, as in practice it's blank. + var firstPasswordOffense = exception.Message.EndsWith(" or password is invalid.") && attemptNumber == 1; + return Fresh(true, firstPasswordOffense, true); + } + + public ActiveMqExceptionArbiterReport GetReport(Exception exception, int attemptNumber) { ArgumentNullException.ThrowIfNull(exception); @@ -79,7 +103,7 @@ public ActiveMqExceptionArbiterReport GetReport(Exception exception) // Unsupported / unreadable payload — a local data issue, not retryable. CouldNotRetrieveMessageBodyException => Fresh(true, false, false), // Auth failures — ops can grant credentials / ACLs externally. - NMSSecurityException => Fresh(true, false, true), + NMSSecurityException securityException => MapSecurityException(securityException, attemptNumber), // Missing / invalid destination — ops can create or restore the queue externally. InvalidDestinationException => Fresh(true, false, true), // Bad local client identity or selector — requires a config change, not an external fix. diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs index 760f0f13..ff84d871 100644 --- a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqRetryWrapperService.cs @@ -34,6 +34,9 @@ internal interface IActiveMqRetryWrapperService /// Task RunAsync(Func> func, CancellationToken cancellationToken = default); + Task RunAsync(Func> func, TState state, + CancellationToken cancellationToken = default); + /// /// Executes with retry for expected transient ActiveMQ failures. /// @@ -52,6 +55,9 @@ internal interface IActiveMqRetryWrapperService /// is true. /// Task RunAsync(Func func, CancellationToken cancellationToken = default); + + Task RunAsync(Func func, TState state, + CancellationToken cancellationToken = default); } /// @@ -98,7 +104,7 @@ private ResiliencePipeline GetRetryPipeline() return PredicateResult.False(); } - var report = exceptionArbiterService.GetReport(exception); + var report = exceptionArbiterService.GetReport(exception, args.AttemptNumber + 1); return report is {IsExpected: true, CouldBeTransient: true} ? PredicateResult.True() : PredicateResult.False(); @@ -131,7 +137,7 @@ await sleepService.DelayAsync(TimeSpan.FromSeconds(Math.Pow(2, args.AttemptNumbe private bool TryGetWrappedException(Exception exception, out Exception? wrappedException) { wrappedException = null; - var report = exceptionArbiterService.GetReport(exception); + var report = exceptionArbiterService.GetReport(exception, ActiveMqRetryCount); // ReSharper disable once DuplicatedSequentialIfBodies if (report.AlreadyHandled && exception is WorkerJobSourceException) @@ -183,6 +189,30 @@ public async Task RunAsync(Func> func, } } + public async Task RunAsync(Func> func, + TState state, CancellationToken cancellationToken = default) + { + try + { + return await GetRetryPipeline().ExecuteAsync( + async (s, token) => await func(s, token), state, cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception exception) + { + if (TryGetWrappedException(exception, out var wrappedException) && wrappedException is not null) + { + throw wrappedException; + } + + // Do a flat throw to preserve stack trace + throw; + } + } + /// public async Task RunAsync(Func func, CancellationToken cancellationToken = default) { @@ -207,4 +237,28 @@ await GetRetryPipeline().ExecuteAsync( throw; } } + + public async Task RunAsync(Func func, TState state, + CancellationToken cancellationToken = default) + { + try + { + await GetRetryPipeline().ExecuteAsync( + async (s, ct) => await func(s, ct), state, cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception exception) + { + if (TryGetWrappedException(exception, out var wrappedException) && wrappedException is not null) + { + throw wrappedException; + } + + // Do a flat throw to preserve stack trace + throw; + } + } } \ No newline at end of file diff --git a/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqSubscribeExceptionArbiterService.cs b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqSubscribeExceptionArbiterService.cs new file mode 100644 index 00000000..e22b815c --- /dev/null +++ b/src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/Resilience/ActiveMqSubscribeExceptionArbiterService.cs @@ -0,0 +1,130 @@ +using Apache.NMS; +using Apache.NMS.ActiveMQ; +using System.Net.Sockets; +using ActiveMqIoException = Apache.NMS.ActiveMQ.IOException; +using IOException = System.IO.IOException; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; + +/// +/// Classifies exceptions for the ActiveMQ subscribe job source: +/// reconnect, halt-on-failure stop, or accounted-for transient noise. +/// +internal interface IActiveMqSubscribeExceptionArbiter +{ + /// + /// Whether (or any inner exception) looks like expected + /// noise that should neither reconnect nor halt. + /// + /// + /// + /// Used to suppress "unaccounted-for" warnings for known brief / non-fatal NMS callbacks + /// (especially broker Exception frames). + /// + /// + /// Typical cases: ; brief broker pressure + /// (, ); + /// local session state (). + /// Permanent auth/config failures belong in ; + /// transport drops belong in . + /// + /// + bool IsAccountedForAndLikelyTransientError(Exception exception); + + /// + /// Whether (or any inner exception) is a reason to reset the + /// consumer and run the subscribe reconnect loop. + /// + /// + /// Covers transport / connection drops and a closed consumer that still needs a fresh + /// subscription. Expected non-reconnect NMS callbacks are classified by + /// or + /// instead. + /// + bool IsReasonToReconnect(Exception exception); + + /// + /// Whether (or any inner exception) is a permanent auth / config + /// failure that should stop the worker when halt-on-failure is enabled. + /// + /// + /// These are expected NMS signals where reconnecting will not help (bad credentials, missing + /// destination, invalid client id/selector). Callers should only stop when + /// HaltOnFailure is true. + /// + bool IsReasonToStopIfHaltOnFailure(Exception exception); +} + +/// +/// Default implementation. +/// +internal class ActiveMqSubscribeExceptionArbiterService : IActiveMqSubscribeExceptionArbiter +{ + /// + public bool IsAccountedForAndLikelyTransientError(Exception exception) + { + for (var current = exception; current is not null; current = current.InnerException) + { + switch (current) + { + // Broker Exception command frames on the connection — often non-fatal from the + // client POV (disconnect races, advisory-style errors). + case BrokerException: + // Brief broker-side contention / rollback. + case ResourceAllocationException: + case TransactionRolledBackException: + // Local session state noise on ExceptionListener. + case IllegalStateException: + return true; + } + } + + return false; + } + + /// + public bool IsReasonToStopIfHaltOnFailure(Exception exception) + { + for (var current = exception; current is not null; current = current.InnerException) + { + switch (current) + { + case NMSSecurityException: + case InvalidDestinationException: + case InvalidClientIDException: + case InvalidSelectorException: + return true; + } + } + + return false; + } + + /// + public bool IsReasonToReconnect(Exception exception) + { + for (var current = exception; current is not null; current = current.InnerException) + { + switch (current) + { + // Abrupt peer close / wire EOF (often wrapped in NMSException). + case EndOfStreamException: + // Socket-level failures (reset, refused, timed out, host unreachable). + case SocketException: + // OpenWire inactivity monitor and other ActiveMQ transport IO failures. + case ActiveMqIoException: + // Generic stream IO from the transport thread. + case IOException: + // Connection lifecycle failures reported by the NMS client. + case NMSConnectionException: + case ConnectionClosedException: + case ConnectionFailedException: + // Consumer gone — rebuild via reconnect/resubscribe rather than treat as noise. + case ConsumerClosedException: + return true; + } + } + + return false; + } +} \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.IntegrationTests/Tests/DependencyInjectionTests.cs b/test/RedShirt.Example.JobWorker.IntegrationTests/Tests/DependencyInjectionTests.cs index aac84043..745e8510 100644 --- a/test/RedShirt.Example.JobWorker.IntegrationTests/Tests/DependencyInjectionTests.cs +++ b/test/RedShirt.Example.JobWorker.IntegrationTests/Tests/DependencyInjectionTests.cs @@ -13,6 +13,32 @@ public void Test_Get_Runner_ActiveMq() ["AWS_SECRET_ACCESS_KEY"] = "bar", ["AWS_SESSION_TOKEN"] = "foobar", ["HEALTH__ENABLED"] = "false", + ["JOB_SOURCE__ACTIVEMQ__SUBSCRIBE"] = "false", + ["UseActiveMq"] = "1", + ["UseAzureQueueStorage"] = "0", + ["UseAzureServiceBus"] = "0", + ["UseGooglePubSub"] = "0", + ["UseNats"] = "0", + ["UseRedisStreams"] = "0", + ["UseRabbitMq"] = "0", + ["UseRabbitMqSubscribe"] = "0", + ["UseKinesis"] = "0", + ["UseKafka"] = "0", + ["UsePulsar"] = "0" + }, () => { Assert.NotNull(Setup.GetHost()); }); + } + + [Fact] + public void Test_Get_Runner_ActiveMqSubscribe() + { + TestUtilities.WrapEnvironment(new Dictionary + { + ["AWS_SERVICE_URL"] = "http://foo.bar", + ["AWS_ACCESS_KEY_ID"] = "foo", + ["AWS_SECRET_ACCESS_KEY"] = "bar", + ["AWS_SESSION_TOKEN"] = "foobar", + ["HEALTH__ENABLED"] = "false", + ["JOB_SOURCE__ACTIVEMQ__SUBSCRIBE"] = "true", ["UseActiveMq"] = "1", ["UseAzureQueueStorage"] = "0", ["UseAzureServiceBus"] = "0", diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs index c58df5df..404cb90c 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Extensions/ServiceCollectionExtensionsTests.cs @@ -27,6 +27,29 @@ public void AddActiveMqJobManagement_RegistersExpectedServices() Assert.Contains(services, d => d.ServiceType == typeof(IJobFailureHandler) && d.ImplementationType == typeof(NoReactionFailureHandler)); Assert.Contains(services, d => d.ServiceType == typeof(IActiveMqExceptionArbiterService)); + Assert.Contains(services, d => d.ServiceType == typeof(IActiveMqSubscribeExceptionArbiter)); Assert.Contains(services, d => d.ServiceType == typeof(IActiveMqRetryWrapperService)); + Assert.Contains(services, d => d.ServiceType == typeof(IActiveMqConsumerRetryWrapper) + && d.ImplementationType == typeof(ActiveMqConsumerRetryWrapper)); + } + + [Fact] + public void AddActiveMqJobManagement_WhenSubscribeTrue_RegistersSubscribeJobSource() + { + var services = new ServiceCollection(); + var configuration = new ConfigurationBuilder() + .AddInMemoryCollection(new Dictionary + { + ["JobSource:ActiveMq:QueueName"] = "jobs", + ["JobSource:ActiveMq:Subscribe"] = "true" + }) + .Build(); + + services.AddActiveMqJobManagement(configuration); + + Assert.Contains(services, d => d.ServiceType == typeof(IJobSource) + && d.ImplementationType == typeof(ActiveMqSubscribeJobSource)); + Assert.DoesNotContain(services, d => d.ServiceType == typeof(IJobSource) + && d.ImplementationType == typeof(ActiveMqJobSource)); } } \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Factories/ActiveMqConnectionFactoryTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Factories/ActiveMqConnectionFactoryTests.cs index c3c481a2..3725955b 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Factories/ActiveMqConnectionFactoryTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Factories/ActiveMqConnectionFactoryTests.cs @@ -6,8 +6,10 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Fact public class ActiveMqConnectionFactoryTests { - [Fact] - public async Task Test_GetConnectionAsync() + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task GetConnectionAsync_PassesForceFlagToInnerFactory(bool forceNewSecretManagerPull) { var mockConnection = new Mock(MockBehavior.Strict); @@ -18,13 +20,20 @@ public async Task Test_GetConnectionAsync() var innerFactory = new Mock(MockBehavior.Strict); innerFactory - .Setup(i => i.GetConnectionFactoryWrapperAsync(TestContext.Current.CancellationToken)) + .Setup(i => i.GetConnectionFactoryWrapperAsync( + forceNewSecretManagerPull, + TestContext.Current.CancellationToken)) .ReturnsAsync(mockWrapper.Object); var factory = new ActiveMqConnectionFactory(innerFactory.Object); - var returnedConnection = await factory.GetConnectionAsync(TestContext.Current.CancellationToken); + var returnedConnection = await factory.GetConnectionAsync( + forceNewSecretManagerPull, + TestContext.Current.CancellationToken); Assert.NotNull(returnedConnection); Assert.Same(returnedConnection, mockConnection.Object); + innerFactory.Verify(i => i.GetConnectionFactoryWrapperAsync( + forceNewSecretManagerPull, + TestContext.Current.CancellationToken), Times.Once); } } \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Factories/InnerActiveMqConnectionFactoryTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Factories/InnerActiveMqConnectionFactoryTests.cs index d625d6f7..8cf4dd2a 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Factories/InnerActiveMqConnectionFactoryTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Factories/InnerActiveMqConnectionFactoryTests.cs @@ -1,3 +1,5 @@ +using Apache.NMS.ActiveMQ; +using RedShirt.Example.JobWorker.Core.Services.Configuration; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Factories; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Models; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; @@ -7,17 +9,116 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Fact public class InnerActiveMqConnectionFactoryTests { + private const string PlainBrokerUri = "tcp://localhost:1234/"; + + private static readonly int DefaultQueuePrefetch = + new ConnectionFactory(PlainBrokerUri).PrefetchPolicy.QueuePrefetch; + + private static Mock CreateConfigSource(string brokerUri = PlainBrokerUri) + { + var configSource = new Mock(MockBehavior.Strict); + configSource + .Setup(cs => cs.GetConfigurationAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(new ActiveMqServerConfigurationModel + { + BrokerUri = brokerUri, + User = "u", + Password = "p" + }); + return configSource; + } + + private static Mock CreateSubscribeConfiguration(bool isSubscription) + { + var subscribeConfiguration = new Mock(MockBehavior.Strict); + subscribeConfiguration.SetupGet(s => s.IsSubscription).Returns(isSubscription); + return subscribeConfiguration; + } + + private static InnerActiveMqConnectionFactory CreateFactory( + IActiveMqServerConfigurationSource configurationSource, + IActiveMqSubscribeConfigurationService subscribeConfiguration, + ICoreConfigurationService coreConfiguration) + { + return new InnerActiveMqConnectionFactory(configurationSource, subscribeConfiguration, coreConfiguration); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task GetWrapperAsync_PassesForceNewSecretManagerPullToConfigurationSource( + bool forceNewSecretManagerPull) + { + var configSource = CreateConfigSource(); + var subscribeConfiguration = CreateSubscribeConfiguration(false); + var coreConfiguration = new Mock(MockBehavior.Strict); + + var innerFactory = CreateFactory(configSource.Object, subscribeConfiguration.Object, coreConfiguration.Object); + + await innerFactory.GetConnectionFactoryWrapperAsync( + forceNewSecretManagerPull, + TestContext.Current.CancellationToken); + + configSource.Verify( + cs => cs.GetConfigurationAsync(forceNewSecretManagerPull, TestContext.Current.CancellationToken), + Times.Once); + } + + [Fact] + public async Task GetWrapperAsync_WhenNotSubscription_LeavesPlainBrokerUriAndDefaultPrefetch() + { + var configSource = CreateConfigSource(); + var subscribeConfiguration = CreateSubscribeConfiguration(false); + var coreConfiguration = new Mock(MockBehavior.Strict); + + var innerFactory = CreateFactory(configSource.Object, subscribeConfiguration.Object, coreConfiguration.Object); + + var wrapper = Assert.IsType( + await innerFactory.GetConnectionFactoryWrapperAsync( + cancellationToken: TestContext.Current.CancellationToken)); + + Assert.Equal(DefaultQueuePrefetch, wrapper.InternalConnectionFactory.PrefetchPolicy.QueuePrefetch); + Assert.Equal(PlainBrokerUri, wrapper.InternalConnectionFactory.BrokerUri.ToString()); + coreConfiguration.Verify(c => c.GetBacklogSize(), Times.Never); + configSource.Verify(cs => cs.GetConfigurationAsync(false, TestContext.Current.CancellationToken), Times.Once); + } + [Fact] - public async Task Test_GetWrapperAsync() + public async Task GetWrapperAsync_WhenSubscriptionAndBrokerUriIsFailover_StripsFailover() + { + const string failoverUri = "failover:(tcp://broker:61616)?transport.maxReconnectAttempts=5"; + var configSource = CreateConfigSource(failoverUri); + var subscribeConfiguration = CreateSubscribeConfiguration(true); + var coreConfiguration = new Mock(MockBehavior.Strict); + coreConfiguration.Setup(c => c.GetBacklogSize()).Returns(1); + + var innerFactory = CreateFactory(configSource.Object, subscribeConfiguration.Object, coreConfiguration.Object); + + var wrapper = Assert.IsType( + await innerFactory.GetConnectionFactoryWrapperAsync( + cancellationToken: TestContext.Current.CancellationToken)); + + Assert.Equal("tcp://broker:61616/", wrapper.InternalConnectionFactory.BrokerUri.ToString()); + Assert.DoesNotContain("failover", wrapper.InternalConnectionFactory.BrokerUri.ToString(), + StringComparison.OrdinalIgnoreCase); + } + + [Theory] + [InlineData(1)] + [InlineData(5)] + [InlineData(100)] + public async Task GetWrapperAsync_WhenSubscription_SetsCredentialsPlainUriAndQueuePrefetchFromBacklogSize( + int backlogSize) { var valueName = Guid.NewGuid().ToString(); var valuePassword = Guid.NewGuid().ToString(); // The factory constructor insists that the URI be properly formatted. // ToStringing the URI also tacks a '/' onto the end. Go figure. - var valueHostname = "tcp://localhost:1234/"; + var valueHostname = PlainBrokerUri; var configSource = new Mock(MockBehavior.Strict); - configSource.Setup(cs => cs.GetConfigurationAsync(TestContext.Current.CancellationToken)) + configSource + .Setup(cs => cs.GetConfigurationAsync(It.IsAny(), It.IsAny())) .ReturnsAsync(new ActiveMqServerConfigurationModel { BrokerUri = valueHostname, @@ -25,15 +126,36 @@ public async Task Test_GetWrapperAsync() Password = valuePassword }); - var innerFactory = new InnerActiveMqConnectionFactory(configSource.Object); + var subscribeConfiguration = CreateSubscribeConfiguration(true); + var coreConfiguration = new Mock(MockBehavior.Strict); + coreConfiguration.Setup(c => c.GetBacklogSize()).Returns(backlogSize); - var rawWrapper = await innerFactory.GetConnectionFactoryWrapperAsync(TestContext.Current.CancellationToken); + var innerFactory = CreateFactory(configSource.Object, subscribeConfiguration.Object, coreConfiguration.Object); + + var rawWrapper = await innerFactory.GetConnectionFactoryWrapperAsync( + cancellationToken: TestContext.Current.CancellationToken); Assert.NotNull(rawWrapper); - var wrapper = rawWrapper as ActiveMqConnectionWrapper; - Assert.NotNull(wrapper); + var wrapper = Assert.IsType(rawWrapper); Assert.Equal(valueName, wrapper.InternalConnectionFactory.UserName); Assert.Equal(valuePassword, wrapper.InternalConnectionFactory.Password); Assert.Equal(valueHostname, wrapper.InternalConnectionFactory.BrokerUri.ToString()); Assert.Same(wrapper.InternalConnectionFactory, wrapper.ConnectionFactory); + Assert.Equal(backlogSize, wrapper.InternalConnectionFactory.PrefetchPolicy.QueuePrefetch); + coreConfiguration.Verify(c => c.GetBacklogSize(), Times.Once); + } + + [Theory] + [InlineData("tcp://localhost:61616", "tcp://localhost:61616")] + [InlineData("tcp://localhost:1234/", "tcp://localhost:1234/")] + [InlineData("failover:(tcp://broker:61616)", "tcp://broker:61616")] + [InlineData( + "failover:(tcp://broker:61616)?transport.maxReconnectAttempts=5", + "tcp://broker:61616")] + [InlineData( + "failover:(tcp://a:61616,tcp://b:61616)?transport.maxReconnectAttempts=5", + "tcp://a:61616")] + public void StripFailoverUri_RemovesFailoverWrapperAndLeavesPlainUri(string input, string expected) + { + Assert.Equal(expected, InnerActiveMqConnectionFactory.StripFailoverUri(input)); } } \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqConsumerRetryWrapperTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqConsumerRetryWrapperTests.cs new file mode 100644 index 00000000..c3addd7d --- /dev/null +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqConsumerRetryWrapperTests.cs @@ -0,0 +1,284 @@ +using Apache.NMS; +using Microsoft.Extensions.Options; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Configuration; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Exceptions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Factories; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Models; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; +using System.Runtime.ExceptionServices; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Services; + +public class ActiveMqConsumerRetryWrapperTests +{ + private static (Mock Factory, Mock Connection, Mock Session, + Mock Queue) + CreateInfrastructure(string queueName, IMessageConsumer? consumer = null) + { + var queue = new Mock(MockBehavior.Strict); + var session = new Mock(MockBehavior.Strict); + session.Setup(s => s.GetQueueAsync(queueName)).ReturnsAsync(queue.Object); + if (consumer is not null) + { + session.Setup(s => s.CreateConsumerAsync(queue.Object)).ReturnsAsync(consumer); + } + + var connection = new Mock(MockBehavior.Strict); + connection.Setup(c => c.StartAsync()).Returns(Task.CompletedTask); + connection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) + .ReturnsAsync(session.Object); + + var factory = new Mock(MockBehavior.Strict); + factory.Setup(f => f.GetConnectionAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(connection.Object); + + return (factory, connection, session, queue); + } + + private static ActiveMqConsumerRetryWrapper CreateWrapper( + IActiveMqRetryWrapperService retry, + IActiveMqConnectionFactory factory, + string queueName, + IActiveMqExceptionArbiterService? exceptionArbiter = null) + { + exceptionArbiter ??= Mock.Of(); + + return new ActiveMqConsumerRetryWrapper( + factory, + retry, + exceptionArbiter, + Options.Create(new ActiveMqConfigurationModel + { + QueueName = queueName + })); + } + + [Fact] + public async Task GetChannelAndDoActionWithRetryAsync_WhenCachedConsumer_ReusesAndSkipsCallbacks() + { + var queueName = Guid.NewGuid().ToString(); + var consumer = new Mock(MockBehavior.Strict); + var (factory, connection, session, _) = CreateInfrastructure(queueName, consumer.Object); + var wrapper = CreateWrapper(new ImmediateRetryWrapper(), factory.Object, queueName); + + var newConnectionCalls = 0; + var newConsumerCalls = 0; + + await wrapper.GetChannelAndDoActionWithRetryAsync( + (_, _) => Task.CompletedTask, + _ => newConnectionCalls++, + _ => newConsumerCalls++, + TestContext.Current.CancellationToken); + await wrapper.GetChannelAndDoActionWithRetryAsync( + (_, _) => Task.CompletedTask, + _ => newConnectionCalls++, + _ => newConsumerCalls++, + TestContext.Current.CancellationToken); + + Assert.Equal(1, newConnectionCalls); + Assert.Equal(1, newConsumerCalls); + factory.Verify(f => f.GetConnectionAsync(false, TestContext.Current.CancellationToken), Times.Once); + connection.Verify(c => c.StartAsync(), Times.Once); + session.Verify(s => s.CreateConsumerAsync(It.IsAny()), Times.Once); + } + + [Fact] + public async Task GetChannelAndDoActionWithRetryAsync_WhenCallbackFails_ResetsConsumerAndRecreates() + { + var queueName = Guid.NewGuid().ToString(); + var firstConsumer = new Mock(MockBehavior.Strict); + var secondConsumer = new Mock(MockBehavior.Strict); + var queue = new Mock(MockBehavior.Strict); + var session = new Mock(MockBehavior.Strict); + session.Setup(s => s.GetQueueAsync(queueName)).ReturnsAsync(queue.Object); + session.SetupSequence(s => s.CreateConsumerAsync(queue.Object)) + .ReturnsAsync(firstConsumer.Object) + .ReturnsAsync(secondConsumer.Object); + + var connection = new Mock(MockBehavior.Strict); + connection.Setup(c => c.StartAsync()).Returns(Task.CompletedTask); + connection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) + .ReturnsAsync(session.Object); + + var factory = new Mock(MockBehavior.Strict); + factory.Setup(f => f.GetConnectionAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(connection.Object); + + var wrapper = CreateWrapper(new ImmediateRetryWrapper(2), factory.Object, queueName); + + var attempts = 0; + var seenConsumers = new List(); + var newConsumerCalls = 0; + + await wrapper.GetChannelAndDoActionWithRetryAsync( + (c, _) => + { + attempts++; + seenConsumers.Add(c); + if (attempts == 1) + { + throw new TimeoutException("transient"); + } + + return Task.CompletedTask; + }, + onNewMessageConsumerCallback: _ => newConsumerCalls++, + cancellationToken: TestContext.Current.CancellationToken); + + Assert.Equal(2, attempts); + Assert.Equal([firstConsumer.Object, secondConsumer.Object], seenConsumers); + Assert.Equal(2, newConsumerCalls); + factory.Verify(f => f.GetConnectionAsync(false, TestContext.Current.CancellationToken), Times.Exactly(2)); + session.Verify(s => s.CreateConsumerAsync(queue.Object), Times.Exactly(2)); + } + + [Fact] + public async Task GetChannelAndDoActionWithRetryAsync_WhenQueueMissing_ThrowsCouldNotLoadQueueException() + { + var queueName = Guid.NewGuid().ToString(); + var session = new Mock(MockBehavior.Strict); + session.Setup(s => s.GetQueueAsync(queueName)).ReturnsAsync((IQueue?) null); + + var connection = new Mock(MockBehavior.Strict); + connection.Setup(c => c.StartAsync()).Returns(Task.CompletedTask); + connection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) + .ReturnsAsync(session.Object); + + var factory = new Mock(MockBehavior.Strict); + factory.Setup(f => f.GetConnectionAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(connection.Object); + + var wrapper = CreateWrapper(new ImmediateRetryWrapper(), factory.Object, queueName); + + await Assert.ThrowsAsync(() => + wrapper.GetChannelAndDoActionWithRetryAsync((_, _) => Task.CompletedTask, + cancellationToken: TestContext.Current.CancellationToken)); + + session.Verify(s => s.CreateConsumerAsync(It.IsAny()), Times.Never); + } + + [Fact] + public async Task GetChannelAndDoActionWithRetryAsync_WhenTransientSecurityException_ForcesSecretRefresh() + { + var queueName = Guid.NewGuid().ToString(); + var firstConsumer = new Mock(MockBehavior.Strict); + var secondConsumer = new Mock(MockBehavior.Strict); + var queue = new Mock(MockBehavior.Strict); + var session = new Mock(MockBehavior.Strict); + session.Setup(s => s.GetQueueAsync(queueName)).ReturnsAsync(queue.Object); + session.SetupSequence(s => s.CreateConsumerAsync(queue.Object)) + .ReturnsAsync(firstConsumer.Object) + .ReturnsAsync(secondConsumer.Object); + + var connection = new Mock(MockBehavior.Strict); + connection.Setup(c => c.StartAsync()).Returns(Task.CompletedTask); + connection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) + .ReturnsAsync(session.Object); + + var factory = new Mock(MockBehavior.Strict); + factory.Setup(f => f.GetConnectionAsync(It.IsAny(), It.IsAny())) + .ReturnsAsync(connection.Object); + + var arbiter = new Mock(MockBehavior.Strict); + arbiter + .Setup(a => a.GetReport(It.IsAny(), It.IsAny())) + .Returns(new ActiveMqExceptionArbiterReport + { + AlreadyHandled = false, + IsExpected = true, + CouldBeTransient = true, + CouldBeExternallySolvable = true + }); + + var wrapper = CreateWrapper(new ImmediateRetryWrapper(2), factory.Object, queueName, arbiter.Object); + + var attempts = 0; + await wrapper.GetChannelAndDoActionWithRetryAsync( + (_, _) => + { + attempts++; + if (attempts == 1) + { + throw new NMSSecurityException("User name or password is invalid."); + } + + return Task.CompletedTask; + }, + cancellationToken: TestContext.Current.CancellationToken); + + Assert.Equal(2, attempts); + factory.Verify(f => f.GetConnectionAsync(false, TestContext.Current.CancellationToken), Times.Once); + factory.Verify(f => f.GetConnectionAsync(true, TestContext.Current.CancellationToken), Times.Once); + arbiter.Verify(a => a.GetReport(It.IsAny(), 1), Times.Once); + } + + [Fact] + public async Task GetChannelAndDoActionWithRetryAsync_WhenUncached_CreatesConsumerAndInvokesCallbacks() + { + var queueName = Guid.NewGuid().ToString(); + var consumer = new Mock(MockBehavior.Strict); + var (factory, connection, _, _) = CreateInfrastructure(queueName, consumer.Object); + var wrapper = CreateWrapper(new ImmediateRetryWrapper(), factory.Object, queueName); + + IConnection? notifiedConnection = null; + IMessageConsumer? notifiedConsumer = null; + IMessageConsumer? received = null; + + await wrapper.GetChannelAndDoActionWithRetryAsync( + (c, _) => + { + received = c; + return Task.CompletedTask; + }, + conn => notifiedConnection = conn, + c => notifiedConsumer = c, + TestContext.Current.CancellationToken); + + Assert.Same(consumer.Object, received); + Assert.Same(connection.Object, notifiedConnection); + Assert.Same(consumer.Object, notifiedConsumer); + connection.Verify(c => c.StartAsync(), Times.Once); + factory.Verify(f => f.GetConnectionAsync(false, TestContext.Current.CancellationToken), Times.Once); + } + + private sealed class ImmediateRetryWrapper(int maxAttempts = 1) : IActiveMqRetryWrapperService + { + public Task RunAsync(Func> func, + CancellationToken cancellationToken = default) + { + throw new NotSupportedException(); + } + + public Task RunAsync(Func> func, + TState state, CancellationToken cancellationToken = default) + { + throw new NotSupportedException(); + } + + public Task RunAsync(Func func, CancellationToken cancellationToken = default) + { + throw new NotSupportedException(); + } + + public async Task RunAsync(Func func, TState state, + CancellationToken cancellationToken = default) + { + Exception? last = null; + for (var attempt = 0; attempt < maxAttempts; attempt++) + { + try + { + await func(state, cancellationToken); + return; + } + catch (Exception e) + { + last = e; + } + } + + ExceptionDispatchInfo.Capture(last!).Throw(); + } + } +} \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs index 7f0ebf7c..871203b9 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqJobSourceTests.cs @@ -3,8 +3,7 @@ using Microsoft.Extensions.Options; using RedShirt.Example.JobWorker.Core.Enums; using RedShirt.Example.JobWorker.Core.Models; -using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Exceptions; -using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Factories; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Configuration; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Models; using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; @@ -12,28 +11,46 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Serv public class ActiveMqJobSourceTests { - private static ActiveMqJobSource CreateJobSource( - IActiveMqConnectionFactory? factory, - ActiveMqJobSource.ConfigurationModel configuration) + private static (ActiveMqJobSource JobSource, Mock ConsumerRetryWrapper) + CreateJobSource(IMessageConsumer consumer, string? queueName = null) { - return new ActiveMqJobSource( - factory!, + var consumerRetryWrapper = new Mock(MockBehavior.Strict); + consumerRetryWrapper + .Setup(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny())) + .Returns((Func callback, Action? _, + Action? __, CancellationToken token) => callback(consumer, token)); + + var jobSource = new ActiveMqJobSource( ActiveMqRetryTestHelpers.CreatePassthroughRetryWrapper().Object, - Options.Create(configuration), + consumerRetryWrapper.Object, + Options.Create(new ActiveMqConfigurationModel + { + QueueName = queueName! + }), new NullLogger()); + + return (jobSource, consumerRetryWrapper); + } + + private static void VerifyWrapperCalled(Mock consumerRetryWrapper, Times times) + { + consumerRetryWrapper.Verify(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + TestContext.Current.CancellationToken), times); } [Fact] public async Task Test_AcknowledgeAsync() { var message = new Mock(); - - var configuration = new ActiveMqJobSource.ConfigurationModel - { - QueueName = null! - }; - - var activeMqJobSource = CreateJobSource(null, configuration); + var consumer = new Mock(MockBehavior.Strict); + var (jobSource, _) = CreateJobSource(consumer.Object); var jobModel = new ActiveMqRawJobModel { @@ -42,7 +59,7 @@ public async Task Test_AcknowledgeAsync() CreatedAtUtc = DateTime.UtcNow }; - await activeMqJobSource.AcknowledgeAsync(jobModel, CoreJobResult.Success, + await jobSource.AcknowledgeAsync(jobModel, CoreJobResult.Success, TestContext.Current.CancellationToken); message.Verify(m => m.AcknowledgeAsync(), Times.Once); @@ -58,13 +75,8 @@ await activeMqJobSource.AcknowledgeAsync(jobModel, CoreJobResult.Success, public async Task Test_AcknowledgeAsync_AlwaysAcknowledges(CoreJobResult result) { var message = new Mock(); - - var configuration = new ActiveMqJobSource.ConfigurationModel - { - QueueName = null! - }; - - var activeMqJobSource = CreateJobSource(null, configuration); + var consumer = new Mock(MockBehavior.Strict); + var (jobSource, _) = CreateJobSource(consumer.Object); var jobModel = new ActiveMqRawJobModel { @@ -73,7 +85,7 @@ public async Task Test_AcknowledgeAsync_AlwaysAcknowledges(CoreJobResult result) CreatedAtUtc = DateTime.UtcNow }; - await activeMqJobSource.AcknowledgeAsync(jobModel, result, + await jobSource.AcknowledgeAsync(jobModel, result, TestContext.Current.CancellationToken); message.Verify(m => m.AcknowledgeAsync(), Times.Once); @@ -83,135 +95,52 @@ await activeMqJobSource.AcknowledgeAsync(jobModel, result, public async Task Test_AcknowledgeAsync_Incompatible() { var job = new Mock(); + var consumer = new Mock(MockBehavior.Strict); + var consumerRetryWrapper = new Mock(MockBehavior.Strict); - var configuration = new ActiveMqJobSource.ConfigurationModel - { - QueueName = null! - }; - - var activeMqJobSource = CreateJobSource(null, configuration); + var jobSource = new ActiveMqJobSource( + ActiveMqRetryTestHelpers.CreatePassthroughRetryWrapper().Object, + consumerRetryWrapper.Object, + Options.Create(new ActiveMqConfigurationModel + { + QueueName = null! + }), + new NullLogger()); - await activeMqJobSource.AcknowledgeAsync(job.Object, CoreJobResult.Success, + await jobSource.AcknowledgeAsync(job.Object, CoreJobResult.Success, TestContext.Current.CancellationToken); + + consumerRetryWrapper.Verify(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny()), Times.Never); + Assert.Empty(consumer.Invocations); } [Fact] public async Task Test_GetJobs_GetNoJobs() { var queueName = Guid.NewGuid().ToString(); - - var configuration = new ActiveMqJobSource.ConfigurationModel - { - QueueName = queueName - }; - var consumer = new Mock(MockBehavior.Strict); - - var queue = new Mock(MockBehavior.Strict); - - var mockSession = new Mock(MockBehavior.Strict); - mockSession.Setup(s => s.GetQueueAsync(queueName)) - .ReturnsAsync(queue.Object); - mockSession.Setup(s => s.CreateConsumerAsync(queue.Object)) - .ReturnsAsync(consumer.Object); - - var mockConnection = new Mock(MockBehavior.Strict); - mockConnection.Setup(c => c.StartAsync()) - .Returns(Task.CompletedTask); - mockConnection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) - .ReturnsAsync(mockSession.Object); - - var activeConnectionFactory = new Mock(MockBehavior.Strict); - activeConnectionFactory.Setup(f => f.GetConnectionAsync(It.IsAny())) - .ReturnsAsync(mockConnection.Object); - consumer .Setup(c => c.ReceiveAsync(It.IsAny())) .ReturnsAsync(() => null); - var jobSource = CreateJobSource(activeConnectionFactory.Object, configuration); + var (jobSource, consumerRetryWrapper) = CreateJobSource(consumer.Object, queueName); var jobResponse = await jobSource.GetJobsAsync(10, TestContext.Current.CancellationToken); Assert.Equal(0, jobSource.RecommendedHeartbeatIntervalSeconds); Assert.Empty(jobResponse.Items); - - Assert.Single(activeConnectionFactory.Invocations); - Assert.Equal(2, mockConnection.Invocations.Count); - mockConnection.Verify(c => c.StartAsync(), Times.Once); - mockConnection.Verify(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge), Times.Once); - Assert.Equal(2, mockSession.Invocations.Count); - mockSession.Verify(s => s.GetQueueAsync(queueName), Times.Once); - mockSession.Verify(s => s.CreateConsumerAsync(queue.Object), Times.Once); - } - - [Fact] - public async Task Test_GetJobs_GetNoQueue() - { - var queueName = Guid.NewGuid().ToString(); - - var configuration = new ActiveMqJobSource.ConfigurationModel - { - QueueName = queueName - }; - - var mockSession = new Mock(MockBehavior.Strict); - mockSession.Setup(s => s.GetQueueAsync(queueName)) - .ReturnsAsync((IQueue?) null); - - var mockConnection = new Mock(MockBehavior.Strict); - mockConnection.Setup(c => c.StartAsync()) - .Returns(Task.CompletedTask); - mockConnection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) - .ReturnsAsync(mockSession.Object); - - var activeConnectionFactory = new Mock(MockBehavior.Strict); - activeConnectionFactory.Setup(f => f.GetConnectionAsync(It.IsAny())) - .ReturnsAsync(mockConnection.Object); - - var jobSource = CreateJobSource(activeConnectionFactory.Object, configuration); - - await Assert.ThrowsAsync(() => - jobSource.GetJobsAsync(1, TestContext.Current.CancellationToken)); - - Assert.Single(activeConnectionFactory.Invocations); - Assert.Equal(2, mockConnection.Invocations.Count); - mockConnection.Verify(c => c.StartAsync(), Times.Once); - mockConnection.Verify(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge), Times.Once); - Assert.Single(mockSession.Invocations); - mockSession.Verify(s => s.GetQueueAsync(queueName), Times.Once); + VerifyWrapperCalled(consumerRetryWrapper, Times.Once()); + consumer.Verify(c => c.ReceiveAsync(TimeSpan.FromMilliseconds(100)), Times.Once); } [Fact] public async Task Test_GetJobs_GotJob() { var queueName = Guid.NewGuid().ToString(); - - var configuration = new ActiveMqJobSource.ConfigurationModel - { - QueueName = queueName - }; - - var consumer = new Mock(MockBehavior.Strict); - - var queue = new Mock(MockBehavior.Strict); - - var mockSession = new Mock(MockBehavior.Strict); - mockSession.Setup(s => s.GetQueueAsync(queueName)) - .ReturnsAsync(queue.Object); - mockSession.Setup(s => s.CreateConsumerAsync(queue.Object)) - .ReturnsAsync(consumer.Object); - - var mockConnection = new Mock(MockBehavior.Strict); - mockConnection.Setup(c => c.StartAsync()) - .Returns(Task.CompletedTask); - mockConnection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) - .ReturnsAsync(mockSession.Object); - - var activeConnectionFactory = new Mock(MockBehavior.Strict); - activeConnectionFactory.Setup(f => f.GetConnectionAsync(It.IsAny())) - .ReturnsAsync(mockConnection.Object); - var messageId = Guid.NewGuid().ToString(); var mockMessage = new Mock(); mockMessage.Setup(m => m.NMSMessageId).Returns(messageId); @@ -220,11 +149,12 @@ public async Task Test_GetJobs_GotJob() var mockChannelQueue = new Queue(); mockChannelQueue.Enqueue(mockMessage.Object); + var consumer = new Mock(MockBehavior.Strict); consumer .Setup(c => c.ReceiveAsync(It.IsAny())) .ReturnsAsync(() => mockChannelQueue.TryDequeue(out var job) ? job : null); - var jobSource = CreateJobSource(activeConnectionFactory.Object, configuration); + var (jobSource, consumerRetryWrapper) = CreateJobSource(consumer.Object, queueName); var jobResponse = await jobSource.GetJobsAsync(1, TestContext.Current.CancellationToken); @@ -232,46 +162,13 @@ public async Task Test_GetJobs_GotJob() var returnedJobItem = Assert.Single(jobResponse.Items); Assert.Equal(messageId, returnedJobItem.MessageId); Assert.Equal("{}", returnedJobItem.Body); - - Assert.Single(activeConnectionFactory.Invocations); - Assert.Equal(2, mockConnection.Invocations.Count); - mockConnection.Verify(c => c.StartAsync(), Times.Once); - mockConnection.Verify(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge), Times.Once); - Assert.Equal(2, mockSession.Invocations.Count); - mockSession.Verify(s => s.GetQueueAsync(queueName), Times.Once); - mockSession.Verify(s => s.CreateConsumerAsync(queue.Object), Times.Once); + VerifyWrapperCalled(consumerRetryWrapper, Times.Once()); } [Fact] public async Task Test_GetJobs_GotJob_BatchSizeZero() { var queueName = Guid.NewGuid().ToString(); - - var configuration = new ActiveMqJobSource.ConfigurationModel - { - QueueName = queueName - }; - - var consumer = new Mock(MockBehavior.Strict); - - var queue = new Mock(MockBehavior.Strict); - - var mockSession = new Mock(MockBehavior.Strict); - mockSession.Setup(s => s.GetQueueAsync(queueName)) - .ReturnsAsync(queue.Object); - mockSession.Setup(s => s.CreateConsumerAsync(queue.Object)) - .ReturnsAsync(consumer.Object); - - var mockConnection = new Mock(MockBehavior.Strict); - mockConnection.Setup(c => c.StartAsync()) - .Returns(Task.CompletedTask); - mockConnection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) - .ReturnsAsync(mockSession.Object); - - var activeConnectionFactory = new Mock(MockBehavior.Strict); - activeConnectionFactory.Setup(f => f.GetConnectionAsync(It.IsAny())) - .ReturnsAsync(mockConnection.Object); - var messageId = Guid.NewGuid().ToString(); var mockMessage = new Mock(); mockMessage.Setup(m => m.NMSMessageId).Returns(messageId); @@ -280,11 +177,12 @@ public async Task Test_GetJobs_GotJob_BatchSizeZero() var mockChannelQueue = new Queue(); mockChannelQueue.Enqueue(mockMessage.Object); + var consumer = new Mock(MockBehavior.Strict); consumer .Setup(c => c.ReceiveAsync(It.IsAny())) .ReturnsAsync(() => mockChannelQueue.TryDequeue(out var job) ? job : null); - var jobSource = CreateJobSource(activeConnectionFactory.Object, configuration); + var (jobSource, consumerRetryWrapper) = CreateJobSource(consumer.Object, queueName); var jobResponse = await jobSource.GetJobsAsync(0, TestContext.Current.CancellationToken); @@ -292,6 +190,8 @@ public async Task Test_GetJobs_GotJob_BatchSizeZero() var returnedJobItem = Assert.Single(jobResponse.Items); Assert.Equal(messageId, returnedJobItem.MessageId); Assert.Equal("{}", returnedJobItem.Body); + // batchSize is floored to 1, so a single successful receive ends the loop. + VerifyWrapperCalled(consumerRetryWrapper, Times.Once()); } [Theory] @@ -301,36 +201,8 @@ public async Task Test_GetJobs_GotJob_BatchSizeZero() public async Task Test_GetJobs_GotJobs_MultipleJobs(int batchSize) { var queueName = Guid.NewGuid().ToString(); - - var configuration = new ActiveMqJobSource.ConfigurationModel - { - QueueName = queueName - }; - - var consumer = new Mock(MockBehavior.Strict); - - var queue = new Mock(MockBehavior.Strict); - - var mockSession = new Mock(MockBehavior.Strict); - mockSession.Setup(s => s.GetQueueAsync(queueName)) - .ReturnsAsync(queue.Object); - mockSession.Setup(s => s.CreateConsumerAsync(queue.Object)) - .ReturnsAsync(consumer.Object); - - var mockConnection = new Mock(MockBehavior.Strict); - mockConnection.Setup(c => c.StartAsync()) - .Returns(Task.CompletedTask); - mockConnection.Setup(c => c.CreateSessionAsync(AcknowledgementMode.ClientAcknowledge)) - .ReturnsAsync(mockSession.Object); - - var activeConnectionFactory = new Mock(MockBehavior.Strict); - activeConnectionFactory.Setup(f => f.GetConnectionAsync(It.IsAny())) - .ReturnsAsync(mockConnection.Object); - var messageIds = new List(); - var mockMessages = new List>(); var bodyStrings = new List(); - var mockChannelQueue = new Queue(); for (var i = 0; i < batchSize; i++) @@ -342,42 +214,63 @@ public async Task Test_GetJobs_GotJobs_MultipleJobs(int batchSize) var mockMessage = new Mock(); mockMessage.Setup(m => m.NMSMessageId).Returns(messageId); mockMessage.Setup(m => m.Text).Returns(body); - - mockMessages.Add(mockMessage); mockChannelQueue.Enqueue(mockMessage.Object); } + var consumer = new Mock(MockBehavior.Strict); consumer .Setup(c => c.ReceiveAsync(It.IsAny())) .ReturnsAsync(() => mockChannelQueue.TryDequeue(out var job) ? job : null); - var jobSource = CreateJobSource(activeConnectionFactory.Object, configuration); + var (jobSource, consumerRetryWrapper) = CreateJobSource(consumer.Object, queueName); var jobResponse = await jobSource.GetJobsAsync(batchSize, TestContext.Current.CancellationToken); Assert.Equal(0, jobSource.RecommendedHeartbeatIntervalSeconds); Assert.Equal(batchSize, jobResponse.Items.Count); + VerifyWrapperCalled(consumerRetryWrapper, Times.Exactly(batchSize)); for (var i = 0; i < batchSize; i++) { - var messageId = messageIds[i]; - var body = bodyStrings[i]; - - var returnedJobItem = jobResponse.Items[i]; - Assert.Equal(messageId, returnedJobItem.MessageId); - Assert.Equal(body, returnedJobItem.Body); + Assert.Equal(messageIds[i], jobResponse.Items[i].MessageId); + Assert.Equal(bodyStrings[i], jobResponse.Items[i].Body); } } [Fact] - public async Task Test_HeartbeatAsync() + public async Task Test_GetJobs_PropagatesConsumerWrapperException() { - var configuration = new ActiveMqJobSource.ConfigurationModel - { - QueueName = null! - }; + var queueName = Guid.NewGuid().ToString(); + var consumer = new Mock(MockBehavior.Strict); + var consumerRetryWrapper = new Mock(MockBehavior.Strict); + consumerRetryWrapper + .Setup(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny())) + .ThrowsAsync(new InvalidOperationException("boom")); + + var jobSource = new ActiveMqJobSource( + ActiveMqRetryTestHelpers.CreatePassthroughRetryWrapper().Object, + consumerRetryWrapper.Object, + Options.Create(new ActiveMqConfigurationModel + { + QueueName = queueName + }), + new NullLogger()); + + await Assert.ThrowsAsync(() => + jobSource.GetJobsAsync(1, TestContext.Current.CancellationToken)); - var jobSource = CreateJobSource(null, configuration); + Assert.Empty(consumer.Invocations); + } + + [Fact] + public async Task Test_HeartbeatAsync() + { + var consumer = new Mock(MockBehavior.Strict); + var (jobSource, _) = CreateJobSource(consumer.Object); await jobSource.HeartbeatAsync(null!, TestContext.Current.CancellationToken); } diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs index f31b453b..746ee998 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqRetryTestHelpers.cs @@ -13,6 +13,35 @@ private static void SetupPassthrough(Mock retry .Returns>, CancellationToken>((func, token) => func(token)); } + private static void SetupStatePassthrough(Mock retry) + { + retry + .Setup(r => r.RunAsync( + It.IsAny>(), + It.IsAny(), + It.IsAny())) + .Returns(new InvocationFunc(invocation => + { + var func = (Delegate) invocation.Arguments[0]; + var state = invocation.Arguments[1]; + var token = (CancellationToken) invocation.Arguments[2]; + return (Task) func.DynamicInvoke(state, token)!; + })); + + retry + .Setup(r => r.RunAsync( + It.IsAny>>(), + It.IsAny(), + It.IsAny())) + .Returns(new InvocationFunc(invocation => + { + var func = (Delegate) invocation.Arguments[0]; + var state = invocation.Arguments[1]; + var token = (CancellationToken) invocation.Arguments[2]; + return (Task) func.DynamicInvoke(state, token)!; + })); + } + public static Mock CreatePassthroughRetryWrapper() { var retry = new Mock(MockBehavior.Strict); @@ -24,6 +53,7 @@ public static Mock CreatePassthroughRetryWrapper() SetupPassthrough(retry); SetupPassthrough(retry); SetupPassthrough(retry); + SetupStatePassthrough(retry); return retry; } diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqServerConfigurationSourceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqServerConfigurationSourceTests.cs index 5379c90d..bf3ab22c 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqServerConfigurationSourceTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqServerConfigurationSourceTests.cs @@ -45,7 +45,8 @@ public async Task GetConfigurationAsync_MapsBrokerUriAndResolvedSecrets() PasswordPath = passwordPath })); - var configuration = await source.GetConfigurationAsync(TestContext.Current.CancellationToken); + var configuration = await source.GetConfigurationAsync( + cancellationToken: TestContext.Current.CancellationToken); Assert.Equal(brokerUri, configuration.BrokerUri); Assert.Equal(user, configuration.User); @@ -57,4 +58,46 @@ public async Task GetConfigurationAsync_MapsBrokerUriAndResolvedSecrets() TestContext.Current.CancellationToken), Times.Once); secrets.VerifyNoOtherCalls(); } + + [Fact] + public async Task GetConfigurationAsync_WhenForceNewSecretManagerPull_PassesForceToCache() + { + var brokerUri = $"tcp://{Guid.NewGuid():N}:61616"; + var userPath = $"/activemq/{Guid.NewGuid():N}/user"; + var passwordPath = $"/activemq/{Guid.NewGuid():N}/password"; + + var secrets = new Mock(MockBehavior.Strict); + secrets + .Setup(s => s.GetSecretsAsync( + It.IsAny>(), + null, + true, + TestContext.Current.CancellationToken)) + .ReturnsAsync(new SecretManagerCacheSecretsResponse + { + Values = new Dictionary + { + [userPath] = "u", + [passwordPath] = "p" + }, + QueriedSecretManager = true + }); + + var source = new ActiveMqServerConfigurationSource( + secrets.Object, + Options.Create(new ActiveMqServerConfigurationSource.ConfigurationModel + { + BrokerUri = brokerUri, + UserPath = userPath, + PasswordPath = passwordPath + })); + + await source.GetConfigurationAsync(true, TestContext.Current.CancellationToken); + + secrets.Verify(s => s.GetSecretsAsync( + It.IsAny>(), + null, + true, + TestContext.Current.CancellationToken), Times.Once); + } } \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqSubscribeJobSourceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqSubscribeJobSourceTests.cs new file mode 100644 index 00000000..fb440758 --- /dev/null +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/ActiveMqSubscribeJobSourceTests.cs @@ -0,0 +1,796 @@ +using Apache.NMS; +using Apache.NMS.ActiveMQ; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using RedShirt.Example.JobWorker.Common.Services.Utility; +using RedShirt.Example.JobWorker.Core.Enums; +using RedShirt.Example.JobWorker.Core.Exceptions; +using RedShirt.Example.JobWorker.Core.Models; +using RedShirt.Example.JobWorker.Core.Services.Configuration; +using RedShirt.Example.JobWorker.Core.Services.ExecutionState; +using RedShirt.Example.JobWorker.Core.Services.Jobs.Subscriptions; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Configuration; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Models; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; +using System.Reflection; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Services; + +public class ActiveMqSubscribeJobSourceTests +{ + private const string QueueName = "jobs"; + + private static ActiveMqSubscribeJobSource CreateJobSource( + Mock consumerRetryWrapper, + Mock? intakeQueue = null, + Mock? executionEndArbiter = null, + Mock? sleepService = null, + ILogger? logger = null, + IActiveMqSubscribeExceptionArbiter? subscribeExceptionArbiter = null, + bool haltOnFailure = true, + bool treatTransientExceptionAsFailure = false) + { + var coreConfiguration = new Mock(MockBehavior.Strict); + coreConfiguration.Setup(c => c.IsHaltOnFailure()).Returns(haltOnFailure); + coreConfiguration.Setup(c => c.IsTreatingTransientExceptionAsFailure()) + .Returns(treatTransientExceptionAsFailure); + + sleepService ??= new Mock(MockBehavior.Strict); + sleepService + .Setup(s => s.DelayAsync(TimeSpan.FromSeconds(1), It.IsAny())) + .Returns(Task.CompletedTask); + + if (executionEndArbiter is null) + { + executionEndArbiter = new Mock(MockBehavior.Strict); + // Background unsubscribe waiter; leave unfinished unless a test signals stop. + executionEndArbiter + .Setup(a => a.WaitForFinishedAsync(It.IsAny())) + .Returns(new TaskCompletionSource().Task); + } + + return new ActiveMqSubscribeJobSource( + ActiveMqRetryTestHelpers.CreatePassthroughRetryWrapper().Object, + consumerRetryWrapper.Object, + coreConfiguration.Object, + (intakeQueue ?? new Mock(MockBehavior.Strict)).Object, + executionEndArbiter.Object, + sleepService.Object, + subscribeExceptionArbiter ?? new ActiveMqSubscribeExceptionArbiterService(), + Options.Create(new ActiveMqConfigurationModel {QueueName = QueueName}), + logger ?? NullLogger.Instance); + } + + private static Mock CreatePassthroughWrapper(IMessageConsumer consumer, + Action?>? captureOnNewConnection = null) + { + var wrapper = new Mock(MockBehavior.Strict); + wrapper + .Setup(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny())) + .Returns((Func callback, Action? onNew, + Action? _, CancellationToken token) => + { + captureOnNewConnection?.Invoke(onNew); + return callback(consumer, token); + }); + return wrapper; + } + + private static void SetupAsyncListener(Mock consumer, + Action? captureListener = null) + { + consumer + .SetupAdd(c => c.AsyncListener += It.IsAny()) + .Callback(handler => captureListener?.Invoke(handler)); + consumer.SetupRemove(c => c.AsyncListener -= It.IsAny()); + } + + private static (Mock Connection, Func GetExceptionHandler, + Func GetResumedHandler) CreateConnectionCapturingListeners() + { + var connection = new Mock(); + ExceptionListener? exceptionHandler = null; + ConnectionResumedListener? resumedHandler = null; + + connection + .SetupAdd(c => c.ExceptionListener += It.IsAny()) + .Callback(handler => exceptionHandler += handler); + connection.SetupRemove(c => c.ExceptionListener -= It.IsAny()); + connection + .SetupAdd(c => c.ConnectionResumedListener += It.IsAny()) + .Callback(handler => resumedHandler += handler); + connection.SetupRemove(c => c.ConnectionResumedListener -= It.IsAny()); + + return (connection, () => exceptionHandler, () => resumedHandler); + } + + private static Task InvokeWaitThenStopSubscriberAsync(ActiveMqSubscribeJobSource jobSource, + CancellationToken cancellationToken) + { + var method = typeof(ActiveMqSubscribeJobSource).GetMethod("WaitThenStopSubscriberAsync", + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.NotNull(method); + return (Task) method.Invoke(jobSource, [cancellationToken])!; + } + + private static void SetSubscribeLoopRunning(ActiveMqSubscribeJobSource jobSource, bool value) + { + var field = typeof(ActiveMqSubscribeJobSource).GetField("_subscribeLoopRunning", + BindingFlags.Instance | BindingFlags.NonPublic); + Assert.NotNull(field); + field.SetValue(jobSource, value); + } + + [Theory] + [InlineData(CoreJobResult.Success)] + [InlineData(CoreJobResult.Failure)] + [InlineData(CoreJobResult.Empty)] + public async Task AcknowledgeAsync_AlwaysAcknowledges(CoreJobResult result) + { + var message = new Mock(MockBehavior.Strict); + message.Setup(m => m.AcknowledgeAsync()).Returns(Task.CompletedTask); + + var consumer = new Mock(MockBehavior.Strict); + var wrapper = CreatePassthroughWrapper(consumer.Object); + var jobSource = CreateJobSource(wrapper); + + await jobSource.AcknowledgeAsync(new ActiveMqRawJobModel + { + Message = message.Object, + MessageId = "m", + CreatedAtUtc = DateTime.UtcNow + }, result, TestContext.Current.CancellationToken); + + message.Verify(m => m.AcknowledgeAsync(), Times.Once); + } + + [Fact] + public async Task AcknowledgeAsync_WhenIncompatibleMessage_DoesNotAck() + { + var message = new Mock(MockBehavior.Strict); + var consumer = new Mock(MockBehavior.Strict); + var wrapper = CreatePassthroughWrapper(consumer.Object); + var jobSource = CreateJobSource(wrapper); + + await jobSource.AcknowledgeAsync(new Mock().Object, CoreJobResult.Success, + TestContext.Current.CancellationToken); + + message.Verify(m => m.AcknowledgeAsync(), Times.Never); + wrapper.Verify(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny()), Times.Never); + } + + public static TheoryData ExceptionListenerReconnectExceptions() + { + return + [ + new EndOfStreamException("peer closed"), + new NMSSecurityException("bad credentials") + ]; + } + + [Fact] + public async Task GetJobsAsync_ThrowsNotSupportedException() + { + var consumer = new Mock(MockBehavior.Strict); + var jobSource = CreateJobSource(CreatePassthroughWrapper(consumer.Object)); + + await Assert.ThrowsAsync(() => + jobSource.GetJobsAsync(1, TestContext.Current.CancellationToken)); + } + + [Fact] + public async Task HeartbeatAsync_Completes() + { + var consumer = new Mock(MockBehavior.Strict); + var jobSource = CreateJobSource(CreatePassthroughWrapper(consumer.Object)); + + await jobSource.HeartbeatAsync(null!, TestContext.Current.CancellationToken); + Assert.True(true); + } + + [Fact] + public void IsSubscriptionSource_IsTrue() + { + var consumer = new Mock(MockBehavior.Strict); + var jobSource = CreateJobSource(CreatePassthroughWrapper(consumer.Object)); + + Assert.True(jobSource.IsSubscriptionSource); + Assert.Equal(0, jobSource.RecommendedHeartbeatIntervalSeconds); + } + + [Fact] + public async Task StartSubscriberAsync_AttachesAsyncListenerAndWaitsForFinished() + { + var consumer = new Mock(MockBehavior.Strict); + SetupAsyncListener(consumer); + + var wrapper = CreatePassthroughWrapper(consumer.Object); + var waitStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var executionEndArbiter = new Mock(MockBehavior.Strict); + var waitForFinishedExecuted = false; + executionEndArbiter + .Setup(a => a.WaitForFinishedAsync(It.IsAny())) + .Returns(() => + { + waitStarted.TrySetResult(); + waitForFinishedExecuted = true; + return Task.CompletedTask; + }); + + var jobSource = CreateJobSource(wrapper, executionEndArbiter: executionEndArbiter); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + await waitStarted.Task.WaitAsync(TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken); + Assert.True(waitForFinishedExecuted); + executionEndArbiter.Verify(a => a.WaitForFinishedAsync(It.IsAny()), Times.Once); + consumer.VerifyAdd(c => c.AsyncListener += It.IsAny(), Times.Once); + } + + [Fact] + public async Task StartSubscriberAsync_WhenConnectionResumed_LogsAndDoesNotResubscribe() + { + var consumer = new Mock(MockBehavior.Strict); + SetupAsyncListener(consumer); + + var (connection, _, getResumedHandler) = CreateConnectionCapturingListeners(); + + var logger = new Mock>(); + logger.Setup(l => l.IsEnabled(LogLevel.Information)).Returns(true); + + var wrapper = CreatePassthroughWrapper(consumer.Object, onNew => onNew?.Invoke(connection.Object)); + var jobSource = CreateJobSource(wrapper, logger: logger.Object); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + var resumedHandler = getResumedHandler(); + Assert.NotNull(resumedHandler); + resumedHandler!(); + + logger.Verify( + l => l.Log( + LogLevel.Information, + It.IsAny(), + It.Is((v, _) => + v.ToString()!.Contains("established", StringComparison.OrdinalIgnoreCase)), + It.IsAny(), + It.IsAny>()), + Times.Once); + // ActiveMQ client library keeps the listener; we only log on resume. + consumer.VerifyAdd(c => c.AsyncListener += It.IsAny(), Times.Once); + } + + [Fact] + public async Task + StartSubscriberAsync_WhenExceptionListenerPermanentErrorAndReconnectFailsWithHaltOnFailure_Stops() + { + var consumer = new Mock(MockBehavior.Strict); + SetupAsyncListener(consumer); + + var (connection, getExceptionHandler, _) = CreateConnectionCapturingListeners(); + + var stopped = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var executionEndArbiter = new Mock(MockBehavior.Strict); + executionEndArbiter + .Setup(a => a.WaitForFinishedAsync(It.IsAny())) + .Returns(new TaskCompletionSource().Task); + executionEndArbiter + .Setup(a => a.Stop(It.IsAny())) + .Callback(() => stopped.TrySetResult()); + + var subscribeCalls = 0; + var wrapper = new Mock(MockBehavior.Strict); + wrapper + .Setup(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny())) + .Returns((Func callback, Action? onNew, + Action? _, CancellationToken token) => + { + subscribeCalls++; + if (subscribeCalls == 1) + { + onNew?.Invoke(connection.Object); + return callback(consumer.Object, token); + } + + return Task.FromException(new WorkerJobSourceException("still unauthorized") + { + CouldBeTransient = false, + IsHandled = true, + CouldBeExternallySolvable = false + }); + }); + wrapper.Setup(w => w.ResetConsumer()); + + var jobSource = CreateJobSource(wrapper, executionEndArbiter: executionEndArbiter, haltOnFailure: true); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + var exceptionHandler = getExceptionHandler(); + Assert.NotNull(exceptionHandler); + exceptionHandler!(new NMSSecurityException("bad credentials")); + + await stopped.Task.WaitAsync(TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken); + + wrapper.Verify(w => w.ResetConsumer(), Times.Once); + executionEndArbiter.Verify(a => a.Stop(It.IsAny()), Times.Once); + } + + [Fact] + public async Task StartSubscriberAsync_WhenExceptionListenerReportsAccountedTransient_DoesNotWarnUnaccounted() + { + var consumer = new Mock(MockBehavior.Strict); + SetupAsyncListener(consumer); + + var (connection, getExceptionHandler, _) = CreateConnectionCapturingListeners(); + + var logger = new Mock>(); + logger.Setup(l => l.IsEnabled(It.IsAny())).Returns(true); + + var wrapper = CreatePassthroughWrapper(consumer.Object, onNew => onNew?.Invoke(connection.Object)); + var jobSource = CreateJobSource(wrapper, logger: logger.Object); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + var exceptionHandler = getExceptionHandler(); + Assert.NotNull(exceptionHandler); + exceptionHandler!(new BrokerException()); + + logger.Verify( + l => l.Log( + LogLevel.Warning, + It.IsAny(), + It.Is((v, _) => + v.ToString()!.Contains("Unaccounted-for", StringComparison.OrdinalIgnoreCase)), + It.IsAny(), + It.IsAny>()), + Times.Never); + wrapper.Verify(w => w.ResetConsumer(), Times.Never); + } + + [Fact] + public async Task + StartSubscriberAsync_WhenExceptionListenerReportsProblemWhileSubscribeLoopRunning_DoesNotReconnect() + { + var consumer = new Mock(MockBehavior.Strict); + SetupAsyncListener(consumer); + + var (connection, getExceptionHandler, _) = CreateConnectionCapturingListeners(); + var wrapper = CreatePassthroughWrapper(consumer.Object, onNew => onNew?.Invoke(connection.Object)); + wrapper.Setup(w => w.ResetConsumer()); + + var jobSource = CreateJobSource(wrapper); + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + SetSubscribeLoopRunning(jobSource, true); + + var exceptionHandler = getExceptionHandler(); + Assert.NotNull(exceptionHandler); + exceptionHandler!(new EndOfStreamException("peer closed")); + + // Early return before ResetConsumer / Task.Run when a subscribe loop is already in flight. + wrapper.Verify(w => w.ResetConsumer(), Times.Never); + wrapper.Verify(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny()), Times.Once); + } + + [Theory] + [MemberData(nameof(ExceptionListenerReconnectExceptions))] + public async Task StartSubscriberAsync_WhenExceptionListenerReportsReconnectOrStopWorthy_Reconnects( + Exception exception) + { + var consumer = new Mock(MockBehavior.Strict); + SetupAsyncListener(consumer); + + var (connection, getExceptionHandler, _) = CreateConnectionCapturingListeners(); + + var logger = new Mock>(); + logger.Setup(l => l.IsEnabled(LogLevel.Warning)).Returns(true); + + var subscribeCalls = 0; + var reSubscribeStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var wrapper = new Mock(MockBehavior.Strict); + wrapper + .Setup(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny())) + .Returns((Func callback, Action? onNew, + Action? _, CancellationToken token) => + { + subscribeCalls++; + if (subscribeCalls == 1) + { + onNew?.Invoke(connection.Object); + } + else + { + reSubscribeStarted.TrySetResult(); + } + + return callback(consumer.Object, token); + }); + wrapper.Setup(w => w.ResetConsumer()); + + var executionEndArbiter = new Mock(MockBehavior.Strict); + executionEndArbiter + .Setup(a => a.WaitForFinishedAsync(It.IsAny())) + .Returns(new TaskCompletionSource().Task); + + var jobSource = CreateJobSource(wrapper, executionEndArbiter: executionEndArbiter, logger: logger.Object); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + var exceptionHandler = getExceptionHandler(); + Assert.NotNull(exceptionHandler); + exceptionHandler!(exception); + + await reSubscribeStarted.Task.WaitAsync(TimeSpan.FromSeconds(2), TestContext.Current.CancellationToken); + + logger.Verify( + l => l.Log( + LogLevel.Warning, + It.IsAny(), + It.Is((v, _) => + v.ToString()!.Contains("ExceptionListener problem", StringComparison.OrdinalIgnoreCase)), + It.IsAny(), + It.IsAny>()), + Times.Once); + wrapper.Verify(w => w.ResetConsumer(), Times.Once); + Assert.True(subscribeCalls >= 2); + // Reconnect succeeded; ExceptionListener itself does not call Stop. + executionEndArbiter.Verify(a => a.Stop(It.IsAny()), Times.Never); + } + + [Fact] + public async Task StartSubscriberAsync_WhenExceptionListenerReportsUnaccountedException_LogsWarning() + { + var consumer = new Mock(MockBehavior.Strict); + SetupAsyncListener(consumer); + + var (connection, getExceptionHandler, _) = CreateConnectionCapturingListeners(); + + var logger = new Mock>(); + logger.Setup(l => l.IsEnabled(LogLevel.Warning)).Returns(true); + + var wrapper = CreatePassthroughWrapper(consumer.Object, onNew => onNew?.Invoke(connection.Object)); + var jobSource = CreateJobSource(wrapper, logger: logger.Object); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + var exceptionHandler = getExceptionHandler(); + Assert.NotNull(exceptionHandler); + exceptionHandler!(new Exception("mystery")); + + logger.Verify( + l => l.Log( + LogLevel.Warning, + It.IsAny(), + It.Is((v, _) => + v.ToString()!.Contains("Unaccounted-for", StringComparison.OrdinalIgnoreCase)), + It.IsAny(), + It.IsAny>()), + Times.Once); + } + + [Fact] + public async Task StartSubscriberAsync_WhenMessageIdMissing_UsesUnknown() + { + var consumer = new Mock(MockBehavior.Strict); + AsyncMessageListener? listener = null; + SetupAsyncListener(consumer, l => listener = l); + + var intakeQueue = new Mock(MockBehavior.Strict); + IJobSourceResponse? loaded = null; + intakeQueue + .Setup(q => q.Load(It.IsAny())) + .Callback(response => loaded = response); + + var wrapper = CreatePassthroughWrapper(consumer.Object); + var jobSource = CreateJobSource(wrapper, intakeQueue); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + var message = new Mock(MockBehavior.Strict); + message.SetupGet(m => m.NMSMessageId).Returns((string) null!); + message.SetupGet(m => m.Text).Returns("body"); + + await listener!(message.Object, TestContext.Current.CancellationToken); + + var job = Assert.IsType(Assert.Single(loaded!.Items)); + Assert.Equal("UNKNOWN", job.MessageId); + Assert.Equal("UNKNOWN", job.IdempotencyId); + Assert.Equal("body", job.Body); + } + + [Fact] + public async Task StartSubscriberAsync_WhenNonTransientAndHaltOnFailure_StopsArbiter() + { + var consumer = new Mock(MockBehavior.Strict); + var wrapper = new Mock(MockBehavior.Strict); + wrapper + .Setup(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny())) + .ThrowsAsync(new WorkerJobSourceException("permanent") + { + CouldBeTransient = false, + IsHandled = true, + CouldBeExternallySolvable = false + }); + + var executionEndArbiter = new Mock(MockBehavior.Strict); + executionEndArbiter + .Setup(a => a.WaitForFinishedAsync(It.IsAny())) + .Returns(new TaskCompletionSource().Task); + executionEndArbiter.Setup(a => a.Stop(It.IsAny())); + + var jobSource = CreateJobSource(wrapper, executionEndArbiter: executionEndArbiter, haltOnFailure: true); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + executionEndArbiter.Verify(a => a.Stop(It.IsAny()), Times.Once); + Assert.Empty(consumer.Invocations); + } + + [Fact] + public async Task StartSubscriberAsync_WhenOperationCanceled_ExitsWithoutStopping() + { + using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestContext.Current.CancellationToken); + await cts.CancelAsync(); + + var wrapper = new Mock(MockBehavior.Strict); + wrapper + .Setup(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny())) + .Returns((Func _, Action? __, + Action? ___, CancellationToken token) => + Task.FromException(new OperationCanceledException(token))); + + var executionEndArbiter = new Mock(MockBehavior.Strict); + executionEndArbiter + .Setup(a => a.WaitForFinishedAsync(It.IsAny())) + .Returns(new TaskCompletionSource().Task); + + var jobSource = CreateJobSource(wrapper, executionEndArbiter: executionEndArbiter); + + await jobSource.StartSubscriberAsync(cts.Token); + + executionEndArbiter.Verify(a => a.Stop(It.IsAny()), Times.Never); + } + + [Fact] + public async Task StartSubscriberAsync_WhenPermanentFailureAndNotHaltOnFailure_RetriesUntilSuccess() + { + var consumer = new Mock(MockBehavior.Strict); + SetupAsyncListener(consumer); + + var wrapper = new Mock(MockBehavior.Strict); + var attempts = 0; + wrapper + .Setup(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny())) + .Returns((Func callback, Action? _, + Action? __, CancellationToken token) => + { + attempts++; + if (attempts == 1) + { + return Task.FromException(new InvalidOperationException("permanent")); + } + + return callback(consumer.Object, token); + }); + + var jobSource = CreateJobSource(wrapper, haltOnFailure: false); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + Assert.Equal(2, attempts); + consumer.VerifyAdd(c => c.AsyncListener += It.IsAny(), Times.Once); + } + + [Fact] + public async Task StartSubscriberAsync_WhenReceiveFires_LoadsIntakeQueue() + { + var consumer = new Mock(MockBehavior.Strict); + AsyncMessageListener? listener = null; + SetupAsyncListener(consumer, l => listener = l); + + var intakeQueue = new Mock(MockBehavior.Strict); + IJobSourceResponse? loaded = null; + intakeQueue + .Setup(q => q.Load(It.IsAny())) + .Callback(response => loaded = response); + + var wrapper = CreatePassthroughWrapper(consumer.Object); + var jobSource = CreateJobSource(wrapper, intakeQueue); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + var message = new Mock(MockBehavior.Strict); + message.SetupGet(m => m.NMSMessageId).Returns("msg-1"); + message.SetupGet(m => m.Text).Returns("payload"); + + await listener!(message.Object, TestContext.Current.CancellationToken); + + var job = Assert.IsType(Assert.Single(loaded!.Items)); + Assert.Equal("msg-1", job.MessageId); + Assert.Equal("msg-1", job.IdempotencyId); + Assert.Equal("payload", job.Body); + } + + [Fact] + public async Task StartSubscriberAsync_WhenReceiveHandlerThrows_FaultsTask() + { + var consumer = new Mock(MockBehavior.Strict); + AsyncMessageListener? listener = null; + SetupAsyncListener(consumer, l => listener = l); + + var intakeQueue = new Mock(MockBehavior.Strict); + intakeQueue + .Setup(q => q.Load(It.IsAny())) + .Throws(new InvalidOperationException("intake failed")); + + var wrapper = CreatePassthroughWrapper(consumer.Object); + var jobSource = CreateJobSource(wrapper, intakeQueue); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + var message = new Mock(MockBehavior.Strict); + message.SetupGet(m => m.NMSMessageId).Returns("msg-1"); + + var faulted = listener!(message.Object, TestContext.Current.CancellationToken); + var exception = await Assert.ThrowsAsync(() => faulted); + Assert.Equal("intake failed", exception.Message); + } + + [Fact] + public async Task StartSubscriberAsync_WhenTransientFailure_RetriesUntilSuccess() + { + var consumer = new Mock(MockBehavior.Strict); + SetupAsyncListener(consumer); + + var wrapper = new Mock(MockBehavior.Strict); + var attempts = 0; + wrapper + .Setup(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny())) + .Returns((Func callback, Action? _, + Action? __, CancellationToken token) => + { + attempts++; + if (attempts == 1) + { + return Task.FromException(new WorkerJobSourceException("transient") + { + IsHandled = true, + CouldBeTransient = true, + CouldBeExternallySolvable = true + }); + } + + return callback(consumer.Object, token); + }); + + var jobSource = CreateJobSource(wrapper); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + Assert.Equal(2, attempts); + consumer.VerifyAdd(c => c.AsyncListener += It.IsAny(), Times.Once); + } + + [Fact] + public async Task StartSubscriberAsync_WhenTransientTreatedAsFailureAndHaltOnFailure_Stops() + { + var wrapper = new Mock(MockBehavior.Strict); + wrapper + .Setup(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny())) + .ThrowsAsync(new WorkerJobSourceException("transient") + { + IsHandled = true, + CouldBeTransient = true, + CouldBeExternallySolvable = true + }); + + var executionEndArbiter = new Mock(MockBehavior.Strict); + executionEndArbiter + .Setup(a => a.WaitForFinishedAsync(It.IsAny())) + .Returns(new TaskCompletionSource().Task); + executionEndArbiter.Setup(a => a.Stop(It.IsAny())); + + var jobSource = CreateJobSource(wrapper, executionEndArbiter: executionEndArbiter, haltOnFailure: true, + treatTransientExceptionAsFailure: true); + + await jobSource.StartSubscriberAsync(TestContext.Current.CancellationToken); + + executionEndArbiter.Verify(a => a.Stop(It.IsAny()), Times.Once); + } + + [Fact] + public async Task WaitThenStopSubscriberAsync_RemovesAsyncListener() + { + var consumer = new Mock(MockBehavior.Strict); + SetupAsyncListener(consumer); + + var executionEndArbiter = new Mock(MockBehavior.Strict); + executionEndArbiter + .Setup(a => a.WaitForFinishedAsync(It.IsAny())) + .Returns(Task.CompletedTask); + + var wrapper = CreatePassthroughWrapper(consumer.Object); + var jobSource = CreateJobSource(wrapper, executionEndArbiter: executionEndArbiter); + + await InvokeWaitThenStopSubscriberAsync(jobSource, TestContext.Current.CancellationToken); + + consumer.VerifyRemove(c => c.AsyncListener -= It.IsAny(), Times.Once); + } + + [Fact] + public async Task WaitThenStopSubscriberAsync_WhenUnsubscribeThrows_IsSwallowed() + { + var wrapper = new Mock(MockBehavior.Strict); + wrapper + .Setup(w => w.GetChannelAndDoActionWithRetryAsync( + It.IsAny>(), + It.IsAny?>(), + It.IsAny?>(), + It.IsAny())) + .ThrowsAsync(new InvalidOperationException("unsubscribe failed")); + + var logger = new Mock>(); + logger.Setup(l => l.IsEnabled(LogLevel.Error)).Returns(true); + + var executionEndArbiter = new Mock(MockBehavior.Strict); + executionEndArbiter + .Setup(a => a.WaitForFinishedAsync(It.IsAny())) + .Returns(Task.CompletedTask); + + var jobSource = CreateJobSource(wrapper, executionEndArbiter: executionEndArbiter, logger: logger.Object); + + await InvokeWaitThenStopSubscriberAsync(jobSource, TestContext.Current.CancellationToken); + + logger.Verify( + l => l.Log( + LogLevel.Error, + It.IsAny(), + It.Is((v, _) => + v.ToString()!.Contains("Could not unsubscribe", StringComparison.OrdinalIgnoreCase)), + It.IsAny(), + It.IsAny>()), + Times.Once); + } +} \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs index cc2a7c5c..1731a9f3 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqExceptionArbiterServiceTests.cs @@ -16,7 +16,7 @@ public class ActiveMqExceptionArbiterServiceTests [Fact] public void GetReport_ArgumentException_IsExpectedAndNotTransient() { - var report = _sut.GetReport(new ArgumentException("bad queue", "queue")); + var report = _sut.GetReport(new ArgumentException("bad queue", "queue"), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -27,7 +27,7 @@ public void GetReport_ArgumentException_IsExpectedAndNotTransient() [Fact] public void GetReport_ConnectionClosedException_IsExpectedAndTransient() { - var report = _sut.GetReport(new ConnectionClosedException("connection closed")); + var report = _sut.GetReport(new ConnectionClosedException("connection closed"), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -38,7 +38,7 @@ public void GetReport_ConnectionClosedException_IsExpectedAndTransient() [Fact] public void GetReport_CouldNotLoadQueueException_IsExpectedAndNotTransient() { - var report = _sut.GetReport(new CouldNotLoadQueueException()); + var report = _sut.GetReport(new CouldNotLoadQueueException(), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -49,7 +49,7 @@ public void GetReport_CouldNotLoadQueueException_IsExpectedAndNotTransient() [Fact] public void GetReport_CouldNotRetrieveMessageBodyException_IsExpectedAndNotTransient() { - var report = _sut.GetReport(new CouldNotRetrieveMessageBodyException()); + var report = _sut.GetReport(new CouldNotRetrieveMessageBodyException(), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -60,7 +60,7 @@ public void GetReport_CouldNotRetrieveMessageBodyException_IsExpectedAndNotTrans [Fact] public void GetReport_InvalidDestinationException_IsExpectedAndNotTransient() { - var report = _sut.GetReport(new InvalidDestinationException("no such queue")); + var report = _sut.GetReport(new InvalidDestinationException("no such queue"), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -71,7 +71,7 @@ public void GetReport_InvalidDestinationException_IsExpectedAndNotTransient() [Fact] public void GetReport_IoException_IsExpectedAndTransient() { - var report = _sut.GetReport(new ActiveMqIoException("transport failed")); + var report = _sut.GetReport(new ActiveMqIoException("transport failed"), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -86,7 +86,7 @@ public void GetReport_MultiInnerAggregateException_IsNotExpected() new NMSConnectionException("disconnected"), new SocketException((int) SocketError.TimedOut)); - var report = _sut.GetReport(exception); + var report = _sut.GetReport(exception, 1); Assert.False(report.AlreadyHandled); Assert.False(report.IsExpected); @@ -97,7 +97,7 @@ public void GetReport_MultiInnerAggregateException_IsNotExpected() [Fact] public void GetReport_NmsConnectionException_IsExpectedAndTransient() { - var report = _sut.GetReport(new NMSConnectionException("broker unavailable")); + var report = _sut.GetReport(new NMSConnectionException("broker unavailable"), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -108,7 +108,7 @@ public void GetReport_NmsConnectionException_IsExpectedAndTransient() [Fact] public void GetReport_NmsException_IsExpectedAndTransient() { - var report = _sut.GetReport(new NMSException("generic nms failure")); + var report = _sut.GetReport(new NMSException("generic nms failure"), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -116,10 +116,32 @@ public void GetReport_NmsException_IsExpectedAndTransient() Assert.True(report.CouldBeExternallySolvable); } + [Fact] + public void GetReport_NmsSecurityException_InvalidPassword_FirstAttempt_IsTransient() + { + var report = _sut.GetReport(new NMSSecurityException("User name or password is invalid."), 1); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.True(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + + [Fact] + public void GetReport_NmsSecurityException_InvalidPassword_LaterAttempt_IsNotTransient() + { + var report = _sut.GetReport(new NMSSecurityException("User name or password is invalid."), 2); + + Assert.False(report.AlreadyHandled); + Assert.True(report.IsExpected); + Assert.False(report.CouldBeTransient); + Assert.True(report.CouldBeExternallySolvable); + } + [Fact] public void GetReport_NmsSecurityException_IsExpectedAndNotTransient() { - var report = _sut.GetReport(new NMSSecurityException("bad credentials")); + var report = _sut.GetReport(new NMSSecurityException("bad credentials"), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -130,13 +152,13 @@ public void GetReport_NmsSecurityException_IsExpectedAndNotTransient() [Fact] public void GetReport_NullException_ThrowsArgumentNullException() { - Assert.Throws(() => _sut.GetReport(null!)); + Assert.Throws(() => _sut.GetReport(null!, 1)); } [Fact] public void GetReport_OperationCanceledException_IsExpectedAndNotTransient() { - var report = _sut.GetReport(new OperationCanceledException("caller cancelled")); + var report = _sut.GetReport(new OperationCanceledException("caller cancelled"), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -147,7 +169,7 @@ public void GetReport_OperationCanceledException_IsExpectedAndNotTransient() [Fact] public void GetReport_RequestTimedOutException_IsExpectedAndTransient() { - var report = _sut.GetReport(new RequestTimedOutException(TimeSpan.FromSeconds(1))); + var report = _sut.GetReport(new RequestTimedOutException(TimeSpan.FromSeconds(1)), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -160,7 +182,7 @@ public void GetReport_SingleInnerAggregateException_Unwraps() { var exception = new AggregateException(new NMSConnectionException("timeout")); - var report = _sut.GetReport(exception); + var report = _sut.GetReport(exception, 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -171,7 +193,7 @@ public void GetReport_SingleInnerAggregateException_Unwraps() [Fact] public void GetReport_SocketException_IsExpectedAndTransient() { - var report = _sut.GetReport(new SocketException((int) SocketError.TimedOut)); + var report = _sut.GetReport(new SocketException((int) SocketError.TimedOut), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -182,7 +204,7 @@ public void GetReport_SocketException_IsExpectedAndTransient() [Fact] public void GetReport_TaskCanceledException_IsExpectedAndTransient() { - var report = _sut.GetReport(new TaskCanceledException()); + var report = _sut.GetReport(new TaskCanceledException(), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -193,7 +215,7 @@ public void GetReport_TaskCanceledException_IsExpectedAndTransient() [Fact] public void GetReport_TimeoutException_IsExpectedAndTransient() { - var report = _sut.GetReport(new TimeoutException("timed out")); + var report = _sut.GetReport(new TimeoutException("timed out"), 1); Assert.False(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -204,7 +226,7 @@ public void GetReport_TimeoutException_IsExpectedAndTransient() [Fact] public void GetReport_UnrecognizedException_IsNotExpected() { - var report = _sut.GetReport(new InvalidOperationException("boom")); + var report = _sut.GetReport(new InvalidOperationException("boom"), 1); Assert.False(report.AlreadyHandled); Assert.False(report.IsExpected); @@ -222,7 +244,7 @@ public void GetReport_WorkerJobSourceException_Handled_DoesNotRetry() CouldBeExternallySolvable = true }; - var report = _sut.GetReport(exception); + var report = _sut.GetReport(exception, 1); Assert.True(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -240,7 +262,7 @@ public void GetReport_WorkerJobSourceException_UnhandledTransient_MayRetry() CouldBeExternallySolvable = true }; - var report = _sut.GetReport(exception); + var report = _sut.GetReport(exception, 1); Assert.True(report.AlreadyHandled); Assert.True(report.IsExpected); @@ -265,7 +287,7 @@ public void GetReport_WorkerSecretManagerException_IsAlreadyHandledWithFlags( CouldBeExternallySolvable = couldBeExternallySolvable }; - var report = _sut.GetReport(exception); + var report = _sut.GetReport(exception, 1); Assert.True(report.AlreadyHandled); Assert.True(report.IsExpected); diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs index 776c63b4..1ecb817e 100644 --- a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqRetryWrapperServiceTests.cs @@ -64,6 +64,58 @@ private static Mock CreateSleepService(IList? capturedD return sleep; } + [Fact] + public async Task RunAsync_NonGenericWithState_WhenFuncSucceeds_PassesState() + { + var arbiter = new Mock(MockBehavior.Strict); + var sleep = CreateSleepService(); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + string? received = null; + + await wrapper.RunAsync( + (value, _) => + { + received = value; + return Task.CompletedTask; + }, + "payload", + TestContext.Current.CancellationToken); + + Assert.Equal("payload", received); + arbiter.VerifyNoOtherCalls(); + } + + [Fact] + public async Task RunAsync_NonGenericWithState_WhenPermanentFailure_WrapsWithoutRetry() + { + var attempts = 0; + var inner = new InvalidOperationException("permanent"); + var arbiter = new Mock(MockBehavior.Strict); + arbiter.Setup(a => a.GetReport(inner, It.IsAny())).Returns(PermanentReport()); + + var sleep = CreateSleepService(); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + var thrown = await Assert.ThrowsAsync(() => wrapper.RunAsync( + (_, _) => + { + attempts++; + throw inner; + }, + "state", + TestContext.Current.CancellationToken)); + + Assert.Equal(1, attempts); + Assert.Same(inner, thrown.InnerException); + Assert.True(thrown.IsHandled); + Assert.False(thrown.CouldBeTransient); + Assert.Empty(sleep.Invocations); + } + [Fact] public async Task RunAsync_NonGeneric_WhenFuncSucceeds_CompletesWithoutSleeping() { @@ -112,7 +164,7 @@ public async Task RunAsync_NonGeneric_WhenTransientFailuresExhaustRetries_Wraps( var inner = new TimeoutException("still failing"); var arbiter = new Mock(MockBehavior.Strict); - arbiter.Setup(a => a.GetReport(It.IsAny())).Returns(TransientReport()); + arbiter.Setup(a => a.GetReport(It.IsAny(), It.IsAny())).Returns(TransientReport()); var sleep = CreateSleepService(); var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, @@ -140,7 +192,7 @@ public async Task RunAsync_WhenAlreadyHandled_RethrowsWithoutWrapping() {CouldBeTransient = false, IsHandled = true, CouldBeExternallySolvable = false}; var arbiter = new Mock(MockBehavior.Strict); - arbiter.Setup(a => a.GetReport(inner)).Returns(AlreadyHandledReport(false)); + arbiter.Setup(a => a.GetReport(inner, It.IsAny())).Returns(AlreadyHandledReport(false)); var sleep = CreateSleepService(); var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, @@ -160,7 +212,7 @@ public async Task RunAsync_WhenCriticalFailure_ThrowsRaw() var inner = new InvalidOperationException("critical"); var arbiter = new Mock(MockBehavior.Strict); - arbiter.Setup(a => a.GetReport(inner)).Returns(CriticalReport()); + arbiter.Setup(a => a.GetReport(inner, It.IsAny())).Returns(CriticalReport()); var sleep = CreateSleepService(); var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, @@ -217,7 +269,7 @@ public async Task RunAsync_WhenPermanentFailure_WrapsWithoutRetry() var inner = new ArgumentException("bad"); var arbiter = new Mock(MockBehavior.Strict); - arbiter.Setup(a => a.GetReport(inner)).Returns(PermanentReport()); + arbiter.Setup(a => a.GetReport(inner, It.IsAny())).Returns(PermanentReport()); var sleep = CreateSleepService(); var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, @@ -248,7 +300,7 @@ public async Task RunAsync_WhenTransientFailuresExhaustRetries_WrapsAsWorkerJobS var inner = new TimeoutException("still failing"); var arbiter = new Mock(MockBehavior.Strict); - arbiter.Setup(a => a.GetReport(It.IsAny())).Returns(TransientReport()); + arbiter.Setup(a => a.GetReport(It.IsAny(), It.IsAny())).Returns(TransientReport()); var sleep = CreateSleepService(delays); var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, @@ -280,7 +332,7 @@ public async Task RunAsync_WhenTransientThenSucceeds_RetriesWithBackoff() var delays = new List(); var arbiter = new Mock(MockBehavior.Strict); - arbiter.Setup(a => a.GetReport(It.IsAny())).Returns(TransientReport()); + arbiter.Setup(a => a.GetReport(It.IsAny(), It.IsAny())).Returns(TransientReport()); var sleep = CreateSleepService(delays); var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, @@ -304,4 +356,55 @@ public async Task RunAsync_WhenTransientThenSucceeds_RetriesWithBackoff() Assert.Equal(2, attempts); Assert.Equal([TimeSpan.FromSeconds(1)], delays); } + + [Fact] + public async Task RunAsync_WithState_WhenFuncSucceeds_PassesStateAndReturnsResult() + { + var arbiter = new Mock(MockBehavior.Strict); + var sleep = CreateSleepService(); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + var result = await wrapper.RunAsync( + (value, _) => Task.FromResult(value * 2), + 21, + TestContext.Current.CancellationToken); + + Assert.Equal(42, result); + arbiter.VerifyNoOtherCalls(); + Assert.Empty(sleep.Invocations); + } + + [Fact] + public async Task RunAsync_WithState_WhenTransientThenSucceeds_RetriesWithBackoff() + { + var attempts = 0; + var delays = new List(); + var arbiter = new Mock(MockBehavior.Strict); + arbiter.Setup(a => a.GetReport(It.IsAny(), It.IsAny())).Returns(TransientReport()); + + var sleep = CreateSleepService(delays); + var wrapper = new ActiveMqRetryWrapperService(arbiter.Object, + NullLogger.Instance, + sleep.Object); + + var result = await wrapper.RunAsync( + (value, _) => + { + attempts++; + if (attempts == 1) + { + throw new TimeoutException(value); + } + + return Task.FromResult(value); + }, + "ok", + TestContext.Current.CancellationToken); + + Assert.Equal("ok", result); + Assert.Equal(2, attempts); + Assert.Equal([TimeSpan.FromSeconds(1)], delays); + } } \ No newline at end of file diff --git a/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqSubscribeExceptionArbiterServiceTests.cs b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqSubscribeExceptionArbiterServiceTests.cs new file mode 100644 index 00000000..6b2e2a6e --- /dev/null +++ b/test/RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests/Tests/Services/Resilience/ActiveMqSubscribeExceptionArbiterServiceTests.cs @@ -0,0 +1,97 @@ +using Apache.NMS; +using Apache.NMS.ActiveMQ; +using RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services.Resilience; +using System.Net.Sockets; +using ActiveMqIoException = Apache.NMS.ActiveMQ.IOException; +using IOException = System.IO.IOException; + +namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.UnitTests.Tests.Services.Resilience; + +public class ActiveMqSubscribeExceptionArbiterServiceTests +{ + private readonly ActiveMqSubscribeExceptionArbiterService _sut = new(); + + public static TheoryData AccountedTransientExceptions() + { + return + [ + new BrokerException(), + new ResourceAllocationException("busy"), + new TransactionRolledBackException("rollback"), + new IllegalStateException("illegal") + ]; + } + + [Fact] + public void Classification_InspectsInnerExceptions() + { + Assert.True(_sut.IsReasonToReconnect(new Exception("outer", new SocketException()))); + Assert.True(_sut.IsReasonToStopIfHaltOnFailure(new Exception("outer", new NMSSecurityException("auth")))); + Assert.True(_sut.IsAccountedForAndLikelyTransientError(new Exception("outer", new BrokerException()))); + } + + [Theory] + [MemberData(nameof(AccountedTransientExceptions))] + public void IsAccountedForAndLikelyTransientError_KnownShapes_ReturnsTrue(Exception exception) + { + Assert.True(_sut.IsAccountedForAndLikelyTransientError(exception)); + } + + [Fact] + public void IsAccountedForAndLikelyTransientError_Unknown_ReturnsFalse() + { + Assert.False(_sut.IsAccountedForAndLikelyTransientError(new Exception("mystery"))); + } + + [Theory] + [MemberData(nameof(ReconnectExceptions))] + public void IsReasonToReconnect_KnownShapes_ReturnsTrue(Exception exception) + { + Assert.True(_sut.IsReasonToReconnect(exception)); + } + + [Fact] + public void IsReasonToReconnect_Unknown_ReturnsFalse() + { + Assert.False(_sut.IsReasonToReconnect(new Exception("mystery"))); + } + + [Theory] + [MemberData(nameof(StopIfHaltOnFailureExceptions))] + public void IsReasonToStopIfHaltOnFailure_KnownShapes_ReturnsTrue(Exception exception) + { + Assert.True(_sut.IsReasonToStopIfHaltOnFailure(exception)); + } + + [Fact] + public void IsReasonToStopIfHaltOnFailure_Unknown_ReturnsFalse() + { + Assert.False(_sut.IsReasonToStopIfHaltOnFailure(new Exception("mystery"))); + } + + public static TheoryData ReconnectExceptions() + { + return + [ + new EndOfStreamException("eof"), + new SocketException(), + new ActiveMqIoException("transport"), + new IOException("io"), + new NMSConnectionException("nms"), + new ConnectionClosedException("closed"), + new ConnectionFailedException("failed"), + new ConsumerClosedException("consumer") + ]; + } + + public static TheoryData StopIfHaltOnFailureExceptions() + { + return + [ + new NMSSecurityException("auth"), + new InvalidDestinationException("missing"), + new InvalidClientIDException("client"), + new InvalidSelectorException("selector") + ]; + } +} \ No newline at end of file diff --git a/test/local/docker-compose.yaml b/test/local/docker-compose.yaml index b40391b6..c1cdc6b6 100644 --- a/test/local/docker-compose.yaml +++ b/test/local/docker-compose.yaml @@ -285,6 +285,7 @@ services: JOB_SOURCE__ACTIVEMQ__USER_PATH: /activemq/user JOB_SOURCE__ACTIVEMQ__PASSWORD_PATH: /activemq/password JOB_SOURCE__ACTIVEMQ__QUEUE_NAME: /queue/ActiveQueue + JOB_SOURCE__ACTIVEMQ__SUBSCRIBE: ${JOB_SOURCE__ACTIVEMQ__SUBSCRIBE:-false} ## Job Management (NATS) USE_NATS: "${USE_NATS-0}" JOB_SOURCE__NATS__URL: nats://nats:4222 diff --git a/test/local/readme.md b/test/local/readme.md index be0c6892..c8bbfdaa 100644 --- a/test/local/readme.md +++ b/test/local/readme.md @@ -343,7 +343,13 @@ To initialize ActiveMQ and queue messages: unset COMMON__DISTRIBUTED__REDIS__CONNECTION_STRING_PATH ``` -7. Bring up the worker: +7. By default, ActiveMQ uses short polling. To subscribe with an async listener instead, set `JOB_SOURCE__ACTIVEMQ__SUBSCRIBE`: + + ```bash + export JOB_SOURCE__ACTIVEMQ__SUBSCRIBE=true + ``` + +8. Bring up the worker: ```bash docker compose up worker