Merge pull request #304 from dmitry-vychikov/feature/first-class-queue

First class queue support

Author: azygis

src/Hangfire.PostgreSql/Hangfire.PostgreSql.csproj

@@ -35,7 +35,7 @@
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>
-    <PackageReference Include="Hangfire.Core" Version="1.7.28" />
+    <PackageReference Include="Hangfire.Core" Version="1.8.0" />
     <PackageReference Include="Microsoft.CSharp" Version="4.7.0" />
     <PackageReference Include="Npgsql" Version="6.0.0" />
   </ItemGroup>

src/Hangfire.PostgreSql/PostgreSqlStorage.cs

@@ -41,6 +41,13 @@ public class PostgreSqlStorage : JobStorage
     private readonly Action<NpgsqlConnection> _connectionSetup;
     private readonly NpgsqlConnectionStringBuilder _connectionStringBuilder;
     private readonly NpgsqlConnection _existingConnection;
+    
+    private readonly Dictionary<string, bool> _features =
+      new(StringComparer.OrdinalIgnoreCase)
+      {
+        { JobStorageFeatures.JobQueueProperty, true },
+      };
+
 
     public PostgreSqlStorage(string connectionString)
       : this(connectionString, new PostgreSqlStorageOptions()) { }
@@ -420,5 +427,14 @@ private NpgsqlConnectionStringBuilder SetupConnectionStringBuilderParameters(Npg
 
       return builder;
     }
+
+    public override bool HasFeature(string featureId)
+    {
+      if (featureId == null) throw new ArgumentNullException(nameof(featureId));
+
+      return _features.TryGetValue(featureId, out bool isSupported) 
+        ? isSupported
+        : base.HasFeature(featureId);
+    }
   }
 }

tests/Hangfire.PostgreSql.Tests/Entities/TestJob.cs

@@ -3,4 +3,12 @@
 namespace Hangfire.PostgreSql.Tests
 {
   public record TestJob(long Id, string InvocationData, string Arguments, DateTime? ExpireAt, string StateName, long? StateId, DateTime CreatedAt);
+
+  public class TestJobs
+  {
+    public void Run(string logMessage)
+    {
+      Console.WriteLine("Running test job: {0}", logMessage);
+    }
+  }
 }

tests/Hangfire.PostgreSql.Tests/FirstClassQueueFeatureSupportTests.cs

@@ -0,0 +1,44 @@
+using System.Threading;
+using Hangfire.PostgreSql.Tests.Utils;
+using Hangfire.Storage;
+using Hangfire.Storage.Monitoring;
+using Xunit;
+
+namespace Hangfire.PostgreSql.Tests;
+
+public class FirstClassQueueFeatureSupportTests
+{
+  public FirstClassQueueFeatureSupportTests()
+  {
+    JobStorage.Current = new PostgreSqlStorage(ConnectionUtils.GetConnectionString());
+  }
+
+  [Fact]
+  public void HasFlag_ShouldReturnTrue_ForJobQueueProperty()
+  {
+    bool supportJobQueueProperty = JobStorage.Current.HasFeature(JobStorageFeatures.JobQueueProperty);
+    Assert.True(supportJobQueueProperty);
+  }
+
+  [Fact]
+  [CleanDatabase]
+  public void EnqueueJobWithSpecificQueue_ShouldEnqueueCorrectlyAndJobMustBeProcessedInThatQueue()
+  {
+    BackgroundJob.Enqueue<TestJobs>("critical", job => job.Run("critical"));
+    BackgroundJob.Enqueue<TestJobs>("offline", job => job.Run("offline"));
+
+    BackgroundJobServer server = new(new BackgroundJobServerOptions() {
+      Queues = new[] { "critical" },
+    });
+
+    Thread.Sleep(200);
+
+    IMonitoringApi monitoringApi = JobStorage.Current.GetMonitoringApi();
+
+    JobList<EnqueuedJobDto> jobsInCriticalQueue = monitoringApi.EnqueuedJobs("critical", 0, 10);
+    JobList<EnqueuedJobDto> jobsInOfflineQueue = monitoringApi.EnqueuedJobs("offline", 0, 10);
+
+    Assert.Empty(jobsInCriticalQueue);   //Job from 'critical' queue must be processed by the server 
+    Assert.NotEmpty(jobsInOfflineQueue); //Job from 'offline' queue must be left untouched because no server is processing it
+  }
+}