- refactored available worker to specify thread id

Shuttle · Jun 30, 2015 · 52395db · 52395db
1 parent 34cf27e
commit 52395db
Show file tree

Hide file tree

Showing 7 changed files with 19 additions and 2 deletions.
diff --git a/Shuttle.ESB.Core/Processing/Distributor/AvailableWorker.cs b/Shuttle.ESB.Core/Processing/Distributor/AvailableWorker.cs
@@ -6,12 +6,14 @@ public class AvailableWorker
     {
         public Guid Identifier { get; private set; }
         public string InboxWorkQueueUri { get; private set; }
+        public int ManagedThreadId { get; private set; }
         public DateTime WorkerSendDate { get; private set; }
 
         public AvailableWorker(WorkerThreadAvailableCommand command)
         {
             Identifier = command.Identifier;
             InboxWorkQueueUri = command.InboxWorkQueueUri;
+            ManagedThreadId = command.ManagedThreadId;
             WorkerSendDate = command.DateSent;
         }
     }

diff --git a/Shuttle.ESB.Core/Processing/Distributor/IWorkerAvailabilityManager.cs b/Shuttle.ESB.Core/Processing/Distributor/IWorkerAvailabilityManager.cs
@@ -7,5 +7,6 @@ public interface IWorkerAvailabilityManager
 
         void WorkerAvailable(WorkerThreadAvailableCommand message);
         void WorkerStarted(WorkerStartedEvent message);
+	    void RemoveByThread(WorkerThreadAvailableCommand message);
     }
 }
diff --git a/Shuttle.ESB.Core/Processing/Distributor/WorkerAvailabilityManager.cs b/Shuttle.ESB.Core/Processing/Distributor/WorkerAvailabilityManager.cs
@@ -101,5 +101,14 @@ public void WorkerStarted(WorkerStartedEvent message)
 					.ToList();
 			}
 		}
+
+		public void RemoveByThread(WorkerThreadAvailableCommand message)
+		{
+			lock (_padlock)
+			{
+				GetAvailableWorkers(message.InboxWorkQueueUri)
+					.RemoveAll(candidate => candidate.ManagedThreadId == message.ManagedThreadId);
+			}
+		}
 	}
 }
diff --git a/Shuttle.ESB.Core/Processing/Distributor/WorkerAvailableHandler.cs b/Shuttle.ESB.Core/Processing/Distributor/WorkerAvailableHandler.cs
@@ -8,6 +8,8 @@ public void ProcessMessage(HandlerContext<WorkerThreadAvailableCommand> context)
 									   ? context.Configuration.Inbox.DistributeSendCount
 									   : 5;
 
+			context.Configuration.WorkerAvailabilityManager.RemoveByThread(context.Message);
+
 			for (var i = 0; i < distributeSendCount; i++)
 			{
 				context.Configuration.WorkerAvailabilityManager.WorkerAvailable(context.Message);

diff --git a/Shuttle.ESB.Core/Processing/Distributor/WorkerThreadActivity.cs b/Shuttle.ESB.Core/Processing/Distributor/WorkerThreadActivity.cs
@@ -1,4 +1,5 @@
 using System;
+using System.Threading;
 using Shuttle.Core.Infrastructure;
 
 namespace Shuttle.ESB.Core
@@ -32,6 +33,7 @@ public void Waiting(IThreadState state)
 					{
 						Identifier = identifier,
 						InboxWorkQueueUri = bus.Configuration.Inbox.WorkQueue.Uri.ToString(),
+						ManagedThreadId = Thread.CurrentThread.ManagedThreadId,
 						DateSent = DateTime.Now
 					},
 				         c => c.WithRecipient(bus.Configuration.Worker.DistributorControlInboxWorkQueue));

diff --git a/Shuttle.ESB.Core/Processing/Distributor/WorkerThreadAvailableCommand.cs b/Shuttle.ESB.Core/Processing/Distributor/WorkerThreadAvailableCommand.cs
@@ -6,6 +6,7 @@ public class WorkerThreadAvailableCommand
     {
         public Guid Identifier { get; set; }
         public string InboxWorkQueueUri { get; set; }
+	    public int ManagedThreadId { get; set; }
         public DateTime DateSent { get; set; }
     }
 }
diff --git a/Shuttle.ESB.Core/Properties/AssemblyInfo.cs b/Shuttle.ESB.Core/Properties/AssemblyInfo.cs
@@ -18,11 +18,11 @@
 [assembly: AssemblyTitle("Shuttle.ESB.Core for .NET Framework 4.5.1")]
 #endif
 
-[assembly: AssemblyVersion("3.6.2.0")]
+[assembly: AssemblyVersion("3.6.3.0")]
 [assembly: AssemblyCopyright("Copyright © Eben Roux 2010-2015")]
 [assembly: AssemblyProduct("Shuttle.ESB")]
 [assembly: AssemblyCompany("Shuttle")]
 [assembly: AssemblyConfiguration("Release")]
-[assembly: AssemblyInformationalVersion("3.6.2")]
+[assembly: AssemblyInformationalVersion("3.6.3")]
 [assembly: ComVisible(false)]