//****************************************************************************************************** // OutputAdapterCollection.cs - Gbtc // // Copyright © 2012, Grid Protection Alliance. All Rights Reserved. // // Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See // the NOTICE file distributed with this work for additional information regarding copyright ownership. // The GPA licenses this file to you under the MIT License (MIT), the "License"; you may // not use this file except in compliance with the License. You may obtain a copy of the License at: // // http://www.opensource.org/licenses/MIT // // Unless agreed to in writing, the subject software distributed under the License is distributed on an // "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the // License for the specific language governing permissions and limitations. // // Code Modification History: // ---------------------------------------------------------------------------------------------------- // 09/02/2010 - J. Ritchie Carroll // Generated original version of source code. // 12/20/2012 - Starlynn Danyelle Gilliam // Modified Header. // //****************************************************************************************************** using System; using System.Collections.Generic; namespace GSF.TimeSeries.Adapters { /// /// Represents a collection of implementations. /// public class OutputAdapterCollection : AdapterCollectionBase, IOutputAdapter { #region [ Members ] // Events /// /// Event is raised every five seconds allowing host to track total number of unprocessed measurements. /// /// /// /// Each implementation reports its current queue size of unprocessed /// measurements so that if queue size reaches an unhealthy threshold, host can take evasive action. /// /// /// is total number of unprocessed measurements. /// /// public event EventHandler> UnprocessedMeasurements; #endregion #region [ Constructors ] /// /// Creates a new . /// public OutputAdapterCollection() { base.Name = "Output Adapter Collection"; base.DataMember = "OutputAdapters"; base.MonitorTimerEnabled = true; } #endregion #region [ Properties ] /// /// Gets the total number of measurements processed and destined for archive thus far by each /// implementation in the . /// public override long ProcessedMeasurements { get { long processedMeasurements = 0; // Calculate new total for all archive destined output adapters lock (this) { foreach (IOutputAdapter item in this) { if (item.OutputIsForArchive) processedMeasurements += item.ProcessedMeasurements; } } return processedMeasurements; } } /// /// Returns a flag that determines if all measurements sent to this are /// destined for archival. /// public virtual bool OutputIsForArchive { get { lock (this) { foreach (IOutputAdapter item in this) { if (item.OutputIsForArchive) return true; } } return false; } } #endregion #region [ Methods ] /// /// Queues a collection of measurements for processing to each implementation in /// this . /// /// Measurements to queue for processing. public virtual void QueueMeasurementsForProcessing(IEnumerable measurements) { try { lock (this) { foreach (IOutputAdapter item in this) { item.QueueMeasurementsForProcessing(measurements); } } } catch (Exception ex) { OnProcessException(new InvalidOperationException("Failed to queue measurements to output adapters: " + ex.Message, ex)); } } /// /// This function removes a range of measurements from the internal measurement queues. Note that the requested /// will be removed from each implementation in this /// . /// /// Total measurements to remove from the each queue. /// /// This method is typically only used to curtail size of measurement queue if it's getting too large. If more points are /// requested than there are points available - all points in the queue will be removed. /// public virtual void RemoveMeasurements(int total) { lock (this) { foreach (IOutputAdapter item in this) { if (item.Enabled) item.RemoveMeasurements(total); } } } /// /// Raises the event. /// /// Total measurements in the queue that have not been processed. protected virtual void OnUnprocessedMeasurements(int unprocessedMeasurements) { try { if (UnprocessedMeasurements != null) UnprocessedMeasurements(this, new EventArgs(unprocessedMeasurements)); } catch (Exception ex) { // We protect our code from consumer thrown exceptions OnProcessException(new InvalidOperationException(string.Format("Exception in consumer handler for UnprocessedMeasurements event: {0}", ex.Message), ex)); } } /// /// Wires events and initializes new implementation. /// /// New implementation. protected override void InitializeItem(IOutputAdapter item) { if (item != null) { // Wire up unprocessed measurements event item.UnprocessedMeasurements += item_UnprocessedMeasurements; base.InitializeItem(item); } } /// /// Unwires events and disposes of implementation. /// /// to dispose. protected override void DisposeItem(IOutputAdapter item) { if (item != null) { // Un-wire unprocessed measurements event item.UnprocessedMeasurements -= item_UnprocessedMeasurements; base.DisposeItem(item); } } // Raise unprocessed measurements event on behalf of each item in collection private void item_UnprocessedMeasurements(object sender, EventArgs e) { if (UnprocessedMeasurements != null) UnprocessedMeasurements(sender, e); } #endregion } }