Added IServiceScopeFactory to threading.

eben-roux · eben-roux · commit 71b14dcb98ec · 2024-11-13T16:12:47.000+02:00
diff --git a/Shuttle.Core.Threading.Tests/MockProcessor.cs b/Shuttle.Core.Threading.Tests/MockProcessor.cs
@@ -15,7 +15,7 @@ public MockProcessor(TimeSpan executionDuration)
 
     public int ExecutionCount { get; private set; }
 
-    public async Task ExecuteAsync(IProcessorThreadContext processorThreadContext, CancellationToken cancellationToken)
+    public async Task ExecuteAsync(IProcessorThreadContext context, CancellationToken cancellationToken)
     {
         await Task.Delay(_executionDuration, cancellationToken).ConfigureAwait(false);
         ExecutionCount++;
diff --git a/Shuttle.Core.Threading.Tests/ProcessorThreadFixture.cs b/Shuttle.Core.Threading.Tests/ProcessorThreadFixture.cs
@@ -1,6 +1,8 @@
 ﻿using System;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using Moq;
 using NUnit.Framework;
 
 namespace Shuttle.Core.Threading.Tests;
@@ -12,9 +14,13 @@ public async Task Should_be_able_to_execute_processor_thread_async()
     {
         const int minimumExecutionCount = 5;
 
+        var serviceScopeFactory = new Mock<IServiceScopeFactory>();
+
+        serviceScopeFactory.Setup(m => m.CreateScope()).Returns(new Mock<IServiceScope>().Object);
+
         var executionDuration = TimeSpan.FromMilliseconds(200);
         var mockProcessor = new MockProcessor(executionDuration);
-        var processorThread = new ProcessorThread("thread", mockProcessor, new());
+        var processorThread = new ProcessorThread("thread", serviceScopeFactory.Object, mockProcessor, new());
         var cancellationTokenSource = new CancellationTokenSource();
         var cancellationToken = cancellationTokenSource.Token;
 
diff --git a/Shuttle.Core.Threading.Tests/ProcessorThreadPoolFixture.cs b/Shuttle.Core.Threading.Tests/ProcessorThreadPoolFixture.cs
@@ -2,6 +2,7 @@
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
 using Moq;
 using NUnit.Framework;
 
@@ -14,14 +15,18 @@ public async Task Should_be_able_to_execute_processor_thread_pool_async()
     {
         const int minimumExecutionCount = 5;
 
+        var serviceScopeFactory = new Mock<IServiceScopeFactory>();
+
+        serviceScopeFactory.Setup(m => m.CreateScope()).Returns(new Mock<IServiceScope>().Object);
+
         var executionDuration = TimeSpan.FromMilliseconds(500);
         var cancellationTokenSource = new CancellationTokenSource();
         var cancellationToken = cancellationTokenSource.Token;
         var processorFactory = new Mock<IProcessorFactory>();
 
         processorFactory.Setup(m => m.Create()).Returns(() => new MockProcessor(executionDuration));
 
-        var processorThreadPool = new ProcessorThreadPool("thread-pool", 5, processorFactory.Object, new());
+        var processorThreadPool = new ProcessorThreadPool("thread-pool", 5, serviceScopeFactory.Object, processorFactory.Object, new());
 
         processorThreadPool.ProcessorThreadCreated += (_, args) =>
         {
diff --git a/Shuttle.Core.Threading/IProcessor.cs b/Shuttle.Core.Threading/IProcessor.cs
@@ -5,5 +5,5 @@ namespace Shuttle.Core.Threading;
 
 public interface IProcessor
 {
-    Task ExecuteAsync(IProcessorThreadContext processorThreadContext, CancellationToken cancellationToken = default);
+    Task ExecuteAsync(IProcessorThreadContext processorThread, CancellationToken cancellationToken = default);
 }
diff --git a/Shuttle.Core.Threading/IProcessorThreadContext.cs b/Shuttle.Core.Threading/IProcessorThreadContext.cs
@@ -1,6 +1,9 @@
+using Microsoft.Extensions.DependencyInjection;
+
 namespace Shuttle.Core.Threading;
 
 public interface IProcessorThreadContext
 {
-    object? GetState(string key);
+    IState State { get; }
+    IServiceScope ServiceScope { get; }
 }
diff --git a/Shuttle.Core.Threading/IState.cs b/Shuttle.Core.Threading/IState.cs
@@ -0,0 +1,11 @@
+namespace Shuttle.Core.Threading;
+
+public interface IState
+{
+    void Add(string key, object? value);
+    void Clear();
+    bool Contains(string key);
+    object? Get(string key);
+    bool Remove(string key);
+    void Replace(string key, object? value);
+}
diff --git a/Shuttle.Core.Threading/ProcessorThread.cs b/Shuttle.Core.Threading/ProcessorThread.cs
@@ -2,13 +2,15 @@
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
 using Shuttle.Core.Contract;
 using Shuttle.Core.Reflection;
 
 namespace Shuttle.Core.Threading;
 
-public class ProcessorThread : IProcessorThreadContext
+public class ProcessorThread
 {
+    private readonly IServiceScopeFactory _serviceScopeFactory;
     private readonly CancellationTokenSource _cancellationTokenSource = new();
     private readonly ProcessorThreadOptions _processorThreadOptions;
 
@@ -18,8 +20,9 @@ public class ProcessorThread : IProcessorThreadContext
     private bool _started;
     private readonly Thread _thread;
 
-    public ProcessorThread(string name, IProcessor processor, ProcessorThreadOptions processorThreadOptions)
+    public ProcessorThread(string name, IServiceScopeFactory serviceScopeFactory, IProcessor processor, ProcessorThreadOptions processorThreadOptions)
     {
+        _serviceScopeFactory = Guard.AgainstNull(serviceScopeFactory);
         Name = Guard.AgainstNull(name);
         Processor = Guard.AgainstNull(processor);
         _processorThreadOptions = Guard.AgainstNull(processorThreadOptions);
@@ -45,12 +48,7 @@ internal void Deactivate()
         _cancellationTokenSource.Cancel();
     }
 
-    public object? GetState(string key)
-    {
-        Guard.AgainstNullOrEmptyString(key);
-
-        return _state.TryGetValue(key, out var value) ? value : null;
-    }
+    public IState State { get; } = new State();
 
     public event EventHandler<ProcessorThreadExceptionEventArgs>? ProcessorException;
     public event EventHandler<ProcessorThreadEventArgs>? ProcessorExecuting;
@@ -60,11 +58,6 @@ internal void Deactivate()
     public event EventHandler<ProcessorThreadEventArgs>? ProcessorThreadStopped;
     public event EventHandler<ProcessorThreadEventArgs>? ProcessorThreadStopping;
 
-    public void SetState(string key, object value)
-    {
-        _state[Guard.AgainstNullOrEmptyString(key)] = value;
-    }
-
     public async Task StartAsync()
     {
         if (_started)
@@ -126,7 +119,10 @@ private async void Work()
 
             try
             {
-                await Processor.ExecuteAsync(this, CancellationToken);
+                using (var context = new ProcessorThreadContext(State, _serviceScopeFactory.CreateScope()))
+                {
+                    await Processor.ExecuteAsync(context, CancellationToken);
+                }
             }
             catch (OperationCanceledException)
             {
diff --git a/Shuttle.Core.Threading/ProcessorThreadPool.cs b/Shuttle.Core.Threading/ProcessorThreadPool.cs
@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
 using Shuttle.Core.Contract;
 using Shuttle.Core.Reflection;
 
@@ -9,17 +10,20 @@ namespace Shuttle.Core.Threading;
 public class ProcessorThreadPool : IProcessorThreadPool
 {
     private readonly List<ProcessorThread> _processorThreads = new();
+    private readonly IServiceScopeFactory _serviceScopeFactory;
     private bool _disposed;
     private bool _started;
 
-    public ProcessorThreadPool(string name, int threadCount, IProcessorFactory processorFactory, ProcessorThreadOptions processorThreadOptions)
+    public ProcessorThreadPool(string name, int threadCount, IServiceScopeFactory serviceScopeFactory, IProcessorFactory processorFactory, ProcessorThreadOptions processorThreadOptions)
     {
         if (threadCount < 1)
         {
             throw new ThreadCountZeroException();
         }
 
+
         Name = name;
+        _serviceScopeFactory = Guard.AgainstNull(serviceScopeFactory);
         ProcessorFactory = Guard.AgainstNull(processorFactory);
         ThreadOptions = Guard.AgainstNull(processorThreadOptions);
         ThreadCount = threadCount;
@@ -60,7 +64,7 @@ public async Task StartAsync()
 
         while (i++ < ThreadCount)
         {
-            var processorThread = new ProcessorThread($"{Name} / {i}", ProcessorFactory.Create(), ThreadOptions);
+            var processorThread = new ProcessorThread($"{Name} / {i}", _serviceScopeFactory, ProcessorFactory.Create(), ThreadOptions);
 
             ProcessorThreadCreated?.Invoke(this, new(processorThread));
 
diff --git a/Shuttle.Core.Threading/ProcessorThreadPoolFactory.cs b/Shuttle.Core.Threading/ProcessorThreadPoolFactory.cs
@@ -1,14 +1,23 @@
 ﻿using System;
+using Microsoft.Extensions.DependencyInjection;
+using Shuttle.Core.Contract;
 
 namespace Shuttle.Core.Threading;
 
 public class ProcessorThreadPoolFactory : IProcessorThreadPoolFactory
 {
+    private readonly IServiceScopeFactory _serviceScopeFactory;
+
+    public ProcessorThreadPoolFactory(IServiceScopeFactory serviceScopeFactory)
+    {
+        _serviceScopeFactory = Guard.AgainstNull(serviceScopeFactory);
+    }
+
     public event EventHandler<ProcessorThreadPoolCreatedEventArgs>? ProcessorThreadPoolCreated;
 
     public IProcessorThreadPool Create(string name, int threadCount, IProcessorFactory processorFactory, ProcessorThreadOptions processorThreadOptions)
     {
-        var result = new ProcessorThreadPool(name, threadCount, processorFactory, processorThreadOptions);
+        var result = new ProcessorThreadPool(name, threadCount, _serviceScopeFactory, processorFactory, processorThreadOptions);
 
         ProcessorThreadPoolCreated?.Invoke(this, new(result));
 
diff --git a/Shuttle.Core.Threading/Shuttle.Core.Threading.csproj b/Shuttle.Core.Threading/Shuttle.Core.Threading.csproj
@@ -16,6 +16,7 @@
 
   <ItemGroup>
     <PackageReference Include="Microsoft.Extensions.Options" Version="8.0.2" />
+    <PackageReference Include="Moq" Version="4.20.72" />
     <PackageReference Include="Shuttle.Core.Contract" Version="20.0.0" />
     <PackageReference Include="Shuttle.Core.Reflection" Version="20.0.0" />
   </ItemGroup>
diff --git a/Shuttle.Core.Threading/State.cs b/Shuttle.Core.Threading/State.cs
@@ -0,0 +1,42 @@
+using System.Collections.Generic;
+using Shuttle.Core.Contract;
+
+namespace Shuttle.Core.Threading;
+
+public class State : IState
+{
+    private readonly Dictionary<string, object?> _state = new();
+
+    public void Clear()
+    {
+        _state.Clear();
+    }
+
+    public void Add(string key, object? value)
+    {
+        _state.Add(Guard.AgainstNull(key), value);
+    }
+
+    public void Replace(string key, object? value)
+    {
+        Guard.AgainstNull(key);
+
+        _state.Remove(key);
+        _state.Add(key, value);
+    }
+
+    public object? Get(string key)
+    {
+        return _state.TryGetValue(Guard.AgainstNull(key), out var result) ? result : default;
+    }
+
+    public bool Contains(string key)
+    {
+        return _state.ContainsKey(Guard.AgainstNull(key));
+    }
+
+    public bool Remove(string key)
+    {
+        return _state.Remove(key);
+    }
+}
diff --git a/Shuttle.Core.Threading/ThreadContext.cs b/Shuttle.Core.Threading/ThreadContext.cs
@@ -0,0 +1,22 @@
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using Shuttle.Core.Contract;
+
+namespace Shuttle.Core.Threading;
+
+public class ProcessorThreadContext : IDisposable, IProcessorThreadContext
+{
+    public IState State { get; }
+    public IServiceScope ServiceScope { get; }
+
+    public ProcessorThreadContext(IState state, IServiceScope serviceScope)
+    {
+        State = Guard.AgainstNull(state);
+        ServiceScope = Guard.AgainstNull(serviceScope);
+    }
+
+    public void Dispose()
+    {
+        ServiceScope.Dispose();
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -15,7 +15,7 @@ public MockProcessor(TimeSpan executionDuration)`
`15`	`15`
`16`	`16`	`public int ExecutionCount { get; private set; }`
`17`	`17`
`18`		`- public async Task ExecuteAsync(IProcessorThreadContext processorThreadContext, CancellationToken cancellationToken)`
	`18`	`+ public async Task ExecuteAsync(IProcessorThreadContext context, CancellationToken cancellationToken)`
`19`	`19`	`{`
`20`	`20`	`await Task.Delay(_executionDuration, cancellationToken).ConfigureAwait(false);`
`21`	`21`	`ExecutionCount++;`
Original file line number	Diff line number	Diff line change
`@@ -5,5 +5,5 @@ namespace Shuttle.Core.Threading;`
`5`	`5`
`6`	`6`	`public interface IProcessor`
`7`	`7`	`{`
`8`		`- Task ExecuteAsync(IProcessorThreadContext processorThreadContext, CancellationToken cancellationToken = default);`
	`8`	`+ Task ExecuteAsync(IProcessorThreadContext processorThread, CancellationToken cancellationToken = default);`
`9`	`9`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,9 @@`
	`1`	`+using Microsoft.Extensions.DependencyInjection;`
	`2`	`+`
`1`	`3`	`namespace Shuttle.Core.Threading;`
`2`	`4`
`3`	`5`	`public interface IProcessorThreadContext`
`4`	`6`	`{`
`5`		`- object? GetState(string key);`
	`7`	`+ IState State { get; }`
	`8`	`+ IServiceScope ServiceScope { get; }`
`6`	`9`	`}`