Big changes

Source/TSRealLib/Common/Raytheon.Common/ThreadWorkers/MsgProcessorWorker.cs

@@ -0,0 +1,247 @@
+//                           UNCLASSIFIED
+/*-------------------------------------------------------------------------
+RAYTHEON PROPRIETARY: THIS DOCUMENT CONTAINS DATA OR INFORMATION
+PROPRIETARY TO RAYTHEON COMPANY AND IS RESTRICTED TO USE ONLY BY PERSONS
+AUTHORIZED BY RAYTHEON COMPANY IN WRITING TO USE IT. DISCLOSURE TO
+UNAUTHORIZED PERSONS WOULD LIKELY CAUSE SUBSTANTIAL COMPETITIVE HARM TO
+RAYTHEON COMPANY'S BUSINESS POSITION. NEITHER SAID DOCUMENT NOR ITS
+CONTENTS SHALL BE FURNISHED OR DISCLOSED TO OR COPIED OR USED BY PERSONS
+OUTSIDE RAYTHEON COMPANY WITHOUT THE EXPRESS WRITTEN APPROVAL OF RAYTHEON
+COMPANY.
+
+THIS PROPRIETARY NOTICE IS NOT APPLICABLE IF DELIVERED TO THE U.S.
+GOVERNMENT.
+
+UNPUBLISHED WORK - COPYRIGHT RAYTHEON COMPANY.
+-------------------------------------------------------------------------*/
+
+using System;
+using System.Threading;
+
+namespace Raytheon.Common
+{
+   /// <summary>
+   /// Processes messages from a buffer
+   /// </summary>
+   public class MsgProcessorWorker : IWorkerInterface
+   {
+      #region PrivateClassMembers
+      private IMsgParser _msgParser;
+      private DataBuffer _dataBuffer;
+      private AutoResetEvent _dataInBufferEvent;
+      private AutoResetEvent _quitEvent;
+      private bool _threadQuitControl;
+      private MsgDevice.CompleteMessageCallback _completeMsgCallback;
+      #endregion
+
+      #region PrivateFunctions
+      /// <summary>
+      /// Dispose of this objects resources
+      /// </summary>
+      /// <param name="disposing"></param>
+      protected virtual void Dispose(bool disposing)
+      {
+         try
+         {
+            if (disposing)
+            {
+               _quitEvent.Dispose();
+            }
+         }
+         catch (Exception err)
+         {
+            try
+            {
+               ErrorLogger.Instance().Write(err.Message + "\r\n" + err.StackTrace);
+            }
+            catch (Exception)
+            {
+               //Do not rethrow. Exception from error logger that has already been garbage collected
+            }
+         }
+      }
+
+      #endregion
+
+      #region PublicFuctions
+
+      /// <summary>
+      /// Constructor
+      /// </summary>
+      /// <param name="msgParser">Parses out the messages</param>
+      /// <param name="dataBuffer">The buffer to parse from</param>
+      /// <param name="dataInBufferEvent">A singal to let us know that data has arrived</param>
+      public MsgProcessorWorker(IMsgParser msgParser, ref DataBuffer dataBuffer, ref AutoResetEvent dataInBufferEvent)
+      {
+         _msgParser = msgParser;
+         _dataBuffer = dataBuffer;
+         _dataInBufferEvent = dataInBufferEvent;
+         _threadQuitControl = false;
+         _quitEvent = new AutoResetEvent(false);
+      }
+
+      /// <summary>
+      /// The finalizer for cleaing up resources
+      /// </summary>
+      ~MsgProcessorWorker()
+      {
+         Dispose(false);
+      }
+
+      /// <summary>
+      /// Dispose of this objects resources
+      /// </summary>
+      public void Dispose()
+      {
+         try
+         {
+            Dispose(true);
+
+            GC.SuppressFinalize(this);
+         }
+         catch (Exception err)
+         {
+            try
+            {
+               ErrorLogger.Instance().Write(err.Message + "\r\n" + err.StackTrace);
+            }
+            catch (Exception)
+            {
+               //Do not rethrow. Exception from error logger that has already been garbage collected
+            }
+         }
+      }
+
+      /// <summary>
+      /// Parses out messages from the data buffer and then issues callback to the handler with the completed message
+      /// </summary>
+      unsafe public void DoWork()
+      {
+         try
+         {
+            if (_completeMsgCallback == null)
+            {
+               throw new Exception("MsgProcessorWorker::DoWork() - Callback not set");
+            }
+
+            WaitHandle[] waithandles = new WaitHandle[2];
+            waithandles[0] = _dataInBufferEvent;
+            waithandles[1] = _quitEvent;
+
+            bool isTheWorkDone = false;
+
+            uint numBytesProcessedInLastIteration = 0;
+
+            while (isTheWorkDone == false)
+            {
+               int eventIndex = 0;
+
+               eventIndex = WaitHandle.WaitAny(waithandles, 100);
+               _dataInBufferEvent.Reset();
+
+               uint dataBufferBytesUsed = _dataBuffer.BytesUsed;
+               if (eventIndex == 0 || dataBufferBytesUsed > 0) // _dataInBufferEvent
+               {
+                  uint numBytesToProcess = 0;
+
+                  IntPtr pStartOfData = _dataBuffer.CheckOutStartOfData(ref numBytesToProcess);
+
+                  bool isThisRoundOfProcessingDone = false;
+
+                  uint ourCopyOfDataIndex = 0;
+
+                  numBytesProcessedInLastIteration = 0;
+
+                  while (isThisRoundOfProcessingDone == false)
+                  {
+                     uint numBytesLeftInTempBuffer = numBytesToProcess - ourCopyOfDataIndex;
+
+                     if (numBytesLeftInTempBuffer == 0)
+                     {
+                        isThisRoundOfProcessingDone = true;
+                     }
+                     else
+                     {
+                        uint numBytesToRemove = 0;
+                        uint msgId = 0;
+                        uint errorCode = 0;
+
+                        IntPtr payLoadPtr = IntPtr.Add(pStartOfData, (int)ourCopyOfDataIndex);
+
+                        if (_msgParser.Run(payLoadPtr, numBytesLeftInTempBuffer, ref numBytesToRemove, ref msgId, ref errorCode) == true)
+                        {
+                           string msg = "MsgProcessorWorker::DoWork() - removing " + numBytesToRemove.ToString() + " bytes, for msg id: " + msgId.ToString("X8");
+
+                           ErrorLogger.Instance().Write(msg, ErrorLogger.LogLevel.INFO);
+
+                           // we have a complete message, invoke the call back
+                           _completeMsgCallback(msgId, payLoadPtr, numBytesToRemove, errorCode);
+                        }
+
+                        if (numBytesToRemove == 0)
+                        {
+                           isThisRoundOfProcessingDone = true;
+                        }
+                        else
+                        {
+                           ourCopyOfDataIndex += numBytesToRemove;
+                           _dataBuffer.Remove(numBytesToRemove);
+                           numBytesProcessedInLastIteration += ourCopyOfDataIndex;
+                        }
+
+                        // were we signaled to quit?
+                        if (_threadQuitControl == true)
+                        {
+                           ErrorLogger.Instance().Write("MsgProcessorWorker::DoWork() - in the midst of procesing data, the quit event was detected, exiting", ErrorLogger.LogLevel.INFO);
+                           isTheWorkDone = true;
+                        }
+                     }
+                  }
+				  
+                  //Check start of data back in since we are done with it
+                  _dataBuffer.CheckInStartOfData();
+               }
+               else if (eventIndex == 1) // _quitEvent
+               {
+                  ErrorLogger.Instance().Write("MsgProcessorWorker::DoWork() - quit event was detected, exiting", ErrorLogger.LogLevel.INFO);
+                  isTheWorkDone = true;
+               }
+               else if (eventIndex == WaitHandle.WaitTimeout)
+               {
+                  // expected, continue
+               }
+               else
+               {
+                  ErrorLogger.Instance().Write("MsgProcessorWorker::DoWork() -  Unhandled return from WaitHandle.WaitAny(): " + eventIndex.ToString());
+               }
+            }
+
+            ErrorLogger.Instance().Write("MsgProcessorWorker::DoWork() - exiting", ErrorLogger.LogLevel.INFO);
+         }
+         catch (Exception err)
+         {
+            ErrorLogger.Instance().Write(err.Message + "\r\n" + err.StackTrace);
+         }
+      }
+
+      /// <summary>
+      /// Sets the call back function
+      /// </summary>
+      /// <param name="callback">The call back function</param>
+      public void SetCallback(MsgDevice.CompleteMessageCallback callback)
+      {
+         _completeMsgCallback = callback;
+      }
+
+      /// <summary>
+      /// Command the worker to stop
+      /// </summary>
+      public void QuitWork()
+      {
+         _threadQuitControl = true;
+         _quitEvent.Set();
+      }
+
+      #endregion
+   }
+}