feat: Dispose Requests on JobPool.Dispose()

Sally7.Tests/RequestExecutor/JobPoolTests.cs

@@ -30,4 +30,19 @@ public void ReturnJobId_Does_Not_Throw_If_Disposed()
         // Assert
         sut.ReturnJobId(jobId);
     }
+
+    [Fact]
+    public void Dispose_Calls_Dispose_On_Requests()
+    {
+        // Arrange
+        var sut = new JobPool(1);
+        var jobId = sut.RentJobIdAsync(CancellationToken.None).Result;
+        var request = sut.GetRequest(jobId);
+
+        // Act
+        sut.Dispose();
+
+        // Assert
+        Should.Throw<ObjectDisposedException>(() => request.GetResult());
+    }
 }

Sally7.Tests/RequestExecutor/RequestTests.cs

@@ -0,0 +1,35 @@
+using FakeItEasy;
+using Sally7.RequestExecutor;
+
+namespace Sally7.Tests.RequestExecutor;
+
+public class RequestTests
+{
+    [Fact]
+    public void Completes_On_Dispose()
+    {
+        // Arrange
+        var sut = new Request();
+        var callback = A.Fake<Action>();
+        sut.OnCompleted(callback);
+
+        // Act
+        sut.Dispose();
+
+        // Assert
+        A.CallTo(() => callback.Invoke()).MustHaveHappenedOnceExactly();
+    }
+
+    [Fact]
+    public async Task Throws_When_Awaited_After_Dispose()
+    {
+        // Arrange
+        var sut = new Request();
+
+        // Act
+        sut.Dispose();
+
+        // Assert
+        await Should.ThrowAsync<ObjectDisposedException>(async () => await sut);
+    }
+}

Sally7.Tests/Sally7.Tests.csproj

@@ -14,6 +14,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <PackageReference Include="FakeItEasy" Version="8.1.0" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.5.0" />
     <PackageReference Include="Shouldly" Version="4.2.1" />
     <PackageReference Include="xunit" Version="2.4.2" />

Sally7/RequestExecutor/JobPool.cs

@@ -3,6 +3,7 @@
 using System.Threading;
 using System.Threading.Channels;
 using System.Threading.Tasks;
+using Sally7.Internal;
 
 namespace Sally7.RequestExecutor;
 
@@ -32,6 +33,11 @@ public void Dispose()
     {
         Volatile.Write(ref _disposed, true);
         _jobIdPool.Writer.Complete();
+
+        foreach (var request in _requests)
+        {
+            request.Dispose();
+        }
     }
 
     public ValueTask<int> RentJobIdAsync(CancellationToken cancellationToken) => _jobIdPool.Reader.ReadAsync(cancellationToken);
@@ -45,7 +51,12 @@ public void ReturnJobId(int jobId)
     }
 
     [DebuggerNonUserCode]
-    public Request GetRequest(int jobId) => _requests[jobId - 1];
+    public Request GetRequest(int jobId)
+    {
+        DisposableHelper.ThrowIf(Volatile.Read(ref _disposed), this);
+
+        return _requests[jobId - 1];
+    }
 
     public void SetBufferForRequest(int jobId, Memory<byte> buffer)
     {

Sally7/RequestExecutor/Request.cs

@@ -1,15 +1,18 @@
 using System;
 using System.Runtime.CompilerServices;
 using System.Threading;
+using Sally7.Internal;
 
 namespace Sally7.RequestExecutor;
 
-internal class Request : INotifyCompletion
+internal class Request : INotifyCompletion, IDisposable
 {
     private static readonly Action Sentinel = () => { };
 
     private Memory<byte> _buffer;
 
+    private bool _disposed;
+
     public bool IsCompleted { get; private set; }
     private int _length;
     private Action? _continuation = Sentinel;
@@ -21,12 +24,13 @@ public void Complete(int length)
         this._length = length;
         IsCompleted = true;
 
-        var prev = _continuation ?? Interlocked.CompareExchange(ref _continuation, Sentinel, null);
-        prev?.Invoke();
+        InvokeContinuation();
     }
 
     public Memory<byte> GetResult()
     {
+        DisposableHelper.ThrowIf(_disposed, this);
+
         return _buffer.Slice(0, _length);
     }
 
@@ -44,11 +48,23 @@ public void OnCompleted(Action continuation)
     public void Reset()
     {
         _continuation = null;
-        IsCompleted = false;
+        IsCompleted = _disposed;
     }
 
     public void SetBuffer(Memory<byte> buffer)
     {
         this._buffer = buffer;
     }
+
+    public void Dispose()
+    {
+        _disposed = true;
+        InvokeContinuation();
+    }
+
+    private void InvokeContinuation()
+    {
+        var prev = _continuation ?? Interlocked.CompareExchange(ref _continuation, Sentinel, null);
+        prev?.Invoke();
+    }
 }