//******************************************************************************************************//  RemoteOutputAdapter.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/01/2009 - J. Ritchie Carroll//       Generated original version of source code.//  09/15/2009 - Stephen C. Will//       Added new header and license agreement.//  09/22/2009 - Pinal C. Patel//       Re-wrote the adapter to utilize new components.//  09/23/2009 - Pinal C. Patel//       Fixed the handling of socket disconnect.//  03/04/2010 - Pinal C. Patel//       Added outputIsForArchive and throttleTransmission setting parameters for more control over //       the adapter.//       Switched to ManualResetEvent for waiting on historian acknowledgement for efficiency.//  01/20/2011 - Pinal C. Patel//       Modified to use Settings for the ConnectionString property of historian socket.//  12/13/2012 - Starlynn Danyelle Gilliam//       Modified Header.////****************************************************************************************************** using System;using System.Collections.Generic;using System.ComponentModel;using System.Text;using System.Threading;using GSF;using GSF.Communication;using GSF.Diagnostics;using GSF.Historian.Packets;using GSF.Parsing;using GSF.TimeSeries;using GSF.TimeSeries.Adapters; namespace HistorianAdapters{    /// <summary>    /// Represents an output adapter that publishes measurements to openHistorian for archival.    /// </summary>    [Description("Remote v1.0 Historian Publisher: Forwards measurements to a remote 1.0 openHistorian for archival")]    public class RemoteOutputAdapter : OutputAdapterBase    {        #region [ Members ]         // Constants        private const int DefaultHistorianPort = 1003;        private const bool DefaultPayloadAware = true;        private const bool DefaultConserveBandwidth = true;        private const bool DefaultOutputIsForArchive = true;        private const bool DefaultThrottleTransmission = true;        private const int DefaultSamplesPerTransmission = 100000;        private const int PublisherWaitTime = 5000;         // Fields        private string m_server;        private int m_port;        private bool m_payloadAware;        private bool m_conserveBandwidth;        private bool m_outputIsForArchive;        private bool m_throttleTransmission;        private int m_samplesPerTransmission;        private readonly TcpClient m_historianPublisher;        private byte[] m_publisherBuffer;        private readonly ManualResetEvent m_publisherWaitHandle;        private Action<IMeasurement[], int, int> m_publisherDelegate;        private bool m_publisherDisconnecting;        private long m_measurementsPublished;        private bool m_disposed;         #endregion         #region [ Constructors ]         /// <summary>        /// Initializes a new instance of the <see cref="RemoteOutputAdapter"/> class.        /// </summary>        public RemoteOutputAdapter()        {            m_historianPublisher = new TcpClient();            m_publisherWaitHandle = new ManualResetEvent(false);             m_port = DefaultHistorianPort;            m_payloadAware = DefaultPayloadAware;            m_conserveBandwidth = DefaultConserveBandwidth;            m_outputIsForArchive = DefaultOutputIsForArchive;            m_throttleTransmission = DefaultThrottleTransmission;            m_samplesPerTransmission = DefaultSamplesPerTransmission;        }         #endregion         #region [ Properties ]         /// <summary>        /// Gets or sets the host name for the server hosting the remote historian.        /// </summary>        [ConnectionStringParameter,        Description("Define the host name of the remote historian.")]        public string Server        {            get            {                return m_server;            }            set            {                m_server = value;            }        }         /// <summary>        /// Gets or sets the port on which the remote historian is listening.        /// </summary>        [ConnectionStringParameter,        Description("Define the port on which the remote historian is listening."),        DefaultValue(1003)]        public int Port        {            get            {                return m_port;            }            set            {                m_port = value;            }        }         /// <summary>        /// Gets or sets a boolean value indicating whether the payload        /// boundaries are to be preserved during transmission.        /// </summary>        [ConnectionStringParameter,        Description("Define a value indicating whether to preserve payload boundaries during transmission."),        DefaultValue(true)]        public bool PayloadAware        {            get            {                return m_payloadAware;            }            set            {                m_payloadAware = value;            }        }         /// <summary>        /// Gets or sets a boolean value that determines the packet        /// type to be used when sending data to the server.        /// </summary>        [ConnectionStringParameter,        Description("Define a value indicating the packet type when sending data to the server."),        DefaultValue(true)]        public bool ConserveBandwidth        {            get            {                return m_conserveBandwidth;            }            set            {                m_conserveBandwidth = value;            }        }         /// <summary>        /// Returns a flag that determines if measurements sent to this <see cref="RemoteOutputAdapter"/> are destined for archival.        /// </summary>        [ConnectionStringParameter,        Description("Define a value that determines whether the measurements are destined for archival."),        DefaultValue(true)]        public override bool OutputIsForArchive => m_outputIsForArchive;         /// <summary>        /// Gets or sets a boolean value that determines whether to wait for        /// acknowledgment from the historian that the last set of points have        /// been received before attempting to send the next set of points.        /// </summary>        [ConnectionStringParameter,        Description("Define a value that determines whether to wait for acknowledgment before sending more points."),        DefaultValue(true)]        public bool ThrottleTransmission        {            get            {                return m_throttleTransmission;            }            set            {                m_throttleTransmission = value;            }        }         /// <summary>        /// Gets or sets an integer that indicates the maximum number        /// of points to be published to the historian at once.        /// </summary>        [ConnectionStringParameter,        Description("Define the maximum number of points to be published at once."),        DefaultValue(100000)]        public int SamplesPerTransmission        {            get            {                return m_samplesPerTransmission;            }            set            {                m_samplesPerTransmission = value;            }        }         /// <summary>        /// Gets flag that determines if this <see cref="RemoteOutputAdapter"/> uses an asynchronous connection.        /// </summary>        protected override bool UseAsyncConnect => true;         /// <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_historianPublisher.Status);                 return status.ToString();            }        }         #endregion         #region [ Methods ]         /// <summary>        /// Initializes this <see cref="RemoteOutputAdapter"/>.        /// </summary>        /// <exception cref="ArgumentException"><b>Server</b> is missing from the <see cref="AdapterBase.Settings"/>.</exception>        public override void Initialize()        {            base.Initialize();             string errorMessage = "{0} is missing from Settings - Example: server=localhost;port=1003;payloadAware=True;conserveBandwidth=True;outputIsForArchive=True;throttleTransmission=True;samplesPerTransmission=100000";            Dictionary<string, string> settings = Settings;            string setting;             // Validate settings.            if (!settings.TryGetValue("server", out m_server))                throw new ArgumentException(string.Format(errorMessage, "server"));             if (settings.TryGetValue("port", out setting))                m_port = int.Parse(setting);            else                settings.Add("port", m_port.ToString());             if (settings.TryGetValue("payloadaware", out setting))                m_payloadAware = setting.ParseBoolean();             if (settings.TryGetValue("conservebandwidth", out setting))                m_conserveBandwidth = setting.ParseBoolean();             if (settings.TryGetValue("outputisforarchive", out setting))                m_outputIsForArchive = setting.ParseBoolean();             if (settings.TryGetValue("throttletransmission", out setting))                m_throttleTransmission = setting.ParseBoolean();             if (settings.TryGetValue("samplespertransmission", out setting))                m_samplesPerTransmission = int.Parse(setting);             // Initialize publisher delegates.            if (m_conserveBandwidth)            {                m_publisherDelegate = TransmitPacketType101;            }            else            {                m_publisherDelegate = TransmitPacketType1;                m_publisherBuffer = new byte[m_samplesPerTransmission * PacketType1.FixedLength];            }             // Initialize publisher socket.            m_historianPublisher.ConnectionString = settings.JoinKeyValuePairs();            m_historianPublisher.PayloadAware = m_payloadAware;            m_historianPublisher.ConnectionAttempt += HistorianPublisher_ConnectionAttempt;            m_historianPublisher.ConnectionEstablished += HistorianPublisher_ConnectionEstablished;            m_historianPublisher.ConnectionTerminated += HistorianPublisher_ConnectionTerminated;            m_historianPublisher.SendDataException += HistorianPublisher_SendDataException;            m_historianPublisher.ReceiveDataComplete += HistorianPublisher_ReceiveDataComplete;            m_historianPublisher.ReceiveDataException += HistorianPublisher_ReceiveDataException;            m_historianPublisher.Initialize();        }         /// <summary>        /// Gets a short one-line status of this <see cref="RemoteOutputAdapter"/>.        /// </summary>        /// <param name="maxLength">Maximum length of the status message.</param>        /// <returns>Text of the status message.</returns>        public override string GetShortStatus(int maxLength)        {            if (m_outputIsForArchive)                return $"Published {m_measurementsPublished} measurements for archival.".CenterText(maxLength);             return $"Published {m_measurementsPublished} measurements for processing.".CenterText(maxLength);        }         /// <summary>        /// Releases the unmanaged resources used by this <see cref="RemoteOutputAdapter"/> 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 (m_historianPublisher != null)                        {                            m_historianPublisher.ConnectionAttempt -= HistorianPublisher_ConnectionAttempt;                            m_historianPublisher.ConnectionEstablished -= HistorianPublisher_ConnectionEstablished;                            m_historianPublisher.ConnectionTerminated -= HistorianPublisher_ConnectionTerminated;                            m_historianPublisher.SendDataException -= HistorianPublisher_SendDataException;                            m_historianPublisher.ReceiveDataComplete -= HistorianPublisher_ReceiveDataComplete;                            m_historianPublisher.ReceiveDataException -= HistorianPublisher_ReceiveDataException;                            m_historianPublisher.Dispose();                        }                         m_publisherWaitHandle.Close();                    }                }                finally                {                    m_disposed = true;          // Prevent duplicate dispose.                    base.Dispose(disposing);    // Call base class Dispose().                }            }        }         /// <summary>        /// Attempts to connect to this <see cref="RemoteOutputAdapter"/>.        /// </summary>        protected override void AttemptConnection()        {            m_publisherDisconnecting = false;            m_historianPublisher.ConnectAsync();        }         /// <summary>        /// Attempts to disconnect from this <see cref="RemoteOutputAdapter"/>.        /// </summary>        protected override void AttemptDisconnection()        {            m_publisherDisconnecting = true;            m_historianPublisher.Disconnect();        }         /// <summary>        /// Publishes <paramref name="measurements"/> for archival.        /// </summary>        /// <param name="measurements">Measurements to be archived.</param>        /// <exception cref="OperationCanceledException">Acknowledgment is not received from historian for published data.</exception>        protected override void ProcessMeasurements(IMeasurement[] measurements)        {            if (m_historianPublisher.CurrentState != ClientState.Connected)                throw new InvalidOperationException("Historian publisher socket is not connected");             try            {                for (int i = 0; i < measurements.Length; i += m_samplesPerTransmission)                {                    // Wait for historian acknowledgment.                    if (m_throttleTransmission)                    {                        if (!m_publisherWaitHandle.WaitOne(PublisherWaitTime))                            throw new OperationCanceledException("Timeout waiting for acknowledgment from historian");                    }                     // Publish measurements to historian.                    m_publisherWaitHandle.Reset();                    m_publisherDelegate(measurements, i, (measurements.Length - i < m_samplesPerTransmission ? measurements.Length : i + m_samplesPerTransmission) - 1);                }                m_measurementsPublished += measurements.Length;            }            catch            {                m_publisherWaitHandle.Set();                throw;            }        }         private void HistorianPublisher_ConnectionAttempt(object sender, EventArgs e)        {            OnStatusMessage(MessageLevel.Info, "Attempting socket connection...");        }         private void HistorianPublisher_ConnectionEstablished(object sender, EventArgs e)        {            OnConnected();            m_publisherWaitHandle.Set();        }         private void HistorianPublisher_ConnectionTerminated(object sender, EventArgs e)        {            m_measurementsPublished = 0;            m_publisherWaitHandle.Reset();             if (!m_publisherDisconnecting)                Start();        }         private void HistorianPublisher_SendDataException(object sender, EventArgs<Exception> e)        {            m_publisherWaitHandle.Set();            OnProcessException(MessageLevel.Warning, e.Argument);        }         private void HistorianPublisher_ReceiveDataComplete(object sender, EventArgs<byte[], int> e)        {            // Check for acknowledgment from historian.            string reply = Encoding.ASCII.GetString(e.Argument1, 0, e.Argument2);            if (reply == "ACK")                m_publisherWaitHandle.Set();        }         private void HistorianPublisher_ReceiveDataException(object sender, EventArgs<Exception> e)        {            m_publisherWaitHandle.Set();            OnProcessException(MessageLevel.Warning, e.Argument);        }         private void TransmitPacketType1(IMeasurement[] measurements, int startIndex, int endIndex)        {            int bufferIndex = 0;             for (int i = startIndex; i <= endIndex; i++)                bufferIndex += new PacketType1(measurements[i]).GenerateBinaryImage(m_publisherBuffer, bufferIndex);             m_historianPublisher.SendAsync(m_publisherBuffer, 0, bufferIndex);        }         private void TransmitPacketType101(IMeasurement[] measurements, int startIndex, int endIndex)        {            PacketType101 packet = new PacketType101();             for (int i = startIndex; i <= endIndex; i++)                packet.Data.Add(new PacketType101DataPoint(measurements[i]));             m_historianPublisher.SendAsync(packet.BinaryImage());        }         #endregion    }}