//******************************************************************************************************//  DataPublisher.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://  ----------------------------------------------------------------------------------------------------//  08/20/2010 - J. Ritchie Carroll//       Generated original version of source code.//  11/15/2010 - Mehulbhai P Thakkar//       Fixed bug when DataSubscriber tries to resubscribe by setting subscriber. Initialized manually //       in ReceiveClientDataComplete event handler.//  12/02/2010 - J. Ritchie Carroll//       Fixed an issue for when DataSubcriber dynamically resubscribes with a different//       synchronization method (e.g., going from unsynchronized to synchronized)//  05/26/2011 - J. Ritchie Carroll//       Implemented subscriber authentication model.//  02/07/2012 - Mehulbhai Thakkar//       Modified m_metadataTables to include filter expression and made the list ";" seperated.////****************************************************************************************************** using GSF.Communication;//using GSF.Data;//using GSF.Security.Cryptography;using GSF.TimeSeries.Adapters;//using GSF.TimeSeries.Statistics;using System;using System.Collections.Concurrent;using System.Collections.Generic;using System.ComponentModel;using System.Data;using System.Globalization;using System.IO;using System.IO.Compression;using System.Linq;using System.Net;using System.Net.Sockets;using System.Text;using System.Text.RegularExpressions;using System.Threading;using System.Xml; namespace GSF.TimeSeries.Transport{    #region [ Enumerations ]     /// <summary>    /// Server commands received by <see cref="DataPublisher"/> and sent by <see cref="DataSubscriber"/>.    /// </summary>    /// <remarks>    /// Solicited server commands will receive a <see cref="ServerResponse.Succeeded"/> or <see cref="ServerResponse.Failed"/>    /// response code along with an associated success or failure message. Message type for successful responses will be based    /// on server command - for example, server response for a successful MetaDataRefresh command will return a serialized    /// <see cref="DataSet"/> of the available server metadata. Message type for failed responses will always be a string of    /// text respresenting the error message.    /// </remarks>    public enum ServerCommand : byte    {        /// <summary>        /// Authenticate command.        /// </summary>        /// <remarks>        /// Requests that server authenticate client with the following encrypted authentication packet.        /// Successful return message type will be string indicating client ID was validated.        /// Client should not attempt further steps if authentication fails, user will need to take action.        /// Client should send MetaDataRefresh and Subscribe commands upon successful authentication.        /// If server is setup to not require authentication, authentication step can be skipped - however        /// it will be up to the implementation to know if authentication is required - the server cannot be        /// queried to determine if authentication is required.        /// It is expected that use of this protocol between gateways over a shared network (i.e., the lower        /// security zone) will require authentication.        /// It is expected that use of this protocol between gateway and other systems (e.g., a PDC) in a        /// private network (i.e., the higher security zone) can be setup to not require authentication.        /// </remarks>        Authenticate = 0x00,        /// <summary>        /// Meta data refresh command.        /// </summary>        /// <remarks>        /// Requests that server send an updated set of metadata so client can refresh its point list.        /// Successful return message type will be a <see cref="DataSet"/> containing server device and measurement metadata.        /// Received device list should be defined as children of the "parent" server device connection similar to the way        /// PMUs are defined as children of a parent PDC device connection.        /// Devices and measurements contain unique Guids that should be used to key metadata updates in local repository.        /// </remarks>        MetaDataRefresh = 0x01,        /// <summary>        /// Subscribe command.        /// </summary>        /// <remarks>        /// Requests a subscription of streaming data from server based on connection string that follows.        /// It will not be necessary to stop an existing subscription before requesting a new one.        /// Successful return message type will be string indicating total number of allowed points.        /// Client should wait for UpdateSignalIndexCache and UpdateBaseTime response codes before attempting        /// to parse data when using the compact measurement format.        /// </remarks>        Subscribe = 0x02,        /// <summary>        /// Unsubscribe command.        /// </summary>        /// <remarks>        /// Requests that server stop sending streaming data to the client and cancel the current subscription.        /// </remarks>        Unsubscribe = 0x03,        /// <summary>        /// Rotate cipher keys.        /// </summary>        /// <remarks>        /// Manually requests that server send a new set of cipher keys for data packet encryption.        /// </remarks>        RotateCipherKeys = 0x04,        /// <summary>        /// Update processing interval.        /// </summary>        /// <remarks>        /// Manually requests server to update the processing interval with the following specified value.        /// </remarks>        UpdateProcessingInterval = 0x05,        /// <summary>        /// Define operational modes for subscriber connection.        /// </summary>        /// <remarks>        /// As soon as connection is established, requests that server set operational        /// modes that affect how the subscriber and publisher will communicate.        /// </remarks>        DefineOperationalModes = 0x06    }     /// <summary>    /// Server responses sent by <see cref="DataPublisher"/> and received by <see cref="DataSubscriber"/>.    /// </summary>    public enum ServerResponse : byte    {        // Although the server commands and responses will be on two different paths, the response enumeration values        // are defined as distinct from the command values to make it easier to identify codes from a wire analysis.         /// <summary>        /// Command succeeded response.        /// </summary>        /// <remarks>        /// Informs client that its solicited server command succeeded, original command and success message follow.        /// </remarks>        Succeeded = 0x80,        /// <summary>        /// Command failed response.        /// </summary>        /// <remarks>        /// Informs client that its solicited server command failed, original command and failure message follow.        /// </remarks>        Failed = 0x81,        /// <summary>        /// Data packet response.        /// </summary>        /// <remarks>        /// Unsolicited response informs client that a data packet follows.        /// </remarks>        DataPacket = 0x82,        /// <summary>        /// Update signal index cache response.        /// </summary>        /// <remarks>        /// Unsolicited response requests that client update its runtime signal index cache with the one that follows.        /// </remarks>        UpdateSignalIndexCache = 0x83,        /// <summary>        /// Update runtime base-timestamp offsets response.        /// </summary>        /// <remarks>        /// Unsolicited response requests that client update its runtime base-timestamp offsets with those that follow.        /// </remarks>        UpdateBaseTimes = 0x84,        /// <summary>        /// Update runtime cipher keys response.        /// </summary>        /// <remarks>        /// Response, solicited or unsolicited, requests that client update its runtime data cipher keys with those that follow.        /// </remarks>        UpdateCipherKeys = 0x85,        /// <summary>        /// Data start time response packet.        /// </summary>        /// <remarks>        /// Unsolicited response provides the start time of data being processed from the first measurement.        /// </remarks>        DataStartTime = 0x86,        /// <summary>        /// Processing complete notification.        /// </summary>        /// <remarks>        /// Unsolicited response provides notification that input processing has completed, typically via temporal constraint.        /// </remarks>        ProcessingComplete = 0x87,        /// <summary>        /// No operation keep-alive ping.        /// </summary>        /// <remarks>        /// The command channel can remain quiet for some time, this command allows a period test of client connectivity.        /// </remarks>        NoOP = 0xFF    }     /// <summary>    /// <see cref="DataPublisher"/> data packet flags.    /// </summary>    [Flags()]    public enum DataPacketFlags : byte    {        /// <summary>        /// Determines if data packet is synchronized.        /// </summary>        /// <remarks>        /// Bit set = synchronized, bit clear = unsynchronized.        /// </remarks>        Synchronized = (byte)Bits.Bit00,        /// <summary>        /// Determines if serialized measurement is compact.        /// </summary>        /// <remarks>        /// Bit set = compact, bit clear = full fidelity.        /// </remarks>        Compact = (byte)Bits.Bit01,        /// <summary>        /// Detemines which cipher index to use when encrypting data packet.        /// </summary>        /// <remarks>        /// Bit set = use odd cipher index (i.e., 1), bit clear = use even cipher index (i.e., 0).        /// </remarks>        CipherIndex = (byte)Bits.Bit02,        /// <summary>        /// No flags set.        /// </summary>        /// <remarks>        /// This would represent unsynchronized, full fidelity measurement data packets.        /// </remarks>        NoFlags = (Byte)Bits.Nil    }     /// <summary>    /// Operational modes that affect how <see cref="DataPublisher"/> and <see cref="DataSubscriber"/> communicate.    /// </summary>    /// <remarks>    /// Operational modes only apply to fundamental protocol control.    /// </remarks>    [Flags()]    public enum OperationalModes : uint    {        /// <summary>        /// Mask to get version number of protocol.        /// </summary>        /// <remarks>        /// Version number is currently set to 0.        /// </remarks>        VersionMask = (uint)(Bits.Bit04 | Bits.Bit03 | Bits.Bit02 | Bits.Bit01 | Bits.Bit00),        /// <summary>        /// Mask to get mode of compression.        /// </summary>        /// <remarks>        /// Compression mode is currently not supported.        /// These bits are simply reserved for different compression modes.        /// </remarks>        CompressionModeMask = (uint)(Bits.Bit07 | Bits.Bit06 | Bits.Bit05),        /// <summary>        /// Mask to get character encoding used when exchanging messages between publisher and subscriber.        /// </summary>        /// <remarks>        /// 00 = UTF-16 little endian<br/>        /// 01 = UTF-16 big endian<br/>        /// 10 = UTF-8<br/>        /// 11 = ANSI        /// </remarks>        EncodingMask = (uint)(Bits.Bit09 | Bits.Bit08),        /// <summary>        /// Determines type of serialization to use when exchanging signal index cache and metadata.        /// </summary>        /// <remarks>        /// Bit set = common serialization format, bit clear = .NET serialization format        /// </remarks>        UseCommonSerializationFormat = (uint)Bits.Bit24,        /// <summary>        /// Determines whether the signal index cache is compressed when exchanging between publisher and subscriber.        /// </summary>        /// <remarks>        /// Bit set = compress, bit clear = no compression        /// </remarks>        CompressSignalIndexCache = (uint)Bits.Bit30,        /// <summary>        /// Determines whether is compressed when exchanging between publisher and subscriber.        /// </summary>        /// <remarks>        /// Bit set = compress, bit clear = no compression        /// </remarks>        CompressMetadata = (uint)Bits.Bit31,        /// <summary>        /// No flags set.        /// </summary>        /// <remarks>        /// This would represent protocol version 0,        /// UTF-16 little endian character encoding,        /// .NET serialization and no compression.        /// </remarks>        NoFlags = (uint)Bits.Nil    }     /// <summary>    /// Enumeration for character encodings supported by the Gateway Exchange Protocol.    /// </summary>    public enum OperationalEncoding : uint    {        /// <summary>        /// UTF-16 little endian        /// </summary>        Unicode = (uint)Bits.Nil,        /// <summary>        /// UTF-16 bit endian        /// </summary>        BigEndianUnicode = (uint)Bits.Bit08,        /// <summary>        /// UTF-8        /// </summary>        UTF8 = (uint)Bits.Bit09,        /// <summary>        /// ANSI        /// </summary>        ANSI = (uint)(Bits.Bit09 | Bits.Bit08)    }     /// <summary>    /// Enumeration for compression modes supported by the Gateway Exchange Protocol.    /// </summary>    public enum GatewayCompressionMode : uint    {        /// <summary>        /// GZip compression        /// </summary>        GZip = (uint)Bits.Bit05,        /// <summary>        /// No compression        /// </summary>        None = (uint)Bits.Nil    }     #endregion     /// <summary>    /// Represents a data publishing server that allows multiple connections for data subscriptions.    /// </summary>    [Description("DataPublisher: server component that allows gateway-style subscription connections.")]    [EditorBrowsable(EditorBrowsableState.Always)]    public class DataPublisher : ActionAdapterCollection    {        #region [ Members ]         // Nested Types         private class LatestMeasurementCache : FacileActionAdapterBase        {            public LatestMeasurementCache(string connectionString)            {                ConnectionString = connectionString;            }             public override string Name            {                get                {                    return "LatestMeasurementCache";                }                set                {                    base.Name = value;                }            }             public override bool SupportsTemporalProcessing            {                get                {                    return false;                }            }             public override string GetShortStatus(int maxLength)            {                return "LatestMeasurementCache happily exists. :)";            }        }         // Events         /// <summary>        /// Indicates that a new client has connected to the publisher.        /// </summary>        /// <remarks>        /// <see cref="EventArgs{T1, T2, T3}.Argument1"/> is the <see cref="Guid"/> based subscriber ID.<br/>        /// <see cref="EventArgs{T1, T2, T3}.Argument2"/> is the connection identification (e.g., IP and DNS name, if available).<br/>        /// <see cref="EventArgs{T1, T2, T3}.Argument3"/> is the subscriber information as reported by the client.        /// </remarks>        public event EventHandler<EventArgs<Guid, string, string>> ClientConnected;         /// <summary>        /// Indicates to the host 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 procesing.        /// </remarks>        public event EventHandler ProcessingComplete;         // Constants         /// <summary>        /// Default value for <see cref="RequireAuthentication"/>.        /// </summary>        public const bool DefaultRequireAuthentication = false;         /// <summary>        /// Default value for <see cref="EncryptPayload"/>.        /// </summary>        public const bool DefaultEncryptPayload = false;         /// <summary>        /// Default value for <see cref="SharedDatabase"/>.        /// </summary>        public const bool DefaultSharedDatabase = false;         /// <summary>        /// Default value for <see cref="AllowSynchronizedSubscription"/>.        /// </summary>        public const bool DefaultAllowSynchronizedSubscription = true;         /// <summary>        /// Default value for <see cref="UseBaseTimeOffsets"/>.        /// </summary>        public const bool DefaultUseBaseTimeOffsets = false;         /// <summary>        /// Default value for <see cref="CipherKeyRotationPeriod"/>.        /// </summary>        public const double DefaultCipherKeyRotationPeriod = 60000.0D;         // Maximum packet size before software fragmentation of UDP payload        internal const int MaxPacketSize = ushort.MaxValue / 2;         // Length of random salt prefix        internal const int CipherSaltLength = 8;         // Fields        private ManualResetEvent m_initializeWaitHandle;        private TcpServer m_commandChannel;        private ConcurrentDictionary<Guid, ClientConnection> m_clientConnections;        private ConcurrentDictionary<Guid, IServer> m_clientPublicationChannels;        private ConcurrentDictionary<MeasurementKey, Guid> m_signalIDCache;        private System.Timers.Timer m_commandChannelRestartTimer;        private System.Timers.Timer m_cipherKeyRotationTimer;        private RoutingTables m_routingTables;        private IAdapterCollection m_parent;        private string m_metadataTables;        private bool m_requireAuthentication;        private bool m_encryptPayload;        private bool m_sharedDatabase;        private bool m_allowSynchronizedSubscription;        private bool m_useBaseTimeOffsets;        private bool m_disposed;         #endregion         #region [ Constructors ]         /// <summary>        /// Creates a new <see cref="DataPublisher"/>.        /// </summary>        public DataPublisher()            : this(null)        {            // When collection is spawned as an adapter, it needs a parameterless constructor        }         /// <summary>        /// Creates a new <see cref="DataPublisher"/>.        /// </summary>        /// <param name="waitHandles">Wait handle dictionary.</param>        public DataPublisher(ConcurrentDictionary<string, AutoResetEvent> waitHandles)            : base(waitHandles)        {            base.Name = "Data Publisher Collection";            base.DataMember = "[internal]";             m_initializeWaitHandle = new ManualResetEvent(false);

Lines 1–500 of 2623. Load more