//******************************************************************************************************//  DataSubscriber.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://  ----------------------------------------------------------------------------------------------------//  08/20/2010 - J. Ritchie Carroll//       Generated original version of source code.//  02/07/2012 - Mehulbhai Thakkar//       Modified SynchronizeMetadata to filter devices by original source and modified insert query//       to populate OriginalSource value. Added to flag to optionally avoid meta-data synchronization.//  12/20/2012 - Starlynn Danyelle Gilliam//       Modified Header.////****************************************************************************************************** using System;using System.Collections.Generic;using System.ComponentModel;using System.Data;using System.IO;using System.IO.Compression;using System.Linq;using System.Net.Security;using System.Net.Sockets;using System.Security.Cryptography.X509Certificates;using System.Text;using System.Text.RegularExpressions;using System.Threading;using System.Xml;using GSF.Collections;using GSF.Communication;using GSF.Configuration;using GSF.Data;using GSF.Diagnostics;using GSF.IO;using GSF.Net.Security;using GSF.Reflection;using GSF.Security.Cryptography;using GSF.Threading;using GSF.TimeSeries.Adapters;using GSF.TimeSeries.Data;using GSF.TimeSeries.Statistics;using GSF.TimeSeries.Transport.TSSC;using GSF.Units;using Random = GSF.Security.Cryptography.Random;using TcpClient = GSF.Communication.TcpClient;using UdpClient = GSF.Communication.UdpClient;using System.Globalization; #pragma warning disable 672 namespace GSF.TimeSeries.Transport{    /// <summary>    /// Represents a data subscribing client that will connect to a data publisher for a data subscription.    /// </summary>    [Description("DataSubscriber: client that subscribes to a publishing server for a streaming data.")]    [EditorBrowsable(EditorBrowsableState.Advanced)] // Normally defined as an input device protocol    public class DataSubscriber : InputAdapterBase    {        #region [ Members ]         // Nested Types         // Local measurement concentrator        private class LocalConcentrator : ConcentratorBase        {            #region [ Members ]             // Fields            private DataSubscriber m_parent;            private bool m_disposed;             #endregion             #region [ Constructors ]             /// <summary>            /// Creates a new local concentrator.            /// </summary>            public LocalConcentrator(DataSubscriber parent)            {                m_parent = parent;            }             #endregion             #region [ Methods ]             /// <summary>            /// Releases the unmanaged resources used by the <see cref="LocalConcentrator"/> object 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                    {                        if (disposing)                            m_parent = null;                    }                    finally                    {                        m_disposed = true;          // Prevent duplicate dispose.                        base.Dispose(disposing);    // Call base class Dispose().                    }                }            }             /// <summary>            /// Publish <see cref="IFrame"/> of time-aligned collection of <see cref="IMeasurement"/> values that arrived within the            /// concentrator's defined <see cref="ConcentratorBase.LagTime"/>.            /// </summary>            /// <param name="frame"><see cref="IFrame"/> of measurements with the same timestamp that arrived within <see cref="ConcentratorBase.LagTime"/> that are ready for processing.</param>            /// <param name="index">Index of <see cref="IFrame"/> within a second ranging from zero to <c><see cref="ConcentratorBase.FramesPerSecond"/> - 1</c>.</param>            protected override void PublishFrame(IFrame frame, int index)            {                // Publish locally sorted measurements                if ((object)m_parent != null)                    m_parent.OnNewMeasurements(frame.Measurements.Values);            }             #endregion        }         private class SubscribedDevice : IDevice, IDisposable        {            #region [ Members ]             // Fields            private readonly string m_name;             private Guid m_statusFlagsID;            private Guid m_frequencyID;            private Guid m_deltaFrequencyID;             private long m_dataQualityErrors;            private long m_timeQualityErrors;            private long m_deviceErrors;            private long m_measurementsReceived;            private double m_measurementsExpected;            private long m_measurementsWithError;            private long m_measurementsDefined;             private bool m_disposed;             #endregion             #region [ Constructors ]             public SubscribedDevice(string name)            {                if ((object)name == null)                    throw new ArgumentNullException(nameof(name));                 m_name = name;                StatisticsEngine.Register(this, name, "Device", "PMU");            }             /// <summary>            /// Releases the unmanaged resources before the <see cref="SubscribedDevice"/> object is reclaimed by <see cref="GC"/>.            /// </summary>            ~SubscribedDevice()            {                Unregister();            }             #endregion             #region [ Properties ]             public string Name => m_name;             public Guid StatusFlagsID            {                get                {                    return m_statusFlagsID;                }                set                {                    m_statusFlagsID = value;                }            }             public Guid FrequencyID            {                get                {                    return m_frequencyID;                }                set                {                    m_frequencyID = value;                }            }             public Guid DeltaFrequencyID            {                get                {                    return m_deltaFrequencyID;                }                set                {                    m_deltaFrequencyID = value;                }            }             public long DataQualityErrors            {                get                {                    return Interlocked.Read(ref m_dataQualityErrors);                }                set                {                    Interlocked.Exchange(ref m_dataQualityErrors, value);                }            }             public long TimeQualityErrors            {                get                {                    return Interlocked.Read(ref m_timeQualityErrors);                }                set                {                    Interlocked.Exchange(ref m_timeQualityErrors, value);                }            }             public long DeviceErrors            {                get                {                    return Interlocked.Read(ref m_deviceErrors);                }                set                {                    Interlocked.Exchange(ref m_deviceErrors, value);                }            }             public long MeasurementsReceived            {                get                {                    return Interlocked.Read(ref m_measurementsReceived);                }                set                {                    Interlocked.Exchange(ref m_measurementsReceived, value);                }            }             public long MeasurementsExpected            {                get                {                    return (long)Interlocked.CompareExchange(ref m_measurementsExpected, 0.0D, 0.0D);                }                set                {                    Interlocked.Exchange(ref m_measurementsExpected, value);                }            }             public long MeasurementsWithError            {                get                {                    return Interlocked.Read(ref m_measurementsWithError);                }                set                {                    Interlocked.Exchange(ref m_measurementsWithError, value);                }            }             public long MeasurementsDefined            {                get                {                    return Interlocked.Read(ref m_measurementsDefined);                }                set                {                    Interlocked.Exchange(ref m_measurementsDefined, value);                }            }             #endregion             #region [ Methods ]             public override bool Equals(object obj)            {                SubscribedDevice subscribedDevice = obj as SubscribedDevice;                return (object)subscribedDevice != null && m_name.Equals(subscribedDevice.m_name);            }             public override int GetHashCode()            {                return m_name.GetHashCode();            }             /// <summary>            /// Releases all the resources used by the <see cref="SubscribedDevice"/> object.            /// </summary>            public void Dispose()            {                Unregister();                GC.SuppressFinalize(this);            }             private void Unregister()            {                if (!m_disposed)                {                    try                    {                        StatisticsEngine.Unregister(this);                    }                    finally                    {                        m_disposed = true;  // Prevent duplicate dispose.                    }                }            }             #endregion        }         /// <summary>        /// EventArgs implementation for handling user commands.        /// </summary>        public class UserCommandArgs : EventArgs        {            /// <summary>            /// Creates a new instance of the <see cref="UserCommandArgs"/> class.            /// </summary>            /// <param name="command">The code for the user command.</param>            /// <param name="response">The code for the server's response.</param>            /// <param name="buffer">Buffer containing the message from the server.</param>            /// <param name="startIndex">Index into the buffer used to skip the header.</param>            /// <param name="length">The length of the message in the buffer, including the header.</param>            public UserCommandArgs(ServerCommand command, ServerResponse response, byte[] buffer, int startIndex, int length)            {                Command = command;                Response = response;                Buffer = buffer;                StartIndex = startIndex;                Length = length;            }             /// <summary>            /// Gets the code for the user command.            /// </summary>            public ServerCommand Command { get; }             /// <summary>            /// Gets the code for the server's response.            /// </summary>            public ServerResponse Response { get; }             /// <summary>            /// Gets the buffer containing the message from the server.            /// </summary>            public byte[] Buffer { get; }             /// <summary>            /// Gets the index into the buffer used to skip the header.            /// </summary>            public int StartIndex { get; }             /// <summary>            /// Gets the length of the message in the buffer, including the header.            /// </summary>            public int Length { get; }        }         // Constants         /// <summary>        /// Defines default value for <see cref="DataSubscriber.OperationalModes"/>.        /// </summary>        public const OperationalModes DefaultOperationalModes = OperationalModes.CompressMetadata | OperationalModes.CompressSignalIndexCache | OperationalModes.ReceiveInternalMetadata | OperationalModes.UseCommonSerializationFormat;         /// <summary>        /// Defines the default value for the <see cref="MetadataSynchronizationTimeout"/> property.        /// </summary>        public const int DefaultMetadataSynchronizationTimeout = 0;         /// <summary>        /// Defines the default value for the <see cref="UseTransactionForMetadata"/> property.        /// </summary>        public const bool DefaultUseTransactionForMetadata = true;         /// <summary>        /// Default value for <see cref="LoggingPath"/>.        /// </summary>        public const string DefaultLoggingPath = "ConfigurationCache";         /// <summary>        /// Specifies the default value for the <see cref="AllowedParsingExceptions"/> property.        /// </summary>        public const int DefaultAllowedParsingExceptions = 10;         /// <summary>        /// Specifies the default value for the <see cref="ParsingExceptionWindow"/> property.        /// </summary>        public const long DefaultParsingExceptionWindow = 50000000L; // 5 seconds         private const int EvenKey = 0;      // Even key/IV index        private const int OddKey = 1;       // Odd key/IV index        private const int KeyIndex = 0;     // Index of cipher key component in keyIV array        private const int IVIndex = 1;      // Index of initialization vector component in keyIV array         private const long MissingCacheWarningInterval = 20000000;         // Events         /// <summary>        /// Occurs when client connection to the data publication server is established.        /// </summary>        public event EventHandler ConnectionEstablished;         /// <summary>        /// Occurs when client connection to the data publication server is terminated.        /// </summary>        public event EventHandler ConnectionTerminated;         /// <summary>        /// Occurs when client connection to the data publication server has successfully authenticated.        /// </summary>        public event EventHandler ConnectionAuthenticated;         /// <summary>        /// Occurs when client receives response from the server.        /// </summary>        public event EventHandler<EventArgs<ServerResponse, ServerCommand>> ReceivedServerResponse;         /// <summary>        /// Occurs when client receives message from the server in response to a user command.        /// </summary>        public event EventHandler<UserCommandArgs> ReceivedUserCommandResponse;         /// <summary>        /// Occurs when client receives requested meta-data transmitted by data publication server.        /// </summary>        public event EventHandler<EventArgs<DataSet>> MetaDataReceived;         /// <summary>        /// Occurs when first measurement is transmitted by data publication server.        /// </summary>        public event EventHandler<EventArgs<Ticks>> DataStartTime;         /// <summary>        /// Indicates that processing for an input adapter (via temporal session) has completed.        /// </summary>        /// <remarks>        /// 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.        /// </remarks>        public new event EventHandler<EventArgs<string>> ProcessingComplete;         /// <summary>        /// Occurs when a notification has been received from the <see cref="DataPublisher"/>.        /// </summary>        public event EventHandler<EventArgs<string>> NotificationReceived;         /// <summary>        /// Occurs when the server has sent a notification that its configuration has changed, this        /// can allow subscriber to request updated meta-data if desired.        /// </summary>        public event EventHandler ServerConfigurationChanged;         /// <summary>        /// Occurs when number of parsing exceptions exceed <see cref="AllowedParsingExceptions"/> during <see cref="ParsingExceptionWindow"/>.        /// </summary>        public event EventHandler ExceededParsingExceptionThreshold;         // Fields        private volatile Dictionary<Guid, DeviceStatisticsHelper<SubscribedDevice>> m_subscribedDevicesLookup;

Lines 1–500 of 5369. Load more