//******************************************************************************************************//  UnsynchronizedClientSubscription.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://  ----------------------------------------------------------------------------------------------------//  06/24/2011 - Ritchie//       Generated original version of source code.//  12/20/2012 - Starlynn Danyelle Gilliam//       Modified Header.////****************************************************************************************************** using GSF.Diagnostics;using GSF.IO;using GSF.Parsing;using GSF.Threading;using GSF.TimeSeries.Adapters;using GSF.TimeSeries.Transport.TSSC;using System;using System.Collections.Generic;using System.IO;using System.Linq;using System.Text; // ReSharper disable PossibleMultipleEnumerationnamespace GSF.TimeSeries.Transport{    /// <summary>    /// Represents an unsynchronized client subscription to the <see cref="DataPublisher" />.    /// </summary>    internal class UnsynchronizedClientSubscription : FacileActionAdapterBase, IClientSubscription    {        #region [ Members ]         // Events         /// <summary>        /// Indicates that a buffer block needed to be retransmitted because        /// it was previously sent, but no confirmation was received.        /// </summary>        public event EventHandler BufferBlockRetransmission;         /// <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 processing.        /// </remarks>        public event EventHandler<EventArgs<IClientSubscription, EventArgs>> ProcessingComplete;         // Fields        private readonly SignalIndexCache m_signalIndexCache;        private readonly Guid m_clientID;        private readonly Guid m_subscriberID;        private DataPublisher m_parent;        private volatile byte m_compressionStrength;        private volatile bool m_usePayloadCompression;        private volatile bool m_useCompactMeasurementFormat;        private readonly CompressionModes m_compressionModes;        private bool m_resetTsscEncoder;        private TsscEncoder m_tsscEncoder;        private readonly object m_tsscSyncLock;        private byte[] m_tsscWorkingBuffer;        private ushort m_tsscSequenceNumber;        private long m_lastPublishTime;        private string m_requestedInputFilter;        private double m_publishInterval;        private bool m_includeTime;        private bool m_useMillisecondResolution;        private bool m_isNaNFiltered;        private volatile long[] m_baseTimeOffsets;        private volatile int m_timeIndex;        private SharedTimer m_baseTimeRotationTimer;        private volatile bool m_startTimeSent;        private IaonSession m_iaonSession;         private readonly List<byte[]> m_bufferBlockCache;        private readonly object m_bufferBlockCacheLock;        private uint m_bufferBlockSequenceNumber;        private uint m_expectedBufferBlockConfirmationNumber;        private SharedTimer m_bufferBlockRetransmissionTimer;        private double m_bufferBlockRetransmissionTimeout;         private bool m_disposed;         #endregion         #region [ Constructors ]         /// <summary>        /// Creates a new <see cref="UnsynchronizedClientSubscription"/>.        /// </summary>        /// <param name="parent">Reference to parent.</param>        /// <param name="clientID"><see cref="Guid"/> based client connection ID.</param>        /// <param name="subscriberID"><see cref="Guid"/> based subscriber ID.</param>        /// <param name="compressionModes"><see cref="CompressionModes"/> requested by client.</param>        public UnsynchronizedClientSubscription(DataPublisher parent, Guid clientID, Guid subscriberID, CompressionModes compressionModes)        {            m_parent = parent;            m_clientID = clientID;            m_subscriberID = subscriberID;            m_compressionModes = compressionModes;             m_signalIndexCache = new SignalIndexCache();            m_signalIndexCache.SubscriberID = subscriberID;             m_bufferBlockCache = new List<byte[]>();            m_bufferBlockCacheLock = new object();             m_tsscSyncLock = new object();        }         #endregion         #region [ Properties ]         /// <summary>        /// Gets name of the action adapter.        /// </summary>        public override string Name        {            get            {                return base.Name;            }             set            {                base.Name = value;                Log.InitialStackMessages = Log.InitialStackMessages.Union("AdapterName", GetType().Name, "HostName", value);            }        }         /// <summary>        /// Gets the <see cref="Guid"/> client TCP connection identifier of this <see cref="UnsynchronizedClientSubscription"/>.        /// </summary>        public Guid ClientID => m_clientID;         /// <summary>        /// Gets the <see cref="Guid"/> based subscriber ID of this <see cref="UnsynchronizedClientSubscription"/>.        /// </summary>        public Guid SubscriberID => m_subscriberID;         /// <summary>        /// Gets the current signal index cache of this <see cref="UnsynchronizedClientSubscription"/>.        /// </summary>        public SignalIndexCache SignalIndexCache => m_signalIndexCache;         /// <summary>        /// Gets the input filter requested by the subscriber when establishing this <see cref="IClientSubscription"/>.        /// </summary>        public string RequestedInputFilter => m_requestedInputFilter;         /// <summary>        /// Gets or sets flag that determines if payload compression should be enabled in data packets of this <see cref="UnsynchronizedClientSubscription"/>.        /// </summary>        public bool UsePayloadCompression        {            get            {                return m_usePayloadCompression;            }            set            {                m_usePayloadCompression = value;                 if (m_usePayloadCompression)                    m_useCompactMeasurementFormat = true;            }        }         /// <summary>        /// Gets or sets the compression strength value to use when <see cref="UsePayloadCompression"/> is <c>true</c> for this <see cref="UnsynchronizedClientSubscription"/>.        /// </summary>        public int CompressionStrength        {            get            {                return m_compressionStrength;            }            set            {                if (value < 0)                    value = 0;                 if (value > 31)                    value = 31;                 m_compressionStrength = (byte)value;            }        }         /// <summary>        /// Gets or sets flag that determines if the compact measurement format should be used in data packets of this <see cref="UnsynchronizedClientSubscription"/>.        /// </summary>        public bool UseCompactMeasurementFormat        {            get            {                return m_useCompactMeasurementFormat;            }            set            {                m_useCompactMeasurementFormat = value || m_usePayloadCompression;            }        }         /// <summary>        /// Gets size of timestamp in bytes.        /// </summary>        public int TimestampSize        {            get            {                if (!m_useCompactMeasurementFormat)                    return 8;                 if (!m_includeTime)                    return 0;                 if (!m_parent.UseBaseTimeOffsets)                    return 8;                 return !m_useMillisecondResolution ? 4 : 2;            }        }         /// <summary>        /// Gets or sets the desired processing interval, in milliseconds, for the 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, over which 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            {                base.ProcessingInterval = value;                 // Update processing interval in private temporal session, if defined                if ((object)m_iaonSession != null && m_iaonSession.AllAdapters != null)                    m_iaonSession.AllAdapters.ProcessingInterval = value;            }        }         /// <summary>        /// Gets or sets primary keys of input measurements the <see cref="UnsynchronizedClientSubscription"/> expects, if any.        /// </summary>        /// <remarks>        /// We override method so assignment can be synchronized such that dynamic updates won't interfere        /// with filtering in <see cref="QueueMeasurementsForProcessing"/>.        /// </remarks>        public override MeasurementKey[] InputMeasurementKeys        {            get            {                return base.InputMeasurementKeys;            }            set            {                lock (this)                {                    // Update signal index cache unless "detaching" from real-time                    if ((object)value != null && !(value.Length == 1 && value[0] == MeasurementKey.Undefined))                    {                        m_parent.UpdateSignalIndexCache(m_clientID, m_signalIndexCache, value);                         if ((object)DataSource != null && (object)m_signalIndexCache != null)                            value = ParseInputMeasurementKeys(DataSource, false, string.Join("; ", m_signalIndexCache.AuthorizedSignalIDs));                    }                     base.InputMeasurementKeys = value;                }            }        }         /// <summary>        /// Gets the flag indicating if this adapter supports temporal processing.        /// </summary>        /// <remarks>        /// Although this adapter provisions support for temporal processing by proxying historical data to a remote sink, the adapter        /// does not need to be automatically engaged within an actual temporal <see cref="IaonSession"/>, therefore this method returns        /// <c>false</c> to make sure the adapter doesn't get automatically instantiated within a temporal session.        /// </remarks>        public override bool SupportsTemporalProcessing => false;         /// <summary>        /// Gets a formatted message describing the status of this <see cref="UnsynchronizedClientSubscription"/>.        /// </summary>        public override string Status        {            get            {                StringBuilder status = new StringBuilder();                ClientConnection connection;                 if (m_parent.ClientConnections.TryGetValue(m_clientID, out connection))                {                    status.Append(connection.Status);                    status.AppendLine();                }                 status.Append(base.Status);                 if ((object)m_iaonSession != null)                    status.Append(m_iaonSession.Status);                 return status.ToString();            }        }         /// <summary>        /// Gets the status of the active temporal session, if any.        /// </summary>        public string TemporalSessionStatus        {            get            {                if ((object)m_iaonSession == null)                    return null;                 return m_iaonSession.Status;            }        }         #endregion         #region [ Methods ]         /// <summary>        /// Releases the unmanaged resources used by the <see cref="UnsynchronizedClientSubscription"/> 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;                         // Dispose base time rotation timer                        if ((object)m_baseTimeRotationTimer != null)                        {                            m_baseTimeRotationTimer.Dispose();                            m_baseTimeRotationTimer = null;                        }                         // Dispose Iaon session                        this.DisposeTemporalSession(ref m_iaonSession);                    }                }                finally                {                    m_disposed = true;          // Prevent duplicate dispose.                    base.Dispose(disposing);    // Call base class Dispose().                }            }        }         /// <summary>        /// Initializes <see cref="UnsynchronizedClientSubscription"/>.        /// </summary>        public override void Initialize()        {            MeasurementKey[] inputMeasurementKeys;            string setting;             if (Settings.TryGetValue("inputMeasurementKeys", out setting))            {                // IMPORTANT: The allowSelect argument of ParseInputMeasurementKeys must be null                //            in order to prevent SQL injection via the subscription filter expression                inputMeasurementKeys = ParseInputMeasurementKeys(DataSource, false, setting);                m_requestedInputFilter = setting;                 // IMPORTANT: We need to remove the setting before calling base.Initialize()                //            or else we will still be subject to SQL injection                Settings.Remove("inputMeasurementKeys");            }            else            {                inputMeasurementKeys = new MeasurementKey[0];                m_requestedInputFilter = null;            }             base.Initialize();             // Set the InputMeasurementKeys property after calling base.Initialize()            // so that the base class does not overwrite our setting            InputMeasurementKeys = inputMeasurementKeys;             if (!Settings.TryGetValue("publishInterval", out setting) || !double.TryParse(setting, out m_publishInterval))                m_publishInterval = -1;             if (Settings.TryGetValue("includeTime", out setting))                m_includeTime = setting.ParseBoolean();            else                m_includeTime = true;             if (Settings.TryGetValue("useMillisecondResolution", out setting))                m_useMillisecondResolution = setting.ParseBoolean();            else                m_useMillisecondResolution = false;             if (Settings.TryGetValue("requestNaNValueFilter", out setting))                m_isNaNFiltered = m_parent.AllowNaNValueFilter && setting.ParseBoolean();            else                m_isNaNFiltered = false;             if (Settings.TryGetValue("bufferBlockRetransmissionTimeout", out setting))                m_bufferBlockRetransmissionTimeout = double.Parse(setting);            else                m_bufferBlockRetransmissionTimeout = 5.0D;             if (m_parent.UseBaseTimeOffsets && m_includeTime)            {                m_baseTimeRotationTimer = Common.TimerScheduler.CreateTimer(m_useMillisecondResolution ? 60000 : 420000);                m_baseTimeRotationTimer.AutoReset = true;                m_baseTimeRotationTimer.Elapsed += BaseTimeRotationTimer_Elapsed;            }             m_bufferBlockRetransmissionTimer = Common.TimerScheduler.CreateTimer((int)(m_bufferBlockRetransmissionTimeout * 1000.0D));            m_bufferBlockRetransmissionTimer.AutoReset = false;            m_bufferBlockRetransmissionTimer.Elapsed += BufferBlockRetransmissionTimer_Elapsed;             // Handle temporal session initialization            if (this.TemporalConstraintIsDefined())                m_iaonSession = this.CreateTemporalSession();        }         /// <summary>        /// Starts the <see cref="UnsynchronizedClientSubscription"/> or restarts it if it is already running.        /// </summary>        public override void Start()        {            if (!Enabled)                m_startTimeSent = false;             // Reset compressor on successful resubscription            m_resetTsscEncoder = true;             base.Start();             if ((object)m_baseTimeRotationTimer != null && m_includeTime)                m_baseTimeRotationTimer.Start();        }         /// <summary>        /// Stops the <see cref="UnsynchronizedClientSubscription"/>.        /// </summary>	        public override void Stop()        {            base.Stop();             if ((object)m_baseTimeRotationTimer != null)            {                m_baseTimeRotationTimer.Stop();                m_baseTimeOffsets = null;            }        }         /// <summary>        /// Gets a short one-line status of this <see cref="UnsynchronizedClientSubscription"/>.        /// </summary>        /// <param name="maxLength">Maximum number of available characters for display.</param>        /// <returns>A short one-line summary of the current status of this <see cref="AdapterBase"/>.</returns>        public override string GetShortStatus(int maxLength)        {            int inputCount = 0, outputCount = 0;             if ((object)InputMeasurementKeys != null)                inputCount = InputMeasurementKeys.Length;             if ((object)OutputMeasurements != null)                outputCount = OutputMeasurements.Length;             return $"Total input measurements: {inputCount}, total output measurements: {outputCount}".PadLeft(maxLength);        }

Lines 1–500 of 959. Load more