<commit_before>//******************************************************************************************************
//  OutputAdapterBase.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 - <NAME>
//       Generated original version of source code.
//
//******************************************************************************************************

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using TVA;
using TVA.Collections;

namespace TimeSeriesFramework.Adapters
{
    /// <summary>
    /// Represents the base class for any outgoing data stream.
    /// </summary>
    /// <remarks>
    /// This base class acts as a measurement queue so that output adapters can temporarily go
    /// offline without losing any measurements to be processed. Derived classes are expected to
    /// override <see cref="ProcessMeasurements"/> to handle queued measurements.
    /// </remarks>
    public abstract class OutputAdapterBase : AdapterBase, IOutputAdapter
    {
        #region [ Members ]

        // Events

        /// <summary>
        /// Event is raised every second allowing host to track total number of unprocessed measurements.
        /// </summary>
        /// <remarks>
        /// <para>
        /// Base class reports current queue size of unprocessed measurements so that if queue size reaches
        /// an unhealthy threshold, host can take evasive action.
        /// </para>
        /// <para>
        /// <see cref="EventArgs{T}.Argument"/> is total number of unprocessed measurements.
        /// </para>
        /// </remarks>
        public event EventHandler<EventArgs<int>> UnprocessedMeasurements;

        // Fields
        private ProcessQueue<IMeasurement> m_measurementQueue;
        private List<string> m_inputSourceIDs;
        private MeasurementKey[] m_requestedInputMeasurementKeys;
        private System.Timers.Timer m_connectionTimer;
        private System.Timers.Timer m_monitorTimer;
        private bool m_bypassProcessing;
        private bool m_disposed;

        #endregion

        #region [ Constructors ]

        /// <summary>
        /// Constructs a new instance of the <see cref="OutputAdapterBase"/>.
        /// </summary>
        protected OutputAdapterBase()
        {
            m_measurementQueue = ProcessQueue<IMeasurement>.CreateRealTimeQueue(ProcessMeasurements);
            m_measurementQueue.ProcessException += m_measurementQueue_ProcessException;

            m_connectionTimer = new System.Timers.Timer();
            m_connectionTimer.Elapsed += m_connectionTimer_Elapsed;

            m_connectionTimer.AutoReset = false;
            m_connectionTimer.Interval = 2000;
            m_connectionTimer.Enabled = false;

            m_monitorTimer = new System.Timers.Timer();
            m_monitorTimer.Elapsed += m_monitorTimer_Elapsed;

            // We monitor total number of unarchived measurements every second - this is a useful statistic to monitor, if
            // total number of unarchived measurements gets very large, measurement archival could be falling behind
            m_monitorTimer.Interval = 1000;
            m_monitorTimer.AutoReset = true;
            m_monitorTimer.Enabled = false;
        }

        #endregion

        #region [ Properties ]

        /// <summary>
        /// Gets or sets whether or not to automatically place measurements back into the processing
        /// queue if an exception occurs while processing.  Defaults to false.
        /// </summary>
        public virtual bool RequeueOnException
        {
            get
            {
                return m_measurementQueue.RequeueOnException;
            }
            set
            {
                m_measurementQueue.RequeueOnException = value;
            }
        }

        /// <summary>
        /// Gets or sets <see cref="MeasurementKey.Source"/> values used to filter input measurements.
        /// </summary>
        /// <remarks>
        /// This allows an adapter to associate itself with entire collections of measurements based on the source of the measurement keys.
        /// Set to <c>null</c> apply no filter.
        /// </remarks>
        public virtual string[] InputSourceIDs
        {
            get
            {
                if (m_inputSourceIDs == null)
                    return null;

                return m_inputSourceIDs.ToArray();
            }
            set
            {
                if (value == null)
                {
                    m_inputSourceIDs = null;
                }
                else
                {
                    m_inputSourceIDs = new List<string>(value);
                    m_inputSourceIDs.Sort();
                }

                // Filter measurements to list of specified source IDs
                AdapterBase.LoadInputSourceIDs(this);
            }
        }

        /// <summary>
        /// Gets or sets input measurement keys that are requested by other adapters based on what adapter says it can provide.
        /// </summary>
        public virtual MeasurementKey[] RequestedInputMeasurementKeys
        {
            get
            {
                return m_requestedInputMeasurementKeys;
            }
            set
            {
                m_requestedInputMeasurementKeys = value;
            }
        }

        /// <summary>
        /// Returns a flag that determines if measurements sent to this <see cref="OutputAdapterBase"/> are
        /// destined for archival.
        /// </summary>
        /// <remarks>
        /// This property allows the <see cref="OutputAdapterCollection"/> to calculate statistics on how
        /// many measurements have been archived per minute. Historians would normally set this property
        /// to <c>true</c>; other custom exports would set this property to <c>false</c>.
        /// </remarks>
        public abstract bool OutputIsForArchive
        {
            get;
        }

        /// <summary>
        /// Gets or sets value that allows output adapter processing to be bypassed.
        /// </summary>
        /// <remarks>
        /// For output adapters it is assumed that the desired behavior when a temporal constraint exists is to disable
        /// processing since the data being processed by the output adapter has already been archived. Note that this
        /// bypass behavior is only enagaged when <see cref="OutputIsForArchive"/> is <c>true</c>. Processing bypass
        /// occurs by setting <see cref="BypassProcessing"/> to <c>true</c>. If you have an output adapter that you
        /// want to process data even when a temporal constaint is defined, override the <see cref="BypassProcessing"/>
        /// property and force the base value to always be <c>false</c>.
        /// </remarks>
        public virtual bool BypassProcessing
        {
            get
            {
                return m_bypassProcessing;
            }
            set
            {
                m_bypassProcessing = value;
            }
        }

        /// <summary>
        /// Gets flag that determines if the data output stream connects asynchronously.
        /// </summary>
        /// <remarks>
        /// Derived classes should return true when data output stream connects asynchronously, otherwise return false.
        /// </remarks>
        protected abstract bool UseAsyncConnect
        {
            get;
        }

        /// <summary>
        /// Gets or sets the connection attempt interval, in milliseconds, for the data output adapter.<commit_msg>Updated default behavior for output adapters to properly respond to changes in processing interval - even if applied dynamically.<commit_after>//******************************************************************************************************
//  OutputAdapterBase.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 - <NAME>
//       Generated original version of source code.
//
//******************************************************************************************************

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using TVA;
using TVA.Collections;

namespace TimeSeriesFramework.Adapters
{
    /// <summary>
    /// Represents the base class for any outgoing data stream.
    /// </summary>
    /// <remarks>
    /// This base class acts as a measurement queue so that output adapters can temporarily go
    /// offline without losing any measurements to be processed. Derived classes are expected to
    /// override <see cref="ProcessMeasurements"/> to handle queued measurements.
    /// </remarks>
    public abstract class OutputAdapterBase : AdapterBase, IOutputAdapter
    {
        #region [ Members ]

        // Events

        /// <summary>
        /// Event is raised every second allowing host to track total number of unprocessed measurements.
        /// </summary>
        /// <remarks>
        /// <para>
        /// Base class reports current queue size of unprocessed measurements so that if queue size reaches
        /// an unhealthy threshold, host can take evasive action.
        /// </para>
        /// <para>
        /// <see cref="EventArgs{T}.Argument"/> is total number of unprocessed measurements.
        /// </para>
        /// </remarks>
        public event EventHandler<EventArgs<int>> UnprocessedMeasurements;

        // Fields
        private ProcessQueue<IMeasurement> m_measurementQueue;
        private List<string> m_inputSourceIDs;
        private MeasurementKey[] m_requestedInputMeasurementKeys;
        private System.Timers.Timer m_connectionTimer;
        private System.Timers.Timer m_monitorTimer;
        private bool m_bypassProcessing;
        private bool m_disposed;

        #endregion

        #region [ Constructors ]

        /// <summary>
        /// Constructs a new instance of the <see cref="OutputAdapterBase"/>.
        /// </summary>
        protected OutputAdapterBase()
        {
            m_measurementQueue = ProcessQueue<IMeasurement>.CreateRealTimeQueue(ProcessMeasurements);
            m_measurementQueue.ProcessException += m_measurementQueue_ProcessException;

            m_connectionTimer = new System.Timers.Timer();
            m_connectionTimer.Elapsed += m_connectionTimer_Elapsed;

            m_connectionTimer.AutoReset = false;
            m_connectionTimer.Interval = 2000;
            m_connectionTimer.Enabled = false;

            m_monitorTimer = new System.Timers.Timer();
            m_monitorTimer.Elapsed += m_monitorTimer_Elapsed;

            // We monitor total number of unarchived measurements every second - this is a useful statistic to monitor, if
            // total number of unarchived measurements gets very large, measurement archival could be falling behind
            m_monitorTimer.Interval = 1000;
            m_monitorTimer.AutoReset = true;
            m_monitorTimer.Enabled = false;
        }

        #endregion

        #region [ Properties ]

        /// <summary>
        /// Gets or sets whether or not to automatically place measurements back into the processing
        /// queue if an exception occurs while processing.  Defaults to false.
        /// </summary>
        public virtual bool RequeueOnException
        {
            get
            {
                return m_measurementQueue.RequeueOnException;
            }
            set
            {
                m_measurementQueue.RequeueOnException = value;
            }
        }

        /// <summary>
        /// Gets or sets <see cref="MeasurementKey.Source"/> values used to filter input measurements.
        /// </summary>
        /// <remarks>
        /// This allows an adapter to associate itself with entire collections of measurements based on the source of the measurement keys.
        /// Set to <c>null</c> apply no filter.
        /// </remarks>
        public virtual string[] InputSourceIDs
        {
            get
            {
                if (m_inputSourceIDs == null)
                    return null;

                return m_inputSourceIDs.ToArray();
            }
            set
            {
                if (value == null)
                {
                    m_inputSourceIDs = null;
                }
                else
                {
                    m_inputSourceIDs = new List<string>(value);
                    m_inputSourceIDs.Sort();
                }

                // Filter measurements to list of specified source IDs
                AdapterBase.LoadInputSourceIDs(this);
            }
        }

        /// <summary>
        /// Gets or sets input measurement keys that are requested by other adapters based on what adapter says it can provide.
        /// </summary>
        public virtual MeasurementKey[] RequestedInputMeasurementKeys
        {
            get
            {
                return m_requestedInputMeasurementKeys;
            }
            set
            {
                m_requestedInputMeasurementKeys = value;
            }
        }

        /// <summary>
        /// Returns a flag that determines if measurements sent to this <see cref="OutputAdapterBase"/> are
        /// destined for archival.
        /// </summary>
        /// <remarks>
        /// This property allows the <see cref="OutputAdapterCollection"/> to calculate statistics on how
        /// many measurements have been archived per minute. Historians would normally set this property
        /// to <c>true</c>; other custom exports would set this property to <c>false</c>.
        /// </remarks>
        public abstract bool OutputIsForArchive
        {
            get;
        }

        /// <summary>
        /// Gets or sets value that allows output adapter processing to be bypassed.
        /// </summary>
        /// <remarks>
        /// For output adapters it is assumed that the desired behavior when a temporal constraint exists is to disable
        /// processing since the data being processed by the output adapter has already been archived. Note that this
        /// bypass behavior is only enagaged when <see cref="OutputIsForArchive"/> is <c>true</c>. Processing bypass
        /// occurs by setting <see cref="BypassProcessing"/> to <c>true</c>. If you have an output adapter that you
        /// want to process data even when a temporal constaint is defined, override the <see cref="BypassProcessing"/>
        /// property and force the base value to always be <c>false</c>.
        /// </remarks>
        public virtual bool BypassProcessing
        {
            get
            {
                return m_bypassProcessing;
            }
            set
            {
                m_bypassProcessing = value;
            }
        }

        /// <summary>
        /// Gets or sets the desired processing interval, in milliseconds, for the output adapter.
        /// </summary>
        /// <remarks>
        /// With the exception of the values of -1 and 0, this value specifies the desired processing interval for data, i.e.,
        /// basically a delay, or timer interval, overwhich to process data. A value of -1 means to use the default processing
        /// interval while a value of 0 means to process data as fast as possible.
        /// </remarks>
        public override int ProcessingInterval
        {
            get
            {
                return base.ProcessingInterval;
            }
            set
            {
                if (base.ProcessingInterval != value)
                {
                    base.ProcessingInterval = value;
                    bool enabled = false;
                    bool requeueOnException = ProcessQueue<IMeasurement>.DefaultRequeueOnException;
                    IMeasurement[] unprocessedMeasurements = null;

                    if (m_measurementQueue != null)
                    {
                        enabled = m_measurementQueue.Enabled;
                        requeueOnException = m_measurementQueue.RequeueOnException;

                        lock (m_measurementQueue.SyncRoot)
                        {
                            if (m_measurementQueue.Count > 0)
                            {
                                m_measurementQueue.Stop();
                                unprocessedMeasurements = m_measurementQueue.ToArray();
                            }
                        }

                        m_measurementQueue.ProcessException -= m_measurementQueue_ProcessException;
                        m_measurementQueue.Dispose();
                    }

                    if (value <= 0)
                    {
                        // The default processing interval is "as fast as possible"
                        m_measurementQueue = ProcessQueue<IMeasurement>.CreateRealTimeQueue(ProcessMeasurements);
                    }
                    else
                    {
                        // Set the desired processing interval
                        m_measurementQueue = ProcessQueue<IMeasurement>.CreateSynchronousQueue(ProcessMeasurements);
                        m_measurementQueue.ProcessInterval = value;
                    }

                    m_measurementQueue.ProcessException += m_measurementQueue_ProcessException;
                    m_measurementQueue.RequeueOnException = requeueOnException;

                    // Requeue any existing measurements
                    if (unprocessedMeasurements != null && unprocessedMeasurements.Length > 0)
                        m_measurementQueue.AddRange(unprocessedMeasurements);

                    m_measurementQueue.Enabled = enabled;
                }
            }
        }

        /// <summary>
        /// Gets flag that determines if the data output stream connects asynchronously.
        /// </summary>
        /// <remarks>
        /// Derived classes should return true when data output stream connects asynchronously, otherwise return false.
        /// </remarks>
        protected abstract bool UseAsyncConnect
        {
            get;
        }

        /// <summary>
        /// Gets or sets the connection attempt interval, in milliseconds, for the data output adapter.
{'fraction_non_alphanumeric': 0.10695134061569017, 'fraction_numerical': 0.002730883813306852, 'mean_word_length': 1.7560207991242474, 'pattern_counts': {'":': 0, '<': 137, '<?xml version=': 0, '>': 138, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7397130', 'n_tokens_mistral': 4644, 'n_tokens_neox': 4406, 'n_words': 1607}