//******************************************************************************************************//  LocalOutputAdapter.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/10/2009 - Pinal C. Patel//       Generated original version of source code.//  09/11/2009 - Pinal C. Patel//       Added support to refresh metadata from one or more external sources.//  09/15/2009 - Stephen C. Wills//       Added new header and license agreement.//  09/17/2009 - Pinal C. Patel//       Added option to refresh metadata during connection.//       Modified RefreshMetadata() to perform synchronous refresh.//       Corrected the implementation of Dispose().//  09/18/2009 - Pinal C. Patel//       Added override to Status property and added event handler to archive rollover notification.//  10/28/2009 - Pinal C. Patel//       Modified to allow for multiple instances of the adapter to be loaded and configured with //       different settings by persisting the settings in the config file under unique categories.//  11/18/2009 - Pinal C. Patel//       Added support for the replication of local historian archive.//  12/01/2009 - Pinal C. Patel//       Modified Initialize() to load all available metadata providers.//  12/11/2009 - Pinal C. Patel//       Fixed the implementation for allowing multiple adapter instances.//       Expanded the adapter status to include dynamically loaded plug-ins.//  04/28/2010 - Pinal C. Patel//       Modified ProcessMeasurements() method to not throw an exception if the archive file is not //       open as this will be handled by ArchiveFile.WriteData() method if necessary.//  06/13/2010 - J. Ritchie Carroll//       Modified loaded plug-in's to use lower-cased instance name for configuration settings for//       consistency and better looking configuration categories. Added static data operation to //       automatically optimize settings for defined local historians.//  09/24/2010 - J. Ritchie Carroll//       Added provider and service section to list of category sections to be removed when unused. //       Added automatic URL namespace reservation for built-in web services.//  11/07/2010 - Pinal C. Patel//       Modified namespace reservation logic to handle the changed URI format in SelfHostingService.//  11/21/2011 - J. Ritchie Carroll//       Modified historian optimization procedure to dynamically add reading adapters.//  12/13/2012 - Starlynn Danyelle Gilliam//       Modified Header.////****************************************************************************************************** using System;using System.Collections.Concurrent;using System.Collections.Generic;using System.ComponentModel;using System.Configuration;using System.Data;using System.IO;using System.Linq;using System.Text;using System.Threading;using GSF;using GSF.Collections;using GSF.Configuration;using GSF.Data;using GSF.Diagnostics;using GSF.Historian;using GSF.Historian.DataServices;using GSF.Historian.Files;using GSF.Historian.MetadataProviders;using GSF.Historian.Replication;using GSF.IO;using GSF.Threading;using GSF.TimeSeries;using GSF.TimeSeries.Adapters; namespace HistorianAdapters{    /// <summary>    /// Represents an output adapter that archives measurements to a local archive.    /// </summary>    [Description("Local v1.0 openHistorian: [Deprecated for v2.0, see openHistorian.com] Archives measurements to a local in-process 1.0 openHistorian.")]    public class LocalOutputAdapter : OutputAdapterBase    {        #region [ Members ]         // Fields        private readonly ArchiveFile m_archive;        private DataServices m_dataServices;        private MetadataProviders m_metadataProviders;        private ReplicationProviders m_replicationProviders;        private bool m_autoRefreshMetadata;        private bool m_attemptingConnection;        private string m_instanceName;        private string m_archivePath;        private long m_archivedMeasurements;        private volatile int m_adapterLoadedCount;        private ulong m_badDataMessageInterval;        private readonly Dictionary<int, ulong> m_orphanCounts;        private readonly ProcessQueue<IDataPoint> m_orphanQueue;        private readonly Dictionary<int, ulong> m_badTimestampCounts;        private readonly ProcessQueue<IDataPoint> m_badTimestampQueue;        private readonly Dictionary<int, ulong> m_outOfSequenceCounts;        private readonly ProcessQueue<IDataPoint> m_outOfSequenceQueue;        private readonly ShortSynchronizedOperation m_updateArchiveFileSettings;        private bool m_disposed;         #endregion         #region [ Constructors ]         /// <summary>        /// Initializes a new instance of the <see cref="LocalOutputAdapter"/> class.        /// </summary>        public LocalOutputAdapter()        {            m_autoRefreshMetadata = true;            m_archive = new ArchiveFile();            m_archive.MetadataFile = new MetadataFile();            m_archive.StateFile = new StateFile();            m_archive.IntercomFile = new IntercomFile();            MetadataRefreshOperation.IsBackground = false;            m_badDataMessageInterval = 900;            m_orphanCounts = new Dictionary<int, ulong>();            m_orphanQueue = ProcessQueue<IDataPoint>.CreateRealTimeQueue(HandleOrphanData);            m_badTimestampCounts = new Dictionary<int, ulong>();            m_badTimestampQueue = ProcessQueue<IDataPoint>.CreateRealTimeQueue(HandleBadTimestampData);            m_outOfSequenceCounts = new Dictionary<int, ulong>();            m_outOfSequenceQueue = ProcessQueue<IDataPoint>.CreateRealTimeQueue(HandleOutOfSequenceData);            m_updateArchiveFileSettings = new ShortSynchronizedOperation(UpdateArchiveFileSettings, ex => OnProcessException(MessageLevel.Warning, new InvalidOperationException($"Failed while updating archive file configuration parameters after configuration reload: {ex.Message}", ex)));        }         #endregion         #region [ Properties ]         /// <summary>        /// Gets or sets instance name defined for this <see cref="LocalOutputAdapter"/>.        /// </summary>        [ConnectionStringParameter,        Description("Define the instance name for the archive. Leave this value blank to default to the adapter name (blank is typical setting)."),        DefaultValue("")]        public string InstanceName        {            get            {                return m_instanceName;            }            set            {                m_instanceName = value;            }        }         /// <summary>        /// Gets or sets a boolean indicating whether or not metadata is        /// refreshed when the adapter attempts to connect to the archive.        /// </summary>        [ConnectionStringParameter,        Description("Define a boolean indicating whether to refresh metadata from database on connect."),        DefaultValue(true)]        public bool AutoRefreshMetadata        {            get            {                return m_autoRefreshMetadata;            }            set            {                m_autoRefreshMetadata = value;            }        }         /// <summary>        /// Gets or sets primary keys of input measurements the <see cref="AdapterBase"/> expects, if any.        /// </summary>        [EditorBrowsable(EditorBrowsableState.Never)]        public override MeasurementKey[] InputMeasurementKeys        {            get            {                return base.InputMeasurementKeys;            }            set            {                base.InputMeasurementKeys = value;            }        }         /// <summary>        /// Gets or sets the path to the archive.        /// </summary>        public string ArchivePath        {            get            {                return m_archivePath;            }            set            {                m_archivePath = value;            }        }         /// <summary>        /// Gets connection info for adapter, if any.        /// </summary>        public override string ConnectionInfo => $"{InstanceName} [{ArchivePath}]";         /// <summary>        /// Returns a flag that determines if measurements sent to this <see cref="LocalOutputAdapter"/> are destined for archival.        /// </summary>        public override bool OutputIsForArchive => true;         /// <summary>        /// Gets flag that determines if this <see cref="LocalOutputAdapter"/> uses an asynchronous connection.        /// </summary>        protected override bool UseAsyncConnect => true;         /// <summary>        /// Gets or sets <see cref="DataSet"/> based data source available to this <see cref="LocalOutputAdapter"/>.        /// </summary>        public override DataSet DataSource        {            get            {                return base.DataSource;            }            set            {                base.DataSource = value;                 // When configuration is reloaded, queue update to adjust archive file size parameters                m_updateArchiveFileSettings.RunOnceAsync();            }        }         /// <summary>        /// Returns the detailed status of the data output source.        /// </summary>        public override string Status        {            get            {                StringBuilder status = new StringBuilder();                 status.Append(base.Status);                status.AppendLine();                status.Append(m_archive.Status);                status.AppendLine();                status.Append(m_archive.MetadataFile.Status);                status.AppendLine();                status.Append(m_archive.StateFile.Status);                status.AppendLine();                status.Append(m_archive.IntercomFile.Status);                status.AppendLine();                status.Append(m_dataServices.Status);                status.AppendLine();                status.Append(m_metadataProviders.Status);                status.AppendLine();                status.Append(m_replicationProviders.Status);                 return status.ToString();            }        }         #endregion         #region [ Methods ]         /// <summary>        /// Refreshes metadata using all available and enabled providers.        /// </summary>        protected override void ExecuteMetadataRefresh()        {            try            {                if ((object)m_archive != null && m_archive.IsOpen && (object)m_archive.StateFile != null && m_archive.StateFile.IsOpen && (object)m_archive.MetadataFile != null && m_archive.MetadataFile.IsOpen)                {                    if (!m_attemptingConnection)                    {                        OnStatusMessage(MessageLevel.Info, "Pausing measurement processing...");                        InternalProcessQueue.Stop();                    }                     // Synchronously refresh the meta-base.                    lock (m_metadataProviders.Adapters)                    {                        foreach (IMetadataProvider provider in m_metadataProviders.Adapters)                        {                            if (provider.Enabled)                                provider.Refresh();                        }                    }                     // Request a state file synchronization in case file watchers are disabled                    m_archive.SynchronizeStateFile();                     // Wait for the meta-base to synchronize, up to five seconds                    int waitCounts = 0;                     while (m_archive.StateFile.RecordsOnDisk != m_archive.MetadataFile.RecordsOnDisk && waitCounts < 50)                    {                        Thread.Sleep(100);                        waitCounts++;                    }                }            }            finally            {                if (Enabled && !InternalProcessQueue.Enabled)                {                    if (m_attemptingConnection)                    {                        // OnConnected starts the internal process queue                        // and sets m_attemptingConnection to false                        OnConnected();                    }                    else                    {                        OnStatusMessage(MessageLevel.Info, "Resuming measurement processing...");                        InternalProcessQueue.Start();                    }                }            }        }         /// <summary>        /// Initializes this <see cref="LocalOutputAdapter"/>.        /// </summary>        /// <exception cref="ArgumentException"><b>InstanceName</b> is missing from the <see cref="AdapterBase.Settings"/>.</exception>        public override void Initialize()        {            base.Initialize();             Dictionary<string, string> settings = Settings;            string setting;            ulong badDataMessageInterval;             // Validate settings.            if (!settings.TryGetValue("instanceName", out m_instanceName) || string.IsNullOrWhiteSpace(m_instanceName))                m_instanceName = Name.ToLower();             // Track instance in static dictionary            Instances[InstanceName] = this;             if (!settings.TryGetValue("archivePath", out m_archivePath))                m_archivePath = FilePath.GetAbsolutePath(FilePath.AddPathSuffix("Archive"));             if (settings.TryGetValue("refreshMetadata", out setting) || settings.TryGetValue("autoRefreshMetadata", out setting))                m_autoRefreshMetadata = setting.ParseBoolean();             if (settings.TryGetValue("badDataMessageInterval", out setting) && ulong.TryParse(setting, out badDataMessageInterval))                m_badDataMessageInterval = badDataMessageInterval;             //if (settings.TryGetValue("useNamespaceReservation", out setting))            //    m_useNamespaceReservation = setting.ParseBoolean();            //else            //    m_useNamespaceReservation = false;             // Initialize metadata file.            m_instanceName = m_instanceName.ToLower();            m_archive.MetadataFile.FileName = Path.Combine(m_archivePath, m_instanceName + "_dbase.dat");            m_archive.MetadataFile.PersistSettings = true;            m_archive.MetadataFile.SettingsCategory = m_instanceName + m_archive.MetadataFile.SettingsCategory;            m_archive.MetadataFile.FileAccessMode = FileAccess.ReadWrite;            m_archive.MetadataFile.Initialize();             // Initialize state file.            m_archive.StateFile.FileName = Path.Combine(m_archivePath, m_instanceName + "_startup.dat");            m_archive.StateFile.PersistSettings = true;            m_archive.StateFile.SettingsCategory = m_instanceName + m_archive.StateFile.SettingsCategory;            m_archive.StateFile.FileAccessMode = FileAccess.ReadWrite;            m_archive.StateFile.Initialize();             // Initialize intercom file.            m_archive.IntercomFile.FileName = Path.Combine(m_archivePath, "scratch.dat");            m_archive.IntercomFile.PersistSettings = true;            m_archive.IntercomFile.SettingsCategory = m_instanceName + m_archive.IntercomFile.SettingsCategory;            m_archive.IntercomFile.FileAccessMode = FileAccess.ReadWrite;            m_archive.IntercomFile.Initialize();             // Initialize data archive file.                       m_archive.FileName = Path.Combine(m_archivePath, m_instanceName + "_archive.d");            m_archive.FileSize = 100;            m_archive.CompressData = false;            m_archive.PersistSettings = true;            m_archive.SettingsCategory = m_instanceName + m_archive.SettingsCategory;             m_archive.RolloverStart += m_archive_RolloverStart;            m_archive.RolloverComplete += m_archive_RolloverComplete;            m_archive.RolloverException += m_archive_RolloverException;             m_archive.DataReadException += m_archive_DataReadException;            m_archive.DataWriteException += m_archive_DataWriteException;             m_archive.OffloadStart += m_archive_OffloadStart;            m_archive.OffloadComplete += m_archive_OffloadComplete;            m_archive.OffloadException += m_archive_OffloadException;             m_archive.OrphanDataReceived += m_archive_OrphanDataReceived;            m_archive.FutureDataReceived += m_archive_FutureDataReceived;            m_archive.OutOfSequenceDataReceived += m_archive_OutOfSequenceDataReceived;             m_archive.Initialize();             // The archive path may have changed based on the configuration file            m_archivePath = FilePath.GetDirectoryName(m_archive.FileName);             // Provide web service support.            m_dataServices = new DataServices();            m_dataServices.AdapterCreated += DataServices_AdapterCreated;            m_dataServices.AdapterLoaded += DataServices_AdapterLoaded;            m_dataServices.AdapterUnloaded += DataServices_AdapterUnloaded;            m_dataServices.AdapterLoadException += AdapterLoader_AdapterLoadException;             // Provide metadata sync support.            m_metadataProviders = new MetadataProviders();            m_metadataProviders.AdapterCreated += MetadataProviders_AdapterCreated;            m_metadataProviders.AdapterLoaded += MetadataProviders_AdapterLoaded;            m_metadataProviders.AdapterUnloaded += MetadataProviders_AdapterUnloaded;            m_metadataProviders.AdapterLoadException += AdapterLoader_AdapterLoadException;             // Provide archive replication support.            m_replicationProviders = new ReplicationProviders();            m_replicationProviders.AdapterCreated += ReplicationProviders_AdapterCreated;            m_replicationProviders.AdapterLoaded += ReplicationProviders_AdapterLoaded;            m_replicationProviders.AdapterUnloaded += ReplicationProviders_AdapterUnloaded;            m_replicationProviders.AdapterLoadException += AdapterLoader_AdapterLoadException;        }         /// <summary>        /// Gets a short one-line status of this <see cref="LocalOutputAdapter"/>.        /// </summary>        /// <param name="maxLength">Maximum length of the status message.</param>        /// <returns>Text of the status message.</returns>        public override string GetShortStatus(int maxLength)        {            return $"Archived {m_archivedMeasurements} measurements locally.".CenterText(maxLength);        }         /// <summary>        /// Releases the unmanaged resources used by this <see cref="LocalOutputAdapter"/> and optionally releases the managed resources.        /// </summary>        /// <param name="disposing">true to release both managed and unmanaged resources; false to release only unmanaged resources.</param>        protected override void Dispose(bool disposing)        {            if (!m_disposed)            {                try                {                    // This will be done regardless of whether the object is finalized or disposed.                    if (disposing)                    {                        // This will be done only when the object is disposed by calling Dispose().                        if ((object)m_dataServices != null)                        {                            m_dataServices.AdapterCreated -= DataServices_AdapterCreated;                            m_dataServices.AdapterLoaded -= DataServices_AdapterLoaded;                            m_dataServices.AdapterUnloaded -= DataServices_AdapterUnloaded;                            m_dataServices.AdapterLoadException -= AdapterLoader_AdapterLoadException;                            m_dataServices.Dispose();                        }                         if ((object)m_metadataProviders != null)                        {                            m_metadataProviders.AdapterCreated -= MetadataProviders_AdapterCreated;                            m_metadataProviders.AdapterLoaded -= MetadataProviders_AdapterLoaded;                            m_metadataProviders.AdapterUnloaded -= MetadataProviders_AdapterUnloaded;                            m_metadataProviders.AdapterLoadException -= AdapterLoader_AdapterLoadException;                            m_metadataProviders.Dispose();                        }                         if ((object)m_replicationProviders != null)                        {                            m_replicationProviders.AdapterCreated -= ReplicationProviders_AdapterCreated;                            m_replicationProviders.AdapterLoaded -= ReplicationProviders_AdapterLoaded;                            m_replicationProviders.AdapterUnloaded -= ReplicationProviders_AdapterUnloaded;                            m_replicationProviders.AdapterLoadException -= AdapterLoader_AdapterLoadException;                            m_replicationProviders.Dispose();                        }                         if ((object)m_archive != null)                        {                            m_archive.RolloverStart -= m_archive_RolloverStart;                            m_archive.RolloverComplete -= m_archive_RolloverComplete;                            m_archive.RolloverException -= m_archive_RolloverException;                             m_archive.DataReadException -= m_archive_DataReadException;                            m_archive.DataWriteException -= m_archive_DataWriteException;                             m_archive.OffloadStart -= m_archive_OffloadStart;

Lines 1–500 of 1396. Load more