ActiveMQ Subscription Support

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

src/RedShirt.Example.JobWorker.Core/Services/Configuration/CoreConfigurationService.cs

@@ -9,7 +9,12 @@ namespace RedShirt.Example.JobWorker.Core.Services.Configuration;
 /// </summary>
 public interface ICoreConfigurationService
 {
+    /// <summary>
+    ///     Maximum number of jobs the worker should hold in backlog.
+    ///     Callers may assume the returned value is at least <c>1</c>.
+    /// </summary>
     int GetBacklogSize();
+
     bool IsHaltOnFailure();
 
     /// <summary>

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 <see cref="JobSubscriberManager" />.
 ///     Subscription sources push batches via <see cref="Load" />; the subscriber manager drains them via
 ///     <see cref="GetNextAsync" /> and submits each batch through job intake.
-///     This interface exists because the implementation of <see cref="IJobIntakeService" /> indirectly uses
+///     This subscriber queue should not be used in a non subscriber context. If the configured <see cref="IJobSource" />
+///     is not a subscriber, then this queue will not be read from.
+///     This interface exists because <see cref="JobIntakeService" /> indirectly uses
 ///     <see cref="IJobSource" /> as a dependency. Creating this queue was the most expedient way to avoid a circular loop.
 /// </summary>
 public interface IJobSubscriberIntakeQueue
@@ -43,7 +45,9 @@ internal class JobSubscriberIntakeQueue : IJobSubscriberIntakeQueue
     private readonly ConcurrentQueue<IJobSourceResponse> _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();

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; }
+}

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<SubscribeConfigurationModel>()?.Subscribe == true;
+
+        if (useSubscribe)
+        {
+            services.AddSingleton<IJobSource, ActiveMqSubscribeJobSource>();
+        }
+        else
+        {
+            services.AddSingleton<IJobSource, ActiveMqJobSource>();
+        }
+
         return services
             // Required
-            .AddSingleton<IJobSource, ActiveMqJobSource>()
             .AddSingleton<IJobFailureHandler, NoReactionFailureHandler>()
             // Supporting
-            .Configure<ActiveMqJobSource.ConfigurationModel>(configuration.GetSection("JobSource:ActiveMq"))
-            .Configure<ActiveMqServerConfigurationSource.ConfigurationModel>(
-                configuration.GetSection("JobSource:ActiveMq"))
+            .AddSingleton<IActiveMqSubscribeConfigurationService>(
+                new ActiveMqSubscribeConfigurationService(useSubscribe))
+            .Configure<ActiveMqConfigurationModel>(section)
+            .Configure<ActiveMqServerConfigurationSource.ConfigurationModel>(section)
             .AddSingleton<IActiveMqServerConfigurationSource, ActiveMqServerConfigurationSource>()
             .AddSingleton<IInnerActiveMqConnectionFactory, InnerActiveMqConnectionFactory>()
             .AddSingleton<IActiveMqConnectionFactory, ActiveMqConnectionFactory>()
             .AddSingleton<IActiveMqExceptionArbiterService, ActiveMqExceptionArbiterService>()
-            .AddSingleton<IActiveMqRetryWrapperService, ActiveMqRetryWrapperService>();
+            .AddSingleton<IActiveMqSubscribeExceptionArbiter, ActiveMqSubscribeExceptionArbiterService>()
+            .AddSingleton<IActiveMqRetryWrapperService, ActiveMqRetryWrapperService>()
+            .AddSingleton<IActiveMqConsumerRetryWrapper, ActiveMqConsumerRetryWrapper>();
+    }
+
+    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
     }
 }

src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Factories/ActiveMqConnectionFactory.cs

@@ -4,15 +4,21 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Factories;
 
 internal interface IActiveMqConnectionFactory
 {
-    Task<IConnection> GetConnectionAsync(CancellationToken cancellationToken = default);
+    Task<IConnection> GetConnectionAsync(
+        bool forceNewSecretManagerPull = false,
+        CancellationToken cancellationToken = default);
 }
 
 internal class ActiveMqConnectionFactory(IInnerActiveMqConnectionFactory innerActiveMqConnectionFactory)
     : IActiveMqConnectionFactory
 {
-    public async Task<IConnection> GetConnectionAsync(CancellationToken cancellationToken = default)
+    public async Task<IConnection> 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);
     }

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<IActiveConnectionWrapper> GetConnectionFactoryWrapperAsync(
+        bool forceNewSecretManagerPull = false,
         CancellationToken cancellationToken = default);
 }
 
 internal class InnerActiveMqConnectionFactory(
-    IActiveMqServerConfigurationSource configurationSource)
+    IActiveMqServerConfigurationSource configurationSource,
+    IActiveMqSubscribeConfigurationService activeMqSubscribeConfigurationService,
+    ICoreConfigurationService coreConfigurationService)
     : IInnerActiveMqConnectionFactory
 {
     public async Task<IActiveConnectionWrapper> 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);
     }
+
+    /// <summary>
+    ///     Removes an NMS <c>failover:</c> wrapper from <paramref name="brokerUri" />, leaving the nested
+    ///     broker address (first composite URI when several are listed). Plain URIs are returned unchanged.
+    /// </summary>
+    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];
+    }
 }

src/RedShirt.Example.JobWorker.JobManagement.ActiveMq/Services/ActiveMqConfigurationSource.cs

@@ -6,18 +6,22 @@ namespace RedShirt.Example.JobWorker.JobManagement.ActiveMq.Services;
 
 internal interface IActiveMqServerConfigurationSource
 {
-    Task<ActiveMqServerConfigurationModel> GetConfigurationAsync(CancellationToken cancellationToken = default);
+    Task<ActiveMqServerConfigurationModel> GetConfigurationAsync(
+        bool forceNewSecretManagerPull = false,
+        CancellationToken cancellationToken = default);
 }
 
 internal class ActiveMqServerConfigurationSource(
     ISecretManagerCacheService secretManagerCacheService,
     IOptions<ActiveMqServerConfigurationSource.ConfigurationModel> options) : IActiveMqServerConfigurationSource
 {
     public async Task<ActiveMqServerConfigurationModel> 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

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<IMessageConsumer, CancellationToken, Task> callback,
+        Action<IConnection>? onNewConnectionCallback = null,
+        Action<IMessageConsumer>? onNewMessageConsumerCallback = null,
+        CancellationToken cancellationToken = default);
+
+    void ResetConsumer();
+}
+
+internal class ActiveMqConsumerRetryWrapper(
+    IActiveMqConnectionFactory connectionFactory,
+    IActiveMqRetryWrapperService retryWrapperService,
+    IActiveMqExceptionArbiterService exceptionArbiterService,
+    IOptions<ActiveMqConfigurationModel> configuration) : IActiveMqConsumerRetryWrapper
+{
+    private IMessageConsumer? _messageConsumer;
+
+    private async Task CallbackAsync(Func<IMessageConsumer, CancellationToken, Task> callback,
+        RetryState state,
+        Action<IConnection>? onNewConnectionCallback,
+        Action<IMessageConsumer>? 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;
+        }
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    /// <param name="onNewConnectionCallback"></param>
+    /// <param name="onNewMessageConsumerCallback"></param>
+    /// <param name="forceNewSecretManagerPull"></param>
+    /// <param name="cancellationToken"></param>
+    /// <returns></returns>
+    /// <exception cref="CouldNotLoadQueueException"></exception>
+    private async Task<IMessageConsumer> GetConsumerAsync(Action<IConnection>? onNewConnectionCallback,
+        Action<IMessageConsumer>? 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<IMessageConsumer, CancellationToken, Task> callback,
+        Action<IConnection>? onNewConnectionCallback = null,
+        Action<IMessageConsumer>? 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; }
+    }
+}