//******************************************************************************************************
// InputAdapterBase.cs - Gbtc
//
// Copyright © 2010, 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 Eclipse Public License -v 1.0 (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/eclipse-1.0.php
//
// 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.
//
//******************************************************************************************************
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
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 procesing.
///
public event EventHandler ProcessingComplete;
// Fields
private List m_outputSourceIDs;
private MeasurementKey[] m_requestedOutputMeasurementKeys;
private System.Timers.Timer m_connectionTimer;
private bool m_isConnected;
private bool m_disposed;
#endregion
#region [ Constructors ]
///
/// Constructs a new instance of the .
///
protected InputAdapterBase()
{
m_connectionTimer = new System.Timers.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 (m_outputSourceIDs == null)
return null;
return m_outputSourceIDs.ToArray();
}
set
{
if (value == null)
{
m_outputSourceIDs = null;
}
else
{
m_outputSourceIDs = new List(value);
m_outputSourceIDs.Sort();
}
// Filter measurements to list of specified source IDs
AdapterBase.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} 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();
status.AppendFormat(" Item reporting interval: {0}", MeasurementReportingInterval);
status.AppendLine();
return status.ToString();
}
}
#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)
{
OnProcessException(new InvalidOperationException(string.Format("Exception occured 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, System.Timers.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)
{
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
}
}