//******************************************************************************************************// DataSubscriber.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.// 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 metadata synchronization.////****************************************************************************************************** using GSF.Collections;using GSF.Communication;//using GSF.Data;using GSF.IO;using GSF.Reflection;//using GSF.Security.Cryptography;using GSF.TimeSeries.Adapters;//using GSF.TimeSeries.Statistics;using System;using System.Collections.Generic;using System.ComponentModel;using System.Data;using System.IO;using System.IO.Compression;using System.Linq;using System.Text;using System.Threading;using System.Xml; 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 } // Constants 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 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 procesing. /// </remarks> public new event EventHandler<EventArgs<string>> ProcessingComplete; // Fields private TcpClient m_commandChannel; private UdpClient m_dataChannel; private LocalConcentrator m_localConcentrator; private System.Timers.Timer m_dataStreamMonitor; private long m_commandChannelConnectionAttempts; private long m_dataChannelConnectionAttempts; private volatile SignalIndexCache m_remoteSignalIndexCache; private volatile SignalIndexCache m_signalIndexCache; private volatile long[] m_baseTimeOffsets; private volatile int m_timeIndex; private volatile byte[][][] m_keyIVs; private volatile int m_cipherIndex; private volatile bool m_authenticated; private volatile bool m_subscribed; private volatile int m_lastBytesReceived; private long m_monitoredBytesReceived; private long m_totalBytesReceived; private long m_lastMissingCacheWarning; private Guid m_nodeID; private int m_gatewayProtocolID; private List<ServerCommand> m_requests; private bool m_synchronizedSubscription; private bool m_requireAuthentication; private bool m_useMillisecondResolution; private bool m_autoConnect; private string m_sharedSecret; private string m_authenticationID; private bool m_includeTime; private bool m_synchronizeMetadata; private bool m_internal; private OperationalModes m_operationalModes; private Encoding m_encoding; private bool m_disposed; #endregion #region [ Constructors ] /// <summary> /// Creates a new <see cref="DataSubscriber"/>. /// </summary> public DataSubscriber() { m_requests = new List<ServerCommand>(); m_encoding = Encoding.Unicode; DataLossInterval = 10.0D; } #endregion #region [ Properties ] /// <summary> /// Gets or sets flag that determines if <see cref="DataPublisher"/> requires subscribers to authenticate before making data requests. /// </summary> public bool RequireAuthentication { get { return m_requireAuthentication; } set { m_requireAuthentication = value; } } /// <summary> /// Gets or sets flag that determines if <see cref="DataSubscriber"/> should attempt to autoconnection to <see cref="DataPublisher"/> using defined connection settings. /// </summary> public bool AutoConnect { get { return m_autoConnect; } set { m_autoConnect = value; } } /// <summary> /// Gets or sets flag that informs publisher if base time-offsets can use millisecond resolution to conserve bandwidth. /// </summary> [Obsolete("SubscriptionInfo object defines this parameter.", false)] public bool UseMillisecondResolution { get { return m_useMillisecondResolution; } set { m_useMillisecondResolution = value; } } /// <summary> /// Gets flag that determines if this <see cref="DataSubscriber"/> has successfully authenticated with the <see cref="DataPublisher"/>. /// </summary> public bool Authenticated { get { return m_authenticated; } } /// <summary> /// Gets total data packet bytes received during this session. /// </summary> public long TotalBytesReceived { get { return m_totalBytesReceived; } } /// <summary> /// Gets or sets data loss monitoring interval, in seconds. Set to zero to disable monitoring. /// </summary> public double DataLossInterval { get { if ((object)m_dataStreamMonitor != null) return m_dataStreamMonitor.Interval / 1000.0D; return 0.0D; } set { if (value > 0.0D) { if ((object)m_dataStreamMonitor == null) { // Create data stream monitoring timer m_dataStreamMonitor = new System.Timers.Timer(); m_dataStreamMonitor.Elapsed += m_dataStreamMonitor_Elapsed; m_dataStreamMonitor.AutoReset = true; m_dataStreamMonitor.Enabled = false; } // Set user specified interval m_dataStreamMonitor.Interval = value * 1000.0D; } else { // Disable data monitor if ((object)m_dataStreamMonitor != null) { m_dataStreamMonitor.Elapsed -= m_dataStreamMonitor_Elapsed; m_dataStreamMonitor.Dispose(); } m_dataStreamMonitor = null; } } } /// <summary> /// Gets or sets a set of flags that define ways in /// which the subscriber and publisher communicate. /// </summary> public OperationalModes OperationalModes { get { return m_operationalModes; } set { OperationalEncoding operationalEncoding; m_operationalModes = value; operationalEncoding = (OperationalEncoding)(value & OperationalModes.EncodingMask); m_encoding = GetCharacterEncoding(operationalEncoding); } } /// <summary> /// Gets or sets the <see cref="OperationalEncoding"/> used by the subscriber and publisher. /// </summary> public OperationalEncoding OperationalEncoding { get { return (OperationalEncoding)(m_operationalModes & OperationalModes.EncodingMask); } set { m_operationalModes &= ~OperationalModes.EncodingMask; m_operationalModes |= (OperationalModes)value; m_encoding = GetCharacterEncoding(value); } } /// <summary> /// Gets the flag indicating if this adapter supports temporal processing. /// </summary> /// <remarks> /// Although the data subscriber provisions support for temporal processing by receiving historical data from a remote source, /// the adapter opens sockets and does not need to be engaged within an actual temporal <see cref="IaonSession"/>, therefore /// this method returns <c>false</c> to make sure the adapter doesn't get instantiated within a temporal session. /// </remarks> public override bool SupportsTemporalProcessing { get { return false; } } /// <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, overwhich 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; // Request server update the processing interval SendServerCommand(ServerCommand.UpdateProcessingInterval, EndianOrder.BigEndian.GetBytes(value)); } } /// <summary> /// Gets the status of this <see cref="DataSubscriber"/>. /// </summary> /// <remarks> /// Derived classes should provide current status information about the adapter for display purposes. /// </remarks> public override string Status { get { StringBuilder status = new StringBuilder(); status.AppendFormat(" Subscription mode: {0}", m_synchronizedSubscription ? "Remotely Synchronized" : (object)m_localConcentrator == null ? "Unsynchronized" : "Locally Synchronized"); status.AppendLine(); status.AppendFormat(" Pending command requests: {0}", m_requests.Count); status.AppendLine(); status.AppendFormat(" Authenticated: {0}", m_authenticated); status.AppendLine(); status.AppendFormat(" Subscribed: {0}", m_subscribed); status.AppendLine(); status.AppendFormat(" Data packet security: {0}", m_keyIVs == null ? "unencrypted" : "encrypted"); status.AppendLine(); if (DataLossInterval > 0.0D) status.AppendFormat("No data reconnect interval: {0} seconds", DataLossInterval.ToString("0.000")); else status.Append("No data reconnect interval: disabled"); status.AppendLine(); if ((object)m_dataChannel != null) { status.AppendLine(); status.AppendLine("Data Channel Status".CenterText(50)); status.AppendLine("-------------------".CenterText(50)); status.Append(m_dataChannel.Status); } if ((object)m_commandChannel != null) { status.AppendLine(); status.AppendLine("Command Channel Status".CenterText(50)); status.AppendLine("----------------------".CenterText(50)); status.Append(m_commandChannel.Status); } if ((object)m_localConcentrator != null) { status.AppendLine(); status.AppendLine("Local Concentrator Status".CenterText(50)); status.AppendLine("-------------------------".CenterText(50)); status.Append(m_localConcentrator.Status); } status.Append(base.Status); return status.ToString(); } } /// <summary> /// Gets a flag that determines if this <see cref="DataSubscriber"/> uses an asynchronous connection. /// </summary> protected override bool UseAsyncConnect { get { return true; } } /// <summary> /// Gets or sets reference to <see cref="UdpClient"/> data channel, attaching and/or detaching to events as needed. /// </summary> protected UdpClient DataChannel { get { return m_dataChannel; } set { if (m_dataChannel != null) { // Detach from events on existing data channel reference m_dataChannel.ConnectionException -= m_dataChannel_ConnectionException; m_dataChannel.ConnectionAttempt -= m_dataChannel_ConnectionAttempt; m_dataChannel.ReceiveData -= m_dataChannel_ReceiveData; m_dataChannel.ReceiveDataException -= m_dataChannel_ReceiveDataException; if (m_dataChannel != value) m_dataChannel.Dispose(); } // Assign new data channel reference m_dataChannel = value; if (m_dataChannel != null) {Lines 1–500 of 2600. Load more
//****************************************************************************************************** // DataSubscriber.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. // 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 metadata synchronization. // //****************************************************************************************************** using GSF.Collections; using GSF.Communication; //using GSF.Data; using GSF.IO; using GSF.Reflection; //using GSF.Security.Cryptography; using GSF.TimeSeries.Adapters; //using GSF.TimeSeries.Statistics; using System; using System.Collections.Generic; using System.ComponentModel; using System.Data; using System.IO; using System.IO.Compression; using System.Linq; using System.Text; using System.Threading; using System.Xml; 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 } // Constants 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 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 procesing. /// </remarks> public new event EventHandler<EventArgs<string>> ProcessingComplete; // Fields private TcpClient m_commandChannel; private UdpClient m_dataChannel; private LocalConcentrator m_localConcentrator; private System.Timers.Timer m_dataStreamMonitor; private long m_commandChannelConnectionAttempts; private long m_dataChannelConnectionAttempts; private volatile SignalIndexCache m_remoteSignalIndexCache; private volatile SignalIndexCache m_signalIndexCache; private volatile long[] m_baseTimeOffsets; private volatile int m_timeIndex; private volatile byte[][][] m_keyIVs; private volatile int m_cipherIndex; private volatile bool m_authenticated; private volatile bool m_subscribed; private volatile int m_lastBytesReceived; private long m_monitoredBytesReceived; private long m_totalBytesReceived; private long m_lastMissingCacheWarning; private Guid m_nodeID; private int m_gatewayProtocolID; private List<ServerCommand> m_requests; private bool m_synchronizedSubscription; private bool m_requireAuthentication; private bool m_useMillisecondResolution; private bool m_autoConnect; private string m_sharedSecret; private string m_authenticationID; private bool m_includeTime; private bool m_synchronizeMetadata; private bool m_internal; private OperationalModes m_operationalModes; private Encoding m_encoding; private bool m_disposed; #endregion #region [ Constructors ] /// <summary> /// Creates a new <see cref="DataSubscriber"/>. /// </summary> public DataSubscriber() { m_requests = new List<ServerCommand>(); m_encoding = Encoding.Unicode; DataLossInterval = 10.0D; } #endregion #region [ Properties ] /// <summary> /// Gets or sets flag that determines if <see cref="DataPublisher"/> requires subscribers to authenticate before making data requests. /// </summary> public bool RequireAuthentication { get { return m_requireAuthentication; } set { m_requireAuthentication = value; } } /// <summary> /// Gets or sets flag that determines if <see cref="DataSubscriber"/> should attempt to autoconnection to <see cref="DataPublisher"/> using defined connection settings. /// </summary> public bool AutoConnect { get { return m_autoConnect; } set { m_autoConnect = value; } } /// <summary> /// Gets or sets flag that informs publisher if base time-offsets can use millisecond resolution to conserve bandwidth. /// </summary> [Obsolete("SubscriptionInfo object defines this parameter.", false)] public bool UseMillisecondResolution { get { return m_useMillisecondResolution; } set { m_useMillisecondResolution = value; } } /// <summary> /// Gets flag that determines if this <see cref="DataSubscriber"/> has successfully authenticated with the <see cref="DataPublisher"/>. /// </summary> public bool Authenticated { get { return m_authenticated; } } /// <summary> /// Gets total data packet bytes received during this session. /// </summary> public long TotalBytesReceived { get { return m_totalBytesReceived; } } /// <summary> /// Gets or sets data loss monitoring interval, in seconds. Set to zero to disable monitoring. /// </summary> public double DataLossInterval { get { if ((object)m_dataStreamMonitor != null) return m_dataStreamMonitor.Interval / 1000.0D; return 0.0D; } set { if (value > 0.0D) { if ((object)m_dataStreamMonitor == null) { // Create data stream monitoring timer m_dataStreamMonitor = new System.Timers.Timer(); m_dataStreamMonitor.Elapsed += m_dataStreamMonitor_Elapsed; m_dataStreamMonitor.AutoReset = true; m_dataStreamMonitor.Enabled = false; } // Set user specified interval m_dataStreamMonitor.Interval = value * 1000.0D; } else { // Disable data monitor if ((object)m_dataStreamMonitor != null) { m_dataStreamMonitor.Elapsed -= m_dataStreamMonitor_Elapsed; m_dataStreamMonitor.Dispose(); } m_dataStreamMonitor = null; } } } /// <summary> /// Gets or sets a set of flags that define ways in /// which the subscriber and publisher communicate. /// </summary> public OperationalModes OperationalModes { get { return m_operationalModes; } set { OperationalEncoding operationalEncoding; m_operationalModes = value; operationalEncoding = (OperationalEncoding)(value & OperationalModes.EncodingMask); m_encoding = GetCharacterEncoding(operationalEncoding); } } /// <summary> /// Gets or sets the <see cref="OperationalEncoding"/> used by the subscriber and publisher. /// </summary> public OperationalEncoding OperationalEncoding { get { return (OperationalEncoding)(m_operationalModes & OperationalModes.EncodingMask); } set { m_operationalModes &= ~OperationalModes.EncodingMask; m_operationalModes |= (OperationalModes)value; m_encoding = GetCharacterEncoding(value); } } /// <summary> /// Gets the flag indicating if this adapter supports temporal processing. /// </summary> /// <remarks> /// Although the data subscriber provisions support for temporal processing by receiving historical data from a remote source, /// the adapter opens sockets and does not need to be engaged within an actual temporal <see cref="IaonSession"/>, therefore /// this method returns <c>false</c> to make sure the adapter doesn't get instantiated within a temporal session. /// </remarks> public override bool SupportsTemporalProcessing { get { return false; } } /// <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, overwhich 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; // Request server update the processing interval SendServerCommand(ServerCommand.UpdateProcessingInterval, EndianOrder.BigEndian.GetBytes(value)); } } /// <summary> /// Gets the status of this <see cref="DataSubscriber"/>. /// </summary> /// <remarks> /// Derived classes should provide current status information about the adapter for display purposes. /// </remarks> public override string Status { get { StringBuilder status = new StringBuilder(); status.AppendFormat(" Subscription mode: {0}", m_synchronizedSubscription ? "Remotely Synchronized" : (object)m_localConcentrator == null ? "Unsynchronized" : "Locally Synchronized"); status.AppendLine(); status.AppendFormat(" Pending command requests: {0}", m_requests.Count); status.AppendLine(); status.AppendFormat(" Authenticated: {0}", m_authenticated); status.AppendLine(); status.AppendFormat(" Subscribed: {0}", m_subscribed); status.AppendLine(); status.AppendFormat(" Data packet security: {0}", m_keyIVs == null ? "unencrypted" : "encrypted"); status.AppendLine(); if (DataLossInterval > 0.0D) status.AppendFormat("No data reconnect interval: {0} seconds", DataLossInterval.ToString("0.000")); else status.Append("No data reconnect interval: disabled"); status.AppendLine(); if ((object)m_dataChannel != null) { status.AppendLine(); status.AppendLine("Data Channel Status".CenterText(50)); status.AppendLine("-------------------".CenterText(50)); status.Append(m_dataChannel.Status); } if ((object)m_commandChannel != null) { status.AppendLine(); status.AppendLine("Command Channel Status".CenterText(50)); status.AppendLine("----------------------".CenterText(50)); status.Append(m_commandChannel.Status); } if ((object)m_localConcentrator != null) { status.AppendLine(); status.AppendLine("Local Concentrator Status".CenterText(50)); status.AppendLine("-------------------------".CenterText(50)); status.Append(m_localConcentrator.Status); } status.Append(base.Status); return status.ToString(); } } /// <summary> /// Gets a flag that determines if this <see cref="DataSubscriber"/> uses an asynchronous connection. /// </summary> protected override bool UseAsyncConnect { get { return true; } } /// <summary> /// Gets or sets reference to <see cref="UdpClient"/> data channel, attaching and/or detaching to events as needed. /// </summary> protected UdpClient DataChannel { get { return m_dataChannel; } set { if (m_dataChannel != null) { // Detach from events on existing data channel reference m_dataChannel.ConnectionException -= m_dataChannel_ConnectionException; m_dataChannel.ConnectionAttempt -= m_dataChannel_ConnectionAttempt; m_dataChannel.ReceiveData -= m_dataChannel_ReceiveData; m_dataChannel.ReceiveDataException -= m_dataChannel_ReceiveDataException; if (m_dataChannel != value) m_dataChannel.Dispose(); } // Assign new data channel reference m_dataChannel = value; if (m_dataChannel != null) {