- replaced ThreadState with CancellationToken

Author: eben-roux

Shuttle.Core.Threading.Tests/ThreadActivityFixture.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using Moq;
 using NUnit.Framework;
 
@@ -17,16 +18,13 @@ public void Should_be_able_to_have_the_thread_wait()
             });
 
             var start = DateTime.Now;
+            var token = new CancellationToken(false);
 
-            var mockState = new Mock<IThreadState>();
-
-            mockState.Setup(mock => mock.Active).Returns(true);
-
-            activity.Waiting(mockState.Object);
+            activity.Waiting(token);
 
             Assert.IsTrue((DateTime.Now - start).TotalMilliseconds >= 250);
 
-            activity.Waiting(mockState.Object);
+            activity.Waiting(token);
 
             Assert.IsTrue((DateTime.Now - start).TotalMilliseconds >= 750);
         }

Shuttle.Core.Threading/IProcessor.cs

@@ -1,7 +1,9 @@
+using System.Threading;
+
 namespace Shuttle.Core.Threading
 {
     public interface IProcessor
     {
-        void Execute(IThreadState state);
+        void Execute(CancellationToken cancellationToken);
     }
 }

Shuttle.Core.Threading/IThreadActivity.cs

@@ -1,8 +1,10 @@
+using System.Threading;
+
 namespace Shuttle.Core.Threading
 {
     public interface IThreadActivity
     {
-        void Waiting(IThreadState state);
+        void Waiting(CancellationToken cancellationToken);
         void Working();
     }
 }

Shuttle.Core.Threading/IThreadState.cs

@@ -7,16 +7,17 @@
 
 namespace Shuttle.Core.Threading
 {
-    public class ProcessorThread : IThreadState
+    public class ProcessorThread
     {
         private static readonly int ThreadJoinTimeoutInterval =
             ConfigurationItem<int>.ReadSetting("ThreadJoinTimeoutInterval", 1000).GetValue();
 
         private readonly ILog _log;
         private readonly string _name;
         private readonly IProcessor _processor;
+        private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();
 
-        private volatile bool _active;
+        private bool _started;
 
         private Thread _thread;
 
@@ -27,14 +28,16 @@ public ProcessorThread(string name, IProcessor processor)
             _name = name;
             _processor = processor;
 
+            CancellationToken = _cancellationTokenSource.Token;
+
             _log = Log.For(this);
         }
 
-        public bool Active => _active;
+        public CancellationToken CancellationToken {get; }
 
         public void Start()
         {
-            if (_active)
+            if (_started)
             {
                 return;
             }
@@ -57,8 +60,6 @@ public void Start()
             _thread.IsBackground = true;
             _thread.Priority = ThreadPriority.Normal;
 
-            _active = true;
-
             _thread.Start();
 
             if (Log.IsTraceEnabled)
@@ -67,15 +68,17 @@ public void Start()
                     _processor.GetType().FullName));
             }
 
-            while (!_thread.IsAlive && _active)
+            while (!_thread.IsAlive && !CancellationToken.IsCancellationRequested)
             {
             }
 
-            if (_active && Log.IsTraceEnabled)
+            if (!CancellationToken.IsCancellationRequested && Log.IsTraceEnabled)
             {
                 _log.Trace(string.Format(Resources.ProcessorThreadActive, _thread.ManagedThreadId,
                     _processor.GetType().FullName));
             }
+
+            _started = true;
         }
 
         public void Stop()
@@ -86,7 +89,7 @@ public void Stop()
                     _processor.GetType().FullName));
             }
 
-            _active = false;
+            _cancellationTokenSource.Cancel();
 
             _processor.AttemptDispose();
 
@@ -98,15 +101,15 @@ public void Stop()
 
         private void Work()
         {
-            while (_active)
+            while (!CancellationToken.IsCancellationRequested)
             {
                 if (Log.IsVerboseEnabled)
                 {
                     _log.Verbose(string.Format(Resources.ProcessorExecuting, _thread.ManagedThreadId,
                         _processor.GetType().FullName));
                 }
 
-                _processor.Execute(this);
+                _processor.Execute(CancellationToken);
             }
 
             if (Log.IsTraceEnabled)
@@ -118,7 +121,7 @@ private void Work()
 
         internal void Deactivate()
         {
-            _active = false;
+            _cancellationTokenSource.Cancel();
         }
     }
 }

Shuttle.Core.Threading/ProcessorThread.cs

@@ -33,10 +33,10 @@
 [assembly: AssemblyTitle(".NET Standard 2.0")]
 #endif
 
-[assembly: AssemblyVersion("10.1.0.0")]
-[assembly: AssemblyCopyright("Copyright © Eben Roux 2018")]
+[assembly: AssemblyVersion("11.0.0.0")]
+[assembly: AssemblyCopyright("Copyright © Eben Roux 2019")]
 [assembly: AssemblyProduct("Shuttle.Core.Threading")]
 [assembly: AssemblyCompany("Shuttle")]
 [assembly: AssemblyConfiguration("Release")]
-[assembly: AssemblyInformationalVersion("10.1.0")]
+[assembly: AssemblyInformationalVersion("11.0.0")]
 [assembly: ComVisible(false)]

Shuttle.Core.Threading/Properties/AssemblyInfo.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using Shuttle.Core.Contract;
 
 namespace Shuttle.Core.Threading
@@ -26,11 +27,11 @@ public ThreadActivity(IThreadActivityConfiguration threadActivityConfiguration)
             _durationIndex = 0;
         }
 
-        public void Waiting(IThreadState state)
+        public void Waiting(CancellationToken cancellationToken)
         {
             var ms = (int) GetSleepTimeSpan().TotalMilliseconds;
 
-            ThreadSleep.While(ms, state);
+            ThreadSleep.While(ms, cancellationToken);
         }
 
         public void Working()

Shuttle.Core.Threading/ThreadActivity.cs

@@ -1,21 +1,24 @@
 using System;
 using System.Threading;
+using Shuttle.Core.Contract;
 
 namespace Shuttle.Core.Threading
 {
     public static class ThreadSleep
     {
         public const int MaxStepSize = 1000;
 
-        public static void While(int ms, IThreadState state)
+        public static void While(int ms, CancellationToken cancellationToken)
         {
             // step size should be as large as possible,
             // max step size by default
-            While(ms, state, MaxStepSize);
+            While(ms, cancellationToken, MaxStepSize);
         }
 
-        public static void While(int ms, IThreadState state, int step)
+        public static void While(int ms, CancellationToken cancellationToken, int step)
         {
+            Guard.AgainstNull(cancellationToken, nameof(cancellationToken));
+
             if (ms < 0)
             {
                 return;
@@ -34,7 +37,7 @@ public static void While(int ms, IThreadState state, int step)
 
             var remaining = (int) (end - DateTime.UtcNow).TotalMilliseconds;
 
-            while (state.Active && remaining > 0)
+            while (!cancellationToken.IsCancellationRequested && remaining > 0)
             {
                 var sleep = remaining < step ? remaining : step;
                 Thread.Sleep(sleep);