//****************************************************************************************************** // InputAdapterBase.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; using System.Text; using System.Threading; using System.Timers; using Timer = System.Timers.Timer; namespace GSF.TimeSeries.Adapters { /// /// Represents the base class for any incoming input adapter. /// /// /// Derived classes are expected to call when new measurements are received. /// public abstract class InputAdapterBase : AdapterBase, IInputAdapter { #region [ Members ] // Events /// /// Provides new measurements from input adapter. /// /// /// is a collection of new measurements for host to process. /// public event EventHandler>> NewMeasurements; /// /// Indicates to the host that processing for the input adapter has completed. /// /// /// This event is expected to only be raised when an input adapter has been designed to process /// a finite amount of data, e.g., reading a historical range of data during temporal processing. /// public event EventHandler ProcessingComplete; // Fields private List m_outputSourceIDs; private MeasurementKey[] m_requestedOutputMeasurementKeys; private Timer m_connectionTimer; private bool m_isConnected; private bool m_disposed; private bool m_enableConnectionErrors = true; #endregion #region [ Constructors ] /// /// Constructs a new instance of the . /// protected InputAdapterBase() { m_connectionTimer = new Timer(); m_connectionTimer.Elapsed += m_connectionTimer_Elapsed; m_connectionTimer.AutoReset = false; m_connectionTimer.Interval = 2000; m_connectionTimer.Enabled = false; } #endregion #region [ Properties ] /// /// Gets or sets values used to filter output measurements. /// /// /// This allows an adapter to associate itself with entire collections of measurements based on the source of the measurement keys. /// Set to null apply no filter. /// public virtual string[] OutputSourceIDs { get { if ((object)m_outputSourceIDs == null) return null; return m_outputSourceIDs.ToArray(); } set { if ((object)value == null) { m_outputSourceIDs = null; } else { m_outputSourceIDs = new List(value); m_outputSourceIDs.Sort(); } // Filter measurements to list of specified source IDs LoadOutputSourceIDs(this); } } /// /// Gets or sets output measurement keys that are requested by other adapters based on what adapter says it can provide. /// public virtual MeasurementKey[] RequestedOutputMeasurementKeys { get { return m_requestedOutputMeasurementKeys; } set { m_requestedOutputMeasurementKeys = value; } } /// /// Gets flag that determines if is connected. /// public virtual bool IsConnected { get { return m_isConnected; } protected set { m_isConnected = value; } } /// /// Gets flag that determines if the data input connects asynchronously. /// /// /// Derived classes should return true when data input source is connects asynchronously, otherwise return false. /// protected abstract bool UseAsyncConnect { get; } /// /// Gets or sets the connection attempt interval, in milliseconds, for the data input source. /// protected double ConnectionAttemptInterval { get { if (m_connectionTimer != null) return m_connectionTimer.Interval; return 2000.0D; } set { if (m_connectionTimer != null) m_connectionTimer.Interval = value; } } /// /// Returns the detailed status of the data input source. /// /// /// Derived classes should extend status with implementation specific information. /// public override string Status { get { const int MaxMeasurementsToShow = 10; StringBuilder status = new StringBuilder(); status.Append(base.Status); if (RequestedOutputMeasurementKeys != null && RequestedOutputMeasurementKeys.Length > 0) { status.AppendFormat(" Requested output keys: {0:N0} defined measurements", RequestedOutputMeasurementKeys.Length); status.AppendLine(); status.AppendLine(); for (int i = 0; i < Common.Min(RequestedOutputMeasurementKeys.Length, MaxMeasurementsToShow); i++) { status.AppendLine(RequestedOutputMeasurementKeys[i].ToString().TruncateRight(25).CenterText(50)); } if (RequestedOutputMeasurementKeys.Length > MaxMeasurementsToShow) status.AppendLine("...".CenterText(50)); status.AppendLine(); } status.AppendFormat(" Connection established: {0}", IsConnected); status.AppendLine(); status.AppendFormat(" Asynchronous connection: {0}", UseAsyncConnect); status.AppendLine(); return status.ToString(); } } /// /// Gets or sets primary keys of input measurements the expects, if any. /// /// /// Redefined to hide attributes defined in the base class. /// public new virtual MeasurementKey[] InputMeasurementKeys { get { return base.InputMeasurementKeys; } set { base.InputMeasurementKeys = value; } } /// /// When false, connection errors do not get logged through OnProcessException. When true, connection errors will be logged normally. /// protected bool EnableConnectionErrors { get { return m_enableConnectionErrors; } set { m_enableConnectionErrors = value; } } #endregion #region [ Methods ] /// /// Releases the unmanaged resources used by the object and optionally releases the managed resources. /// /// true to release both managed and unmanaged resources; false to release only unmanaged resources. protected override void Dispose(bool disposing) { if (!m_disposed) { try { if (disposing) { if (m_connectionTimer != null) { m_connectionTimer.Elapsed -= m_connectionTimer_Elapsed; m_connectionTimer.Dispose(); } m_connectionTimer = null; } } finally { m_disposed = true; // Prevent duplicate dispose. base.Dispose(disposing); // Call base class Dispose(). } } } /// /// Initializes . /// public override void Initialize() { base.Initialize(); Dictionary settings = Settings; string setting; // Load optional parameters if (settings.TryGetValue("outputSourceIDs", out setting) || settings.TryGetValue("sourceids", out setting)) OutputSourceIDs = setting.Split(','); else OutputSourceIDs = null; } /// /// Starts this and initiates connection cycle to data input source. /// public override void Start() { base.Start(); // Start the connection cycle if (m_connectionTimer != null) m_connectionTimer.Enabled = true; } /// /// Attempts to connect to data input source. /// /// /// Derived classes should attempt connection to data input source here. Any exceptions thrown /// by this implementation will result in restart of the connection cycle. /// protected abstract void AttemptConnection(); /// /// Called when data input source connection is established. /// /// /// Derived classes should call this method manually if is true. /// protected virtual void OnConnected() { IsConnected = true; OnStatusMessage("Connection established."); } /// /// Stops this and disconnects from data input source. /// public override void Stop() { try { bool performedDisconnect = Enabled; // Stop the connection cycle if (m_connectionTimer != null) m_connectionTimer.Enabled = false; base.Stop(); // Attempt disconnection from data source AttemptDisconnection(); if (performedDisconnect && !UseAsyncConnect) OnDisconnected(); } catch (ThreadAbortException) { // This exception can be safely ignored... } catch (Exception ex) { if(EnableConnectionErrors) OnProcessException(new InvalidOperationException(string.Format("Exception occurred during disconnect: {0}", ex.Message), ex)); } } /// /// Attempts to disconnect from data input source. /// /// /// Derived classes should attempt disconnect from data input source here. Any exceptions thrown /// by this implementation will be reported to host via event. /// protected abstract void AttemptDisconnection(); /// /// Called when data input source is disconnected. /// /// /// Derived classes should call this method manually if is true. /// protected virtual void OnDisconnected() { IsConnected = false; OnStatusMessage("Disconnected."); } /// /// Raises the event. /// protected virtual void OnNewMeasurements(ICollection measurements) { try { if (NewMeasurements != null) NewMeasurements(this, new EventArgs>(measurements)); IncrementProcessedMeasurements(measurements.Count); } catch (Exception ex) { // We protect our code from consumer thrown exceptions OnProcessException(new InvalidOperationException(string.Format("Exception in consumer handler for NewMeasurements event: {0}", ex.Message), ex)); } } /// /// Raises the event. /// protected virtual void OnProcessingComplete() { try { if (ProcessingComplete != null) ProcessingComplete(this, EventArgs.Empty); } catch (Exception ex) { // We protect our code from consumer thrown exceptions OnProcessException(new InvalidOperationException(string.Format("Exception in consumer handler for ProcessingComplete event: {0}", ex.Message), ex)); } } private void m_connectionTimer_Elapsed(object sender, ElapsedEventArgs e) { try { // So long as user hasn't requested to stop, attempt connection if (Enabled) { OnStatusMessage("Attempting connection..."); // Attempt connection to data source AttemptConnection(); if (!UseAsyncConnect) OnConnected(); } } catch (ThreadAbortException) { // This exception can be safely ignored... } catch (Exception ex) { if(EnableConnectionErrors) OnProcessException(new InvalidOperationException(string.Format("Connection attempt failed: {0}", ex.Message), ex)); // So long as user hasn't requested to stop, keep trying connection if (Enabled) Start(); } } #endregion } }