GSF.Core: 1) Modified ProcessQueue to use ISynchronizedOperation for real-time processing mode. 2) Fixed Enabled state of MultiSourceFrameImageParserBase to not continue parsing buffers when disabled. 3) Added SynchronizedOperationType enum to ISynchronizedOperation.cs.

StephenCWills committed 12 years ago

Commit aa6665d1 parent a4927c53 files · +103 −195Browse files

Showing the changes to Source/Libraries/GSF.Core/Parsing/MultiSourceFrameImageParserBase.cs (1 of 3 files). Show all

Unified · Split

Source/Libraries/GSF.Core/Parsing/MultiSourceFrameImageParserBase.cs+33−62

         public event EventHandler<EventArgs<TSourceIdentifier, IList<TOutputType>>> SourceDataParsed;
         // Fields
-        private AsyncDoubleBufferedQueue<SourceIdentifiableBuffer> m_bufferQueue;
+        private readonly AsyncDoubleBufferedQueue<SourceIdentifiableBuffer> m_bufferQueue;
         private readonly ConcurrentDictionary<TSourceIdentifier, bool> m_sourceInitialized;
         private readonly ConcurrentDictionary<TSourceIdentifier, byte[]> m_unparsedBuffers;
         private readonly ConcurrentDictionary<TSourceIdentifier, List<TOutputType>> m_parsedSourceData;
         /// </summary>
         protected MultiSourceFrameImageParserBase()
         {
-            m_bufferQueue = CreateBufferQueue();
+            m_bufferQueue = new AsyncDoubleBufferedQueue<SourceIdentifiableBuffer>();
+            m_bufferQueue.ProcessItemsFunction = ParseQueuedBuffers;
             m_bufferQueue.ProcessException += ProcessExceptionHandler;
             if (ProtocolUsesSyncBytes)
         {
             get
             {
-                if ((object)m_bufferQueue != null)
+                return m_bufferQueue.Count;
-                    return m_bufferQueue.Count;
-                return 0;
             }
         }
             throw new NotImplementedException("This method should not be called directly, call the Parse(TSourceIdentifier,byte[],int,int) method to queue data for parsing instead.");
         }
-        /// <summary>
+        // This method is used by the internal <see cref="ProcessQueue{T}"/> to process all queued data buffers.
-        /// Creates the internal buffer queue.
+        private void ParseQueuedBuffers(IList<SourceIdentifiableBuffer> buffers)
-        /// </summary>
-        /// <remarks>
-        /// <para>
-        /// This method is virtual to allow derived classes to customize the style of processing queue used when consumers
-        /// choose to implement an internal buffer queue.  Default type is a real-time queue with the default settings.
-        /// When overriding, use the <see cref="ParseQueuedBuffers"/> method for the <see cref="ProcessQueue{T}"/>) item
-        /// processing delegate.
-        /// </para>
-        /// <para>
-        /// Note that current design only supports serial processing - consumer overriding this method to return an asynchronous
-        /// process queue to process multiple items at once on separate threads will need to redesign the processing delegate.
-        /// </para>
-        /// </remarks>
-        /// <returns>New internal buffer processing queue (i.e., a new <see cref="ProcessQueue{T}"/>).</returns>
-        protected virtual AsyncDoubleBufferedQueue<SourceIdentifiableBuffer> CreateBufferQueue()
-        {
-            AsyncDoubleBufferedQueue<SourceIdentifiableBuffer> queue = new AsyncDoubleBufferedQueue<SourceIdentifiableBuffer>();
-            queue.ProcessItemsFunction = ParseQueuedBuffers;
-            queue.ProcessException += ProcessExceptionHandler;
-            return queue;
-        }
-        /// <summary>
-        /// This method is used by the internal <see cref="ProcessQueue{T}"/> to process all queued data buffers.
-        /// </summary>
-        /// <param name="buffers">Identifiable buffers to process.</param>
-        /// <remarks>
-        /// This is the <see cref="SourceIdentifiableBuffer"/> processing delegate to use when overriding the <see cref="CreateBufferQueue"/> method.
-        /// </remarks>
-        protected virtual void ParseQueuedBuffers(IList<SourceIdentifiableBuffer> buffers)
         {
-            try
+            if (Enabled)
             {
-                // Process all queued data buffers...
+                try
-                foreach (SourceIdentifiableBuffer buffer in buffers)
                 {
-                    // Track current buffer source
+                    // Process all queued data buffers...
-                    m_source = buffer.Source;
+                    foreach (SourceIdentifiableBuffer buffer in buffers)
+                    {
+                        // Track current buffer source
+                        m_source = buffer.Source;
-                    // Check to see if this data source has been initialized
+                        // Check to see if this data source has been initialized
-                    if ((object)m_sourceInitialized != null)
+                        if ((object)m_sourceInitialized != null)
-                        StreamInitialized = m_sourceInitialized.GetOrAdd(m_source, true);
+                            StreamInitialized = m_sourceInitialized.GetOrAdd(m_source, true);
-                    // Restore any unparsed buffers for this data source, if any
+                        // Restore any unparsed buffers for this data source, if any
-                    UnparsedBuffer = m_unparsedBuffers.GetOrAdd(m_source, (byte[])null);
+                        UnparsedBuffer = m_unparsedBuffers.GetOrAdd(m_source, (byte[])null);
-                    // Start parsing sequence for this buffer - this will begin publication of new parsed outputs
+                        // Start parsing sequence for this buffer - this will begin publication of new parsed outputs
-                    base.Write(buffer.Buffer, 0, buffer.Count);
+                        base.Write(buffer.Buffer, 0, buffer.Count);
-                    // Track last unparsed buffer for this data source
+                        // Track last unparsed buffer for this data source
-                    m_unparsedBuffers[m_source] = UnparsedBuffer;
+                        m_unparsedBuffers[m_source] = UnparsedBuffer;
+                    }
                 }
-            }
+                finally
-            finally
-            {
-                // If user has attached to SourceDataParsed event, expose list of parsed data per source
-                if ((object)SourceDataParsed != null)
                 {
-                    foreach (KeyValuePair<TSourceIdentifier, List<TOutputType>> parsedData in m_parsedSourceData)
+                    // If user has attached to SourceDataParsed event, expose list of parsed data per source
+                    if ((object)SourceDataParsed != null)
                     {
-                        OnSourceDataParsed(parsedData.Key, parsedData.Value);
+                        foreach (KeyValuePair<TSourceIdentifier, List<TOutputType>> parsedData in m_parsedSourceData)
+                        {
+                            OnSourceDataParsed(parsedData.Key, parsedData.Value);
+                        }
+                        // Clear parsed data dictionary for next pass
+                        m_parsedSourceData.Clear();
                     }
-                    // Clear parsed data dictionary for next pass
-                    m_parsedSourceData.Clear();
                 }
             }
             // Dispose of source buffers - no rush

GSF.Core: 1) Modified ProcessQueue to use ISynchronizedOperation for real-time processing mode. 2) Fixed Enabled state of MultiSourceFrameImageParserBase to not continue parsing buffers when disabled. 3) Added SynchronizedOperationType enum to ISynchronizedOperation.cs.

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code