Tweaked shutdown code for async Udp Listener

TimberWinR.TestGenerator/Program.cs

@@ -337,11 +337,11 @@ namespace TimberWinR.TestGenerator
 
         private static JObject ShutdownTimberWinR()
         {
-            _timberWinR.Shutdown();
-
             // Cancel any/all other threads
             _cancellationTokenSource.Cancel();
 
+            _timberWinR.Shutdown();
+          
             var json = Diagnostics.DiagnosticsOutput();
 
             LogManager.GetCurrentClassLogger()

TimberWinR/Inputs/UdpInputListener.cs

@@ -15,7 +15,7 @@ namespace TimberWinR.Inputs
         private IPEndPoint _udpEndpointV4;
         private readonly BlockingCollection<byte[]> _unprocessedRawData;
         private readonly Thread _rawDataProcessingThread;
-
+        public bool Stop { get; set; }
         private readonly int _port;
         private long _receivedMessages;
         private long _parseErrors;
@@ -29,7 +29,7 @@ namespace TimberWinR.Inputs
                     new JObject(new JProperty("port", _port),
                         new JProperty("receive_errors", _receiveErrors),
                         new JProperty("parse_errors", _parseErrors),
-                        new JProperty("received_messages", _receivedMessages),
+                        new JProperty("messages", _receivedMessages),
                         new JProperty("parsed_messages", _parsedMessages),
                         new JProperty("unprocessed_messages", _unprocessedRawData.Count))));
 
@@ -52,11 +52,15 @@ namespace TimberWinR.Inputs
 
         public override void Shutdown()
         {
+            Stop = true;
+
             LogManager.GetCurrentClassLogger().Info("Shutting Down {0}", InputType);
 
             // close UDP listeners, which will end the listener threads          
             _udpListenerV4.Close();
 
+            Finished();
+
             base.Shutdown();
         }
 
@@ -76,7 +80,8 @@ namespace TimberWinR.Inputs
 
         private void StartReceiving()
         {
-            _udpListenerV4.BeginReceive(DataReceived, null);
+            if (!CancelToken.IsCancellationRequested)
+                _udpListenerV4.BeginReceive(DataReceived, null);
         }
 
         private void DataReceived(IAsyncResult result)
@@ -114,13 +119,18 @@ namespace TimberWinR.Inputs
         }
 
         private void ProcessDataLoop()
-        {
-            while (!_unprocessedRawData.IsCompleted)
+        {           
+            while (!Stop && !_unprocessedRawData.IsCompleted)
             {
                 try
                 {
                     ProcessData(_unprocessedRawData.Take());
                 }
+                catch(OperationCanceledException)
+                {
+                    // we are shutting down.
+                    break;
+                }
                 catch (InvalidOperationException)
                 {
                     // when the collection is marked as completed
@@ -131,9 +141,7 @@ namespace TimberWinR.Inputs
                     LogManager.GetCurrentClassLogger().ErrorException("Error while processing data", ex);
                     Thread.Sleep(100);
                 }
-            }
-
-            Finished();
+            }              
         }
 
         private void ProcessData(byte[] bytes)