Fixed ContinuousTask test

src/Directory.Packages.props

@@ -27,6 +27,7 @@
     <PackageVersion Include="Microsoft.Azure.Functions.Worker.Core" Version="1.4.0" />
     <PackageVersion Include="Microsoft.Azure.Functions.Worker.Extensions.Abstractions" Version="1.1.0" />
     <PackageVersion Include="Microsoft.Bcl.HashCode" Version="1.1.1" />
+    <PackageVersion Include="Microsoft.Bcl.TimeProvider" Version="8.0.1" />
     <PackageVersion Include="Microsoft.CodeAnalysis.Analyzers" Version="3.3.3" />
     <PackageVersion Include="Microsoft.CodeAnalysis.CSharp" Version="4.1.0" />
     <PackageVersion Include="Microsoft.CodeAnalysis.CSharp.Workspaces" Version="4.1.0" />

src/HotChocolate/Core/src/Subscriptions.Postgres/AsyncTaskDispatcher.cs

@@ -14,7 +14,7 @@ internal sealed class AsyncTaskDispatcher : IAsyncDisposable
     public AsyncTaskDispatcher(Func<CancellationToken, Task> handler)
     {
         _handler = handler;
-        _eventProcessorTask = new ContinuousTask(EventHandler);
+        _eventProcessorTask = new ContinuousTask(EventHandler, TimeProvider.System);
     }
 
     public async Task Initialize(CancellationToken cancellationToken)

src/HotChocolate/Core/src/Subscriptions.Postgres/ContinuousTask.cs

@@ -2,16 +2,18 @@ namespace HotChocolate.Subscriptions.Postgres;
 
 internal sealed class ContinuousTask : IAsyncDisposable
 {
-    private const int _waitOnFailureInMs = 1000;
+    private readonly TimeSpan _waitOnFailure = TimeSpan.FromSeconds(1);
 
     private readonly CancellationTokenSource _completion = new();
     private readonly Func<CancellationToken, Task> _handler;
+    private readonly TimeProvider _timeProvider;
     private readonly Task _task;
     private bool _disposed;
 
-    public ContinuousTask(Func<CancellationToken, Task> handler)
+    public ContinuousTask(Func<CancellationToken, Task> handler, TimeProvider timeProvider)
     {
         _handler = handler;
+        _timeProvider = timeProvider;
 
         // We do not use Task.Factory.StartNew here because RunContinuously is an async method and
         // the LongRunning flag only works until the first await.
@@ -38,7 +40,11 @@ private async Task RunContinuously()
             {
                 if (!_completion.IsCancellationRequested)
                 {
-                    await Task.Delay(_waitOnFailureInMs, _completion.Token);
+#if NET8_0_OR_GREATER
+                    await Task.Delay(_waitOnFailure, _timeProvider, _completion.Token);
+#else
+                    await _timeProvider.Delay(_waitOnFailure, _completion.Token);
+#endif
                 }
             }
         }

src/HotChocolate/Core/src/Subscriptions.Postgres/HotChocolate.Subscriptions.Postgres.csproj

@@ -36,6 +36,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <PackageReference Include="Microsoft.Bcl.TimeProvider" Condition="'$(TargetFramework)' == 'net7.0'" />
     <PackageReference Include="Npgsql" />
   </ItemGroup>
 

src/HotChocolate/Core/src/Subscriptions.Postgres/PostgresChannel.cs

@@ -78,7 +78,9 @@ private async ValueTask OnConnect(CancellationToken cancellationToken = default)
         _subscription = new ChannelSubscription(_channelName, connection, OnNotification);
         await _subscription.ConnectAsync(cancellationToken);
 
-        _waitOnNotificationTask = new ContinuousTask(ct => connection.WaitAsync(ct));
+        _waitOnNotificationTask = new ContinuousTask(
+            ct => connection.WaitAsync(ct),
+            TimeProvider.System);
 
         _diagnosticEvents.ProviderInfo(PostgresChannel_ConnectionEstablished);
     }

src/HotChocolate/Core/src/Subscriptions.Postgres/PostgresChannelWriter.cs

@@ -46,7 +46,7 @@ private ValueTask OnConnect(CancellationToken cancellationToken = default)
             throw new InvalidOperationException("Connection was not yet initialized.");
 
         // on connection we start a task that will read from the channel and send the messages
-        _task = new ContinuousTask(ct => HandleMessage(connection, ct));
+        _task = new ContinuousTask(ct => HandleMessage(connection, ct), TimeProvider.System);
 
         _diagnosticEvents.ProviderInfo(ChannelWriter_ConnectionEstablished);
 

src/HotChocolate/Core/test/Subscriptions.Postgres.Tests/BackgroundTaskTests.cs

@@ -1,4 +1,4 @@
-using System.Diagnostics;
+using Moq;
 
 namespace HotChocolate.Subscriptions.Postgres;
 
@@ -15,7 +15,7 @@ public async Task DisposeAsync_Should_CancelAndDisposeCompletion_When_Invoked()
             return Task.CompletedTask;
         });
 
-        var backgroundTask = new ContinuousTask(handler);
+        var backgroundTask = new ContinuousTask(handler, TimeProvider.System);
 
         // Act
         await backgroundTask.DisposeAsync();
@@ -35,7 +35,7 @@ public async Task DisposeAsync_Should_ExecuteTask_When_Initialized()
             return Task.CompletedTask;
         });
 
-        var backgroundTask = new ContinuousTask(handler);
+        var backgroundTask = new ContinuousTask(handler, TimeProvider.System);
 
         // Act
         await backgroundTask.DisposeAsync();
@@ -55,7 +55,7 @@ public async Task RunContinuously_Should_ExecuteHandler_When_NotCancelled()
             return Task.CompletedTask;
         });
 
-        var backgroundTask = new ContinuousTask(handler);
+        var backgroundTask = new ContinuousTask(handler, TimeProvider.System);
 
         // Act
         SpinWait.SpinUntil(() => handlerCalled, TimeSpan.FromSeconds(1));
@@ -78,7 +78,7 @@ public async Task RunContinuously_Should_NotExecuteHandler_When_Cancelled()
             return Task.CompletedTask;
         });
 
-        var backgroundTask = new ContinuousTask(handler);
+        var backgroundTask = new ContinuousTask(handler, TimeProvider.System);
         await backgroundTask.DisposeAsync();
 
         handlerCalled = false;
@@ -95,35 +95,27 @@ public async Task RunContinuously_Should_WaitForASecond_When_HandlerThrowsExcept
     {
         // Arrange
         var hitCount = 0;
-        var stopwatch = new Stopwatch();
-        var handler = new Func<CancellationToken, Task>(async (token) =>
+        var handler = new Func<CancellationToken, Task>(_ =>
         {
             hitCount++;
 
-            if (hitCount == 1)
-            {
-                stopwatch.Start();
-                throw new Exception("First call exception");
-            }
-
-            if (hitCount == 2)
-            {
-                stopwatch.Stop();
-            }
-
-            if (hitCount > 2)
-            {
-                await Task.Delay(-1, token); // Wait indefinitely
-            }
+            throw new Exception("First call exception");
         });
 
-        var backgroundTask = new ContinuousTask(handler);
+        var mockTimeProvider = new Mock<TimeProvider>();
+        var backgroundTask = new ContinuousTask(handler, mockTimeProvider.Object);
 
         // Act
-        SpinWait.SpinUntil(() => hitCount == 2, TimeSpan.FromSeconds(5));
+        SpinWait.SpinUntil(() => hitCount == 1, TimeSpan.FromSeconds(5));
 
         // Assert
-        Assert.InRange(stopwatch.ElapsedMilliseconds, 1000, 2000);
+        mockTimeProvider.Verify(
+            tp => tp.CreateTimer(
+                It.IsAny<TimerCallback>(),
+                It.IsAny<object>(),
+                TimeSpan.FromSeconds(1),
+                Timeout.InfiniteTimeSpan),
+            Times.Once);
 
         // Cleanup
         await backgroundTask.DisposeAsync();