Adjust locks to fix sync problem

Author: macsux

src/KubernetesClient/Controllers/ResourceEventDeltaBlock.cs

@@ -7,6 +7,7 @@
 using k8s.Informers.Notifications;
 using Microsoft.Extensions.Logging;
 using Microsoft.Extensions.Logging.Abstractions;
+using Org.BouncyCastle.Asn1.X509;
 
 namespace k8s.Controllers
 {
@@ -104,7 +105,8 @@ public DataflowMessageStatus OfferMessage(DataflowMessageHeader messageHeader, R
                 _log.LogTrace($"Queuing up {resourceEvent}");
                 QueueActionLocked(resourceEvent);
 			}
-
+            if (_queue.Count == 1) // we've just added to empty queue, kick off processing
+                OfferMessagesToLinks();
 			return DataflowMessageStatus.Accepted;
 		}
 
@@ -139,8 +141,7 @@ private void QueueActionLocked(ResourceEvent<TResource> obj)
 
 			deltas.Add(obj);
 			CombineDeltas(deltas);
-			if (_queue.Count == 1) // we've just added to empty queue, kick off processing
-				OfferMessagesToLinks();
+
 
 		}
 
@@ -210,27 +211,32 @@ public List<ResourceEvent<TResource>> ConsumeMessage(DataflowMessageHeader messa
 		private void OfferMessagesToLinks()
 		{
             _log.LogTrace("Offering messages to links");
-			lock (_lock)
-			{
-				foreach (var link in _targets.ToList().Where(x => x.LastOfferedMessageReply != DataflowMessageStatus.Postponed))
+            List<TargetLink> linksToOfferTo;
+            lock (_lock)
+            {
+                linksToOfferTo = _targets.Where(x => x.LastOfferedMessageReply != DataflowMessageStatus.Postponed).ToList();
+            }
+
+            foreach (var link in linksToOfferTo)
+            {
+                DataflowMessageStatus reply;
+				do // keep feeding the link messages until queue is either empty or it tells us that it can't handle any more
 				{
-					do // keep feeding the link messages until queue is either empty or it tells us that it can't handle any more
-					{
-						OfferMessageToLink(link);
-					} while (_queue.Count > 0 && link.LastOfferedMessageReply == DataflowMessageStatus.Accepted);
-				}
+					reply = OfferMessageToLink(link);
+				} while (reply == DataflowMessageStatus.Accepted);
 			}
+
 		}
 
-		private void OfferMessageToLink(TargetLink link)
+		private DataflowMessageStatus OfferMessageToLink(TargetLink link)
 		{
 			List<ResourceEvent<TResource>> msg;
 			lock (_lock)
 			{
 				if (!_queue.TryPeek(out msg))
 				{
                     _log.LogTrace("Nothing to offer - queue is empty");
-					return; // queue is empty
+					return DataflowMessageStatus.NotAvailable; // queue is empty
 				}
 			}
 
@@ -242,6 +248,8 @@ private void OfferMessageToLink(TargetLink link)
                 _targets.Remove(link);
                 _log.LogTrace("Link is no longer accepting messages - removing link");
             }
+
+            return link.LastOfferedMessageReply;
         }
 
 		public IDisposable LinkTo(ITargetBlock<List<ResourceEvent<TResource>>> target, DataflowLinkOptions linkOptions)