//******************************************************************************************************
// UdpClient.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:
// ----------------------------------------------------------------------------------------------------
// 07/06/2006 - Pinal C. Patel
// Original version of source code generated.
// 09/06/2006 - J. Ritchie Carroll
// Added bypass optimizations for high-speed socket access.
// 09/27/2007 - J. Ritchie Carroll
// Added disconnect timeout overload.
// 09/29/2008 - J. Ritchie Carroll
// Converted to C#.
// 07/08/2009 - J. Ritchie Carroll
// Added WaitHandle return value from asynchronous connection.
// 07/09/2009 - Pinal C. Patel
// Modified to attempt resuming reception on SocketException for non-Handshake enabled connection.
// 07/15/2009 - Pinal C. Patel
// Modified Disconnect() to add error checking.
// 07/17/2009 - Pinal C. Patel
// Added support to specify a specific interface address on a multiple interface machine.
// 07/20/2009 - Pinal C. Patel
// Allowed for UDP endpoint to not be bound to a local interface by specifying -1 for port number.
// 09/14/2009 - Stephen C. Wills
// Added new header and license agreement.
// 10/30/2009 - Pinal C. Patel
// Added true multicast support by allowing for socket level subscription to a multicast group.
// 11/17/2009 - Pinal C. Patel
// Fixed a issue in the creation of random server endpoint when server endpoint information is
// omitted from the ConnectionString.
// 03/24/2010 - Pinal C. Patel
// Updated the interpretation of server property in ConnectionString to correctly interpret
// IPv6 IP addresses according to IETF - A Recommendation for IPv6 Address Text Representation.
// 11/29/2010 - Pinal C. Patel
// Corrected the implementation of ConnectAsync() method.
// 02/13/2011 - Pinal C. Patel
// Modified ConnectAsync() to handle loop-back address resolution failure on IPv6 enabled OSes.
// 07/23/2012 - Stephen C. Wills
// Performed a full refactor to use the SocketAsyncEventArgs API calls.
// 10/31/2012 - Stephen C. Wills
// Replaced single-threaded BlockingCollection pattern with asynchronous loop pattern.
// 12/13/2012 - Starlynn Danyelle Gilliam
// Modified Header.
// 09/24/2015 - Allan V. Scheid
// Fixed Mono socket error with System.Net.Sockets.Socket.IOControl method and SIO_UDP_CONNRESET
// inside OpenPort().
//
//******************************************************************************************************
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.ComponentModel;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using GSF.Configuration;
using GSF.Diagnostics;
using GSF.IO;
using GSF.Threading;
namespace GSF.Communication
{
///
/// Represents a UDP-based communication server.
///
///
///
/// Use when the primary purpose is to receive data.
///
///
/// The socket can be bound to a specified interface on a machine with multiple interfaces by
/// specifying the interface in the (Example: "Server=localhost:8888; Port=8989; Interface=127.0.0.1")
///
///
/// The socket can be used just for transmitting data without being bound to a local interface
/// by specifying -1 for the port number in the (Example: "Server=localhost:8888; Port=-1")
///
///
///
/// This example shows how to use the component:
///
/// using System;
/// using GSF;
/// using GSF.Communication;
/// using GSF.Security.Cryptography;
/// using GSF.IO.Compression;
///
/// class Program
/// {
/// static UdpClient s_client;
///
/// static void Main(string[] args)
/// {
/// // Initialize the client.
/// s_client = new UdpClient("Server=localhost:8888; Port=8989");
/// s_client.Handshake = false;
/// s_client.ReceiveTimeout = -1;
/// s_client.Encryption = CipherStrength.None;
/// s_client.Compression = CompressionStrength.NoCompression;
/// s_client.SecureSession = false;
/// s_client.Initialize();
/// // Register event handlers.
/// s_client.ConnectionAttempt += s_client_ConnectionAttempt;
/// s_client.ConnectionEstablished += s_client_ConnectionEstablished;
/// s_client.ConnectionTerminated += s_client_ConnectionTerminated;
/// s_client.ReceiveDataComplete += s_client_ReceiveDataComplete;
/// // Connect the client.
/// s_client.Connect();
///
/// // Transmit user input to the server.
/// string input;
/// while (string.Compare(input = Console.ReadLine(), "Exit", true) != 0)
/// {
/// s_client.Send(input);
/// }
///
/// // Disconnect the client on shutdown.
/// s_client.Dispose();
/// }
///
/// static void s_client_ConnectionAttempt(object sender, EventArgs e)
/// {
/// Console.WriteLine("Client is connecting to server.");
/// }
///
/// static void s_client_ConnectionEstablished(object sender, EventArgs e)
/// {
/// Console.WriteLine("Client connected to server.");
/// }
///
/// static void s_client_ConnectionTerminated(object sender, EventArgs e)
/// {
/// Console.WriteLine("Client disconnected from server.");
/// }
///
/// static void s_client_ReceiveDataComplete(object sender, EventArgs<byte[], int> e)
/// {
/// Console.WriteLine(string.Format("Received data - {0}.", s_client.TextEncoding.GetString(e.Argument1, 0, e.Argument2)));
/// }
/// }
///
///
// ReSharper disable UnusedVariable
public class UdpClient : ClientBase
{
#region [ Members ]
// Nested Types
private class UdpClientPayload
{
public EndPoint Destination;
public byte[] Data;
public int Offset;
public int Length;
public ManualResetEventSlim WaitHandle;
}
// Constants
///
/// Specifies the default value for the property.
///
public new const int DefaultReceiveBufferSize = 65536;
///
/// Specifies the default value for the property.
///
public const int DefaultMaxPacketSize = 65536;
///
/// Specifies the default value for the property.
///
public const bool DefaultAllowDualStackSocket = true;
///
/// Specifies the default value for the property.
///
public const int DefaultMaxSendQueueSize = 500000;
///
/// Specifies the default value for the property.
///
public const string DefaultConnectionString = "Server=localhost:8888; Port=8989";
///
/// Specifies the constant to be used for disabling when endpoint is not listening.
///
private const int SIO_UDP_CONNRESET = -1744830452;
// Events
///
/// Occurs when unprocessed data has been received from the server.
///
///
///
/// This event can be used to receive a notification that server data has arrived. The method can then be used
/// to copy data to an existing buffer. In many cases it will be optimal to use an existing buffer instead of subscribing to the
/// event.
///
///
/// is the end-point that data has been received from.
/// is the number of bytes received in the buffer from the server.
///
///
[Category("Data")]
[Description("Occurs when unprocessed data has been received from the server.")]
public event EventHandler> ReceiveDataFrom;
///
/// Occurs when data received from the server has been processed and is ready for consumption.
///
///
/// is the end-point that data has been received from.
/// is a new buffer containing post-processed data received from the server starting at index zero.
/// is the number of post-processed bytes received in the buffer from the server.
///
[Category("Data")]
[Description("Occurs when data received from the server has been processed and is ready for consumption.")]
public event EventHandler> ReceiveDataFromComplete;
// Fields
//private bool m_destinationReachableCheck;
private IPEndPoint m_udpServer;
private TransportProvider m_udpClient;
private IPStack m_ipStack;
private int m_maxPacketSize;
private Dictionary m_connectData;
private ManualResetEvent m_connectionHandle;
private Thread m_connectionThread;
private int m_sending;
private int m_receiving;
private readonly object m_sendLock;
private readonly ConcurrentQueue m_sendQueue;
private readonly ShortSynchronizedOperation m_dumpPayloadsOperation;
private SocketAsyncEventArgs m_sendArgs;
private SocketAsyncEventArgs m_receiveArgs;
private readonly EventHandler m_sendHandler;
private readonly EventHandler m_receiveHandler;
private bool m_receivePacketInfo;
private bool m_disposed;
#endregion
#region [ Constructors ]
///
/// Initializes a new instance of the class.
///
public UdpClient() : this(DefaultConnectionString)
{
}
///
/// Initializes a new instance of the class.
///
/// Connect string of the . See for format.
public UdpClient(string connectString) : base(TransportProtocol.Udp, connectString)
{
base.ReceiveBufferSize = DefaultReceiveBufferSize;
m_maxPacketSize = DefaultMaxPacketSize;
AllowDualStackSocket = DefaultAllowDualStackSocket;
MaxSendQueueSize = DefaultMaxSendQueueSize;
m_sendLock = new object();
m_sendQueue = new ConcurrentQueue();
m_dumpPayloadsOperation = new ShortSynchronizedOperation(DumpPayloads, OnSendDataException);
m_sendHandler += (sender, args) => ProcessSend();
m_receiveHandler += (sender, args) => ProcessReceive();
}
///
/// Initializes a new instance of the class.
///
/// object that contains the .
public UdpClient(IContainer container) : this() => container?.Add(this);
#endregion
#region [ Properties ]
///
/// Gets or sets a boolean value that determines if dual-mode socket is allowed when endpoint address is IPv6.
///
[Category("Settings")]
[DefaultValue(DefaultAllowDualStackSocket)]
[Description("Determines if dual-mode socket is allowed when endpoint address is IPv6.")]
public bool AllowDualStackSocket { get; set; }
///
/// Gets or sets the maximum size for the send queue before payloads are dumped from the queue.
///
[Category("Settings")]
[DefaultValue(DefaultMaxSendQueueSize)]
[Description("The maximum size for the send queue before payloads are dumped from the queue.")]
public int MaxSendQueueSize { get; set; }
///
/// Gets the object for the .
///
[Browsable(false)]
public Socket Client => m_udpClient.Provider;
///
/// Gets the server URI of the .
///
[Browsable(false)]
public override string ServerUri
{
get
{
if (m_connectData is null)
return $"{TransportProtocol}://0.0.0.0:0";
if (m_connectData.ContainsKey("server"))
return $"{TransportProtocol}://{m_connectData["server"]}".ToLower();
if (!m_connectData.TryGetValue("interface", out string targetInterface) || string.IsNullOrWhiteSpace(targetInterface))
{
IPStack ipStack = m_ipStack == IPStack.Default ? Transport.GetDefaultIPStack() : m_ipStack;
targetInterface = ipStack == IPStack.IPv6 ? "[::0]" : "0.0.0.0";
}
if (!m_connectData.TryGetValue("port", out string targetPort) || !ushort.TryParse(targetPort, out ushort port))
port = 0;
return $"{TransportProtocol}://{targetInterface}:{port}".ToLower();
}
}
///
/// Gets or sets the size of the buffer used by the client for receiving data from the server.
///
/// The value being assigned is either zero or negative.
public override int ReceiveBufferSize
{
get => base.ReceiveBufferSize;
set
{
base.ReceiveBufferSize = value;
if (m_udpClient?.Provider != null)
m_udpClient.Provider.ReceiveBufferSize = value;
}
}
///
/// Gets or sets the maximum expected size for packets being received by this .
///
public int MaxPacketSize
{
get => m_maxPacketSize;
set
{
m_maxPacketSize = value;
m_udpClient?.SetReceiveBuffer(value);
}
}
///
/// Gets or sets the flag that determines whether the UDP client
/// should attempt to receive packet info when receiving data
/// from the socket.
///
public bool ReceivePacketInfo
{
get
{
return m_receivePacketInfo;
}
set
{
#if MONO
if (value)
throw new NotImplementedException("Not supported under Mono.");
m_receivePacketInfo = false;
#else
m_receivePacketInfo = value;
m_udpClient?.Provider?.SetSocketOption(m_udpClient.Provider.AddressFamily == AddressFamily.InterNetworkV6 ? SocketOptionLevel.IPv6 : SocketOptionLevel.IP, SocketOptionName.PacketInformation, value);
#endif
}
}
///
/// Determines whether the base class should track statistics.
///
protected override bool TrackStatistics => false;
///
/// Gets the descriptive status of the client.
///
public override string Status
{
get
{
StringBuilder statusBuilder = new StringBuilder(base.Status);
if (m_sendQueue != null)
{
statusBuilder.AppendFormat(" Queued payloads: {0}", m_sendQueue.Count);
statusBuilder.AppendLine();
}
return statusBuilder.ToString();
}
}
#endregion
#region [ Methods ]
///
/// Reads a number of bytes from the current received data buffer and writes those bytes into a byte array at the specified offset.
///
/// Destination buffer used to hold copied bytes.
/// 0-based starting index into destination to begin writing data.
/// The number of bytes to read from current received data buffer and write into .
/// The number of bytes read.
///
/// This function should only be called from within the event handler. Calling this method outside
/// this event will have unexpected results.
///
/// No received data buffer has been defined to read.
/// is null.
///
/// or is less than 0 -or-
/// and will exceed length.
///
public override int Read(byte[] buffer, int startIndex, int length)
{
buffer.ValidateParameters(startIndex, length);
if (m_udpClient.ReceiveBuffer == null)
throw new InvalidOperationException("No received data buffer has been defined to read.");
int sourceLength = m_udpClient.BytesReceived - ReadIndex;
int readBytes = length > sourceLength ? sourceLength : length;
Buffer.BlockCopy(m_udpClient.ReceiveBuffer, ReadIndex, buffer, startIndex, readBytes);
// Update read index for next call
ReadIndex += readBytes;
if (ReadIndex >= m_udpClient.BytesReceived)
ReadIndex = 0;
return readBytes;
}
///
/// Saves settings to the config file if the property is set to true.
///
public override void SaveSettings()
{
base.SaveSettings();
if (!PersistSettings)
return;
// Save settings under the specified category.
ConfigurationFile config = ConfigurationFile.Current;
CategorizedSettingsElementCollection settings = config.Settings[SettingsCategory];
settings["AllowDualStackSocket", true].Update(AllowDualStackSocket);
settings["MaxSendQueueSize", true].Update(MaxSendQueueSize);
config.Save();
}
///
/// Loads saved settings from the config file if the property is set to true.
///
public override void LoadSettings()
{
base.LoadSettings();
if (!PersistSettings)
return;
// Load settings from the specified category.
ConfigurationFile config = ConfigurationFile.Current;
CategorizedSettingsElementCollection settings = config.Settings[SettingsCategory];
settings.Add("AllowDualStackSocket", AllowDualStackSocket, "True if dual-mode socket is allowed when IP address is IPv6, otherwise False.");
settings.Add("MaxSendQueueSize", MaxSendQueueSize, "The maximum size of the send queue before payloads are dumped from the queue.");
AllowDualStackSocket = settings["AllowDualStackSocket"].ValueAs(AllowDualStackSocket);
MaxSendQueueSize = settings["MaxSendQueueSize"].ValueAs(MaxSendQueueSize);
// Overwrite config file if max send queue size exists in connection string.
if (m_connectData.ContainsKey("maxSendQueueSize") && int.TryParse(m_connectData["maxSendQueueSize"], out int maxSendQueueSize))
MaxSendQueueSize = maxSendQueueSize;
}
///
/// Disconnects the from the connected server synchronously.
///
public override void Disconnect()
{
if (CurrentState == ClientState.Disconnected)
return;
try
{
base.Disconnect();
if (m_udpServer != null && m_udpClient.Provider != null)
{
// If the IP specified for the server is a multicast IP, unsubscribe from the specified multicast group.
IPEndPoint serverEndpoint = m_udpServer;
if (Transport.IsMulticastIP(serverEndpoint.Address))
DropMulticastMembership(serverEndpoint.Address, null, m_udpClient.MulticastMembershipAddresses);
}
}
catch (Exception ex)
{
OnSendDataException(new InvalidOperationException($"Failed to drop multicast membership: {ex.Message}", ex));
}
m_udpClient.Reset();
m_connectionThread?.Abort();
}
///
/// Connects the to the server asynchronously.
///
/// Server property in is invalid.
/// Attempt is made to connect the when it is not disconnected.
/// for the asynchronous operation.
public override WaitHandle ConnectAsync()
{
// Client may still be attempting to receive data from a prior connection
if (Interlocked.CompareExchange(ref m_receiving, 0, 0) != 0)
throw new InvalidOperationException("Client is not yet fully disconnected");
m_connectionHandle = (ManualResetEvent)base.ConnectAsync();
m_udpClient = new TransportProvider();
m_udpClient.SetReceiveBuffer(m_maxPacketSize);
// Create a server endpoint.
if (m_connectData.ContainsKey("server"))
{
// Client has a server endpoint specified.
Match endpoint = Regex.Match(m_connectData["server"], Transport.EndpointFormatRegex);
if (endpoint != Match.Empty)
m_udpServer = Transport.CreateEndPoint(endpoint.Groups["host"].Value, int.Parse(endpoint.Groups["port"].Value), m_ipStack);
else
throw new FormatException($"Server property in ConnectionString is invalid (Example: {DefaultConnectionString})");
}
else
{
// Create a random server endpoint since one is not specified.
m_udpServer = Transport.CreateEndPoint(m_connectData["interface"], 0, m_ipStack);
}
m_connectionThread = new Thread(OpenPort)
{
IsBackground = true
};
m_connectionThread.Start();
return m_connectionHandle;
}
///
/// Adds a multicast membership to the UDP socket.
///
/// Multicast address to join.
/// Address which defines the source of the data or null if the membership is not source-specific.
public void AddMulticastMembership(IPAddress serverAddress, IPAddress sourceAddress) => AddMulticastMembership(serverAddress, sourceAddress, out _);
///
/// Drops a multicast membership from the UDP socket.
///
/// Multicast address to drop.
/// Address which defines the source of the data or null if the membership is not source-specific.
public void DropMulticastMembership(IPAddress serverAddress, IPAddress sourceAddress) => DropMulticastMembership(serverAddress, sourceAddress, null);
///
/// Releases the unmanaged resources used by the and optionally releases the managed resources.
///
/// true to release both managed and unmanaged resources; false to release only unmanaged resources.
protected override void Dispose(bool disposing)
{
if (m_disposed)
return;
try
{
if (!disposing)
return;
if (m_connectionHandle != null)
{
m_connectionHandle.Dispose();
m_connectionHandle = null;
}
if (m_sendArgs != null)
{
m_sendArgs.Dispose();
m_sendArgs = null;
}
if (m_receiveArgs != null)
{
m_receiveArgs.Dispose();
m_receiveArgs = null;
}
}
finally
{
m_disposed = true; // Prevent duplicate dispose.
base.Dispose(disposing); // Call base class Dispose().
}
}
///
/// Validates the specified .
///
/// Connection string to be validated.
/// Port property is missing.
/// Port property value is not between and .
protected override void ValidateConnectionString(string connectionString)
{
m_connectData = connectionString.ParseKeyValuePairs();
// Derive desired IP stack based on specified "interface" setting, adding setting if it's not defined
m_ipStack = Transport.GetInterfaceIPStack(m_connectData);
// Backwards compatibility adjustments.
// New Format: Server=localhost:8888; Port=8989
// Old Format: Server=localhost; RemotePort=8888; LocalPort=8888
if (m_connectData.ContainsKey("localPort") && !m_connectData.ContainsKey("port"))
m_connectData.Add("port", m_connectData["localPort"]);
if (m_connectData.ContainsKey("server") && m_connectData.ContainsKey("remotePort"))
m_connectData["server"] = $"{m_connectData["server"]}:{m_connectData["remotePort"]}";
// Check if 'port' property is missing.
if (!m_connectData.ContainsKey("port"))
throw new ArgumentException($"Port property is missing (Example: {DefaultConnectionString})");
// Check if 'port' property is valid.
if (!Transport.IsPortNumberValid(m_connectData["port"]) && int.Parse(m_connectData["port"]) != -1)
throw new ArgumentOutOfRangeException(nameof(connectionString), $"Port number must be {-1} or between {Transport.PortRangeLow} and {Transport.PortRangeHigh}");
if (!m_connectData.ContainsKey("multicastTimeToLive"))
m_connectData.Add("multicastTimeToLive", "10");
// Make sure a valid multi-cast time-to-live value is defined in the connection string
if (!(m_connectData.TryGetValue("multicastTimeToLive", out string setting) && int.TryParse(setting, out _)))
m_connectData["multicastTimeToLive"] = "10";
}
///
/// Connects to the .
///
private void OpenPort()
{
int connectionAttempts = 0;
while (MaxConnectionAttempts == -1 || connectionAttempts < MaxConnectionAttempts)
{
try
{
OnConnectionAttempt();
m_udpClient.Provider = Transport.CreateSocket(m_connectData["interface"], int.Parse(m_connectData["port"]), ProtocolType.Udp, m_ipStack, AllowDualStackSocket);
// Disable SocketError.ConnectionReset exception from being thrown when the endpoint is not listening.
// Fixes MONO issue with SIO_UDP_CONNRESET
try
{
m_udpClient.Provider.IOControl(SIO_UDP_CONNRESET, new[] { Convert.ToByte(false) }, null);
}
catch (Exception ex)
{
Logger.SwallowException(ex, "Failed to set SIO_UDP_CONNRESET operating mode on UDP socket");
}
m_udpClient.Provider.ReceiveBufferSize = ReceiveBufferSize;
// If the IP specified for the server is a multicast IP, subscribe to the specified multicast group.
IPEndPoint serverEndpoint = m_udpServer;
if (Transport.IsMulticastIP(serverEndpoint.Address))
{
IPAddress sourceAddress = null;
if (m_connectData.TryGetValue("multicastSource", out string multicastSource))
sourceAddress = IPAddress.Parse(multicastSource);
AddMulticastMembership(serverEndpoint.Address, sourceAddress, out byte[] multicastMembershipAddresses);
m_udpClient.MulticastMembershipAddresses = multicastMembershipAddresses;
}
#if !MONO
// If the client requires packet info when
// receiving data, set the socket option now.
if (m_receivePacketInfo)
m_udpClient.Provider.SetSocketOption(serverEndpoint.AddressFamily == AddressFamily.InterNetworkV6 ? SocketOptionLevel.IPv6 : SocketOptionLevel.IP, SocketOptionName.PacketInformation, true);
#endif
// Listen for data to send.
using (SocketAsyncEventArgs sendArgs = m_sendArgs)
m_sendArgs = FastObjectFactory.CreateObjectFunction();
m_udpClient.SetSendBuffer(SendBufferSize);
m_sendArgs.SetBuffer(m_udpClient.SendBuffer, 0, m_udpClient.SendBufferSize);
m_sendArgs.Completed += m_sendHandler;
m_connectionHandle.Set();
OnConnectionEstablished();
// Listen for incoming data only if endpoint is bound to a local interface.
if (m_udpClient.Provider.LocalEndPoint != null)
{
using (SocketAsyncEventArgs receiveArgs = m_receiveArgs)
{
m_receiveArgs = FastObjectFactory.CreateObjectFunction();
}
try
{
Interlocked.Exchange(ref m_receiving, 1);
m_receiveArgs.Completed += m_receiveHandler;
ReceivePayloadAsync();
}
catch
{
Interlocked.Exchange(ref m_receiving, 0);
throw;
}
}
break;
}
catch (ThreadAbortException)
{
break;
}
catch (Exception ex)
{
connectionAttempts++;
OnConnectionException(ex);
}
}
}
///
/// Sends data to the server asynchronously.
///
/// The buffer that contains the binary data to be sent.
/// The zero-based position in the at which to begin sending data.
/// The number of bytes to be sent from starting at the .
/// for the asynchronous operation.
protected override WaitHandle SendDataAsync(byte[] data, int offset, int length) => SendDataToAsync(data, offset, length, m_udpServer);
///
/// Sends data to the server asynchronously.
///
/// The buffer that contains the binary data to be sent.
/// The zero-based position in the at which to begin sending data.
/// The number of bytes to be sent from starting at the .
/// The end point which identifies the destination for the data.
/// for the asynchronous operation.
public WaitHandle SendDataToAsync(byte[] data, int offset, int length, EndPoint destination)
{
if (CurrentState != ClientState.Connected)
throw new InvalidOperationException("Client is not connected");
// Execute operation to see if the client has reached the maximum send queue size.
m_dumpPayloadsOperation.TryRun();
// Create payload and wait handle.
UdpClientPayload payload = FastObjectFactory.CreateObjectFunction();
ManualResetEventSlim handle = FastObjectFactory.CreateObjectFunction();
payload.Destination = destination;
payload.Data = data;
payload.Offset = offset;
payload.Length = length;
payload.WaitHandle = handle;
handle.Reset();
// Queue payload for sending.
m_sendQueue.Enqueue(payload);
lock (m_sendLock)
{
// Send the next queued payload.
if (Interlocked.CompareExchange(ref m_sending, 1, 0) == 0)
{
if (m_sendQueue.TryDequeue(out UdpClientPayload dequeuedPayload))
ThreadPool.QueueUserWorkItem(state => SendPayload((UdpClientPayload)state), dequeuedPayload);
else
Interlocked.Exchange(ref m_sending, 0);
}
}
// Notify that the send operation has started.
OnSendDataStart();
// Return the async handle that can be used to wait for the async operation to complete.
return handle.WaitHandle;
}
///
/// Raises the event.
///
/// Exception to send to event.
protected override void OnConnectionException(Exception ex)
{
base.Disconnect();
base.OnConnectionException(ex);
}
///
/// Raises the event.
///
/// Exception to send to event.
protected override void OnSendDataException(Exception ex)
{
if (CurrentState != ClientState.Disconnected)
base.OnSendDataException(ex);
else
Logger.SwallowException(ex, "UdpClient.cs: The client state was disconnected");
}
///
/// Raises the event.
///
/// Exception to send to event.
protected override void OnReceiveDataException(Exception ex)
{
if (CurrentState != ClientState.Disconnected)
base.OnReceiveDataException(ex);
else
Logger.SwallowException(ex, "UdpClient.cs: The client state was disconnected");
}
///
/// Raises the event.
///
protected override void OnConnectionTerminated()
{
if (CurrentState != ClientState.Disconnected)
base.OnConnectionTerminated();
}
///
/// Sends a payload on the socket.
///
private void SendPayload(UdpClientPayload payload)
{
try
{
// Set the user token of the socket args.
m_sendArgs.UserToken = payload;
// Copy payload into send buffer.
int copyLength = Math.Min(payload.Length, m_udpClient.SendBufferSize);
Buffer.BlockCopy(payload.Data, payload.Offset, m_udpClient.SendBuffer, 0, copyLength);
// Set buffer and end point of send args.
m_sendArgs.SetBuffer(0, copyLength);
m_sendArgs.RemoteEndPoint = payload.Destination;
// Update payload offset and length.
payload.Offset += copyLength;
payload.Length -= copyLength;
// Send data over socket.
if (!m_udpClient.Provider.SendToAsync(m_sendArgs))
ProcessSend();
}
catch (Exception ex)
{
OnSendDataException(ex);
// Assume process send was not able
// to continue the asynchronous loop.
Interlocked.Exchange(ref m_sending, 0);
}
}
///
/// Callback method for asynchronous send operation.
///
private void ProcessSend()
{
UdpClientPayload payload = null;
try
{
// Get the payload and its wait handle.
payload = (UdpClientPayload)m_sendArgs.UserToken;
ManualResetEventSlim handle = payload.WaitHandle;
// Determine whether we are finished with this
// payload and, if so, set the wait handle.
if (payload.Length <= 0)
handle.Set();
// Check for errors during send operation.
if (m_sendArgs.SocketError != SocketError.Success)
throw new SocketException((int)m_sendArgs.SocketError);
// Update statistics.
UpdateBytesSent(m_sendArgs.BytesTransferred);
m_udpClient.Statistics.UpdateBytesSent(m_sendArgs.BytesTransferred);
// Send operation is complete.
if (payload.Length <= 0)
OnSendDataComplete();
}
catch (Exception ex)
{
// Send operation failed to complete.
OnSendDataException(ex);
}
finally
{
if (payload != null)
{
try
{
if (payload.Length > 0)
{
// Still more to send for this payload.
ThreadPool.QueueUserWorkItem(state => SendPayload((UdpClientPayload)state), payload);
}
else
{
payload.WaitHandle = null;
// Begin sending next client payload.
if (m_sendQueue.TryDequeue(out payload))
{
ThreadPool.QueueUserWorkItem(state => SendPayload((UdpClientPayload)state), payload);
}
else
{
lock (m_sendLock)
{
if (m_sendQueue.TryDequeue(out payload))
ThreadPool.QueueUserWorkItem(state => SendPayload((UdpClientPayload)state), payload);
else
Interlocked.Exchange(ref m_sending, 0);
}
}
}
}
catch (Exception ex)
{
string errorMessage = $"Exception encountered while attempting to send next payload: {ex.Message}";
OnSendDataException(new Exception(errorMessage, ex));
Interlocked.Exchange(ref m_sending, 0);
}
}
}
}
///
/// Initiate method for asynchronous receive operation of payload data.
///
private void ReceivePayloadAsync()
{
// Set up event args for receive operation.
if (m_udpClient.ReceiveBufferSize != m_receiveArgs.Count)
m_receiveArgs.SetBuffer(m_udpClient.ReceiveBuffer, 0, m_udpClient.ReceiveBufferSize);
else
m_receiveArgs.SetBuffer(0, m_udpClient.ReceiveBufferSize);
m_receiveArgs.RemoteEndPoint = m_udpServer;
if (!m_receivePacketInfo)
{
if (!m_udpClient.Provider.ReceiveFromAsync(m_receiveArgs))
ThreadPool.QueueUserWorkItem(state => ProcessReceive());
}
else
{
if (!m_udpClient.Provider.ReceiveMessageFromAsync(m_receiveArgs))
ThreadPool.QueueUserWorkItem(state => ProcessReceive());
}
}
///
/// Callback method for asynchronous receive operation of payload data.
///
private void ProcessReceive()
{
try
{
if (CurrentState == ClientState.Disconnected)
throw new SocketException((int)SocketError.Disconnecting);
if (m_receiveArgs.SocketError != SocketError.Success)
throw new SocketException((int)m_receiveArgs.SocketError);
// Update statistics and pointers.
UpdateBytesReceived(m_receiveArgs.BytesTransferred);
m_udpClient.Statistics.UpdateBytesReceived(m_receiveArgs.BytesTransferred);
m_udpClient.BytesReceived = m_udpClient.Statistics.LastBytesReceived;
// Notify of received data and resume receive operation.
OnReceive(m_receiveArgs.RemoteEndPoint, m_receiveArgs.ReceiveMessageFromPacketInfo, m_udpClient.ReceiveBuffer, m_udpClient.BytesReceived);
ReceivePayloadAsync();
}
catch (ObjectDisposedException)
{
// Make sure connection is terminated when client is disposed.
TerminateConnection(true);
}
catch (SocketException ex)
{
try
{
// Notify and resume receive.
OnReceiveDataException(ex);
ReceivePayloadAsync();
}
catch
{
// Terminate connection if resuming receiving fails.
TerminateConnection(true);
}
}
catch (Exception ex)
{
try
{
// For any other exception, notify and resume receive.
OnReceiveDataException(ex);
ReceivePayloadAsync();
}
catch
{
// Terminate connection if resuming receiving fails.
TerminateConnection(true);
}
}
}
private void AddMulticastMembership(IPAddress serverAddress, IPAddress sourceAddress, out byte[] multicastMembershipAddresses)
{
multicastMembershipAddresses = null;
try
{
if (!Transport.IsMulticastIP(serverAddress))
throw new InvalidOperationException("Cannot add multicast membership if server address is not a multicast IP.");
SocketOptionLevel level = serverAddress.AddressFamily == AddressFamily.InterNetworkV6 ? SocketOptionLevel.IPv6 : SocketOptionLevel.IP;
if (sourceAddress == null)
{
// Execute multicast subscribe for any source
m_udpClient.Provider.SetSocketOption(level, SocketOptionName.AddMembership, new MulticastOption(serverAddress));
m_udpClient.Provider.SetSocketOption(level, SocketOptionName.MulticastTimeToLive, int.Parse(m_connectData["multicastTimeToLive"]));
}
else
{
IPAddress localAddress = ((IPEndPoint)m_udpClient.Provider.LocalEndPoint).Address;
if (sourceAddress.AddressFamily != serverAddress.AddressFamily)
throw new InvalidOperationException($"Source address \"{sourceAddress}\" is not in the same IP format as server address \"{serverAddress}\"");
if (localAddress.AddressFamily != serverAddress.AddressFamily)
throw new InvalidOperationException($"Local address \"{localAddress}\" is not in the same IP format as server address \"{serverAddress}\"");
using (BlockAllocatedMemoryStream membershipAddresses = new BlockAllocatedMemoryStream())
{
byte[] serverAddressBytes = serverAddress.GetAddressBytes();
byte[] sourceAddressBytes = sourceAddress.GetAddressBytes();
byte[] localAddressBytes = localAddress.GetAddressBytes();
membershipAddresses.Write(serverAddressBytes, 0, serverAddressBytes.Length);
membershipAddresses.Write(sourceAddressBytes, 0, sourceAddressBytes.Length);
membershipAddresses.Write(localAddressBytes, 0, localAddressBytes.Length);
multicastMembershipAddresses = membershipAddresses.ToArray();
}
// Execute multicast subscribe for specific source
m_udpClient.Provider.SetSocketOption(level, SocketOptionName.AddSourceMembership, multicastMembershipAddresses);
m_udpClient.Provider.SetSocketOption(level, SocketOptionName.MulticastTimeToLive, int.Parse(m_connectData["multicastTimeToLive"]));
}
}
catch (SocketException ex)
{
// Need to add comment as to why this is necessary...
if (ex.SocketErrorCode != SocketError.InvalidArgument)
throw;
}
}
private void DropMulticastMembership(IPAddress serverAddress, IPAddress sourceAddress, byte[] multicastMembershipAddresses)
{
try
{
if (!Transport.IsMulticastIP(serverAddress))
throw new InvalidOperationException("Cannot drop multicast membership if server address is not a multicast IP.");
if (sourceAddress == null && multicastMembershipAddresses == null)
{
// Execute multicast unsubscribe for any source
m_udpClient.Provider.SetSocketOption(serverAddress.AddressFamily == AddressFamily.InterNetworkV6 ? SocketOptionLevel.IPv6 : SocketOptionLevel.IP, SocketOptionName.DropMembership, new MulticastOption(serverAddress));
}
else
{
if (multicastMembershipAddresses == null)
{
IPAddress localAddress = ((IPEndPoint)m_udpClient.Provider.LocalEndPoint).Address;
if (sourceAddress.AddressFamily != serverAddress.AddressFamily)
throw new InvalidOperationException($"Source address \"{sourceAddress}\" is not in the same IP format as server address \"{serverAddress}\"");
if (localAddress.AddressFamily != serverAddress.AddressFamily)
throw new InvalidOperationException($"Local address \"{localAddress}\" is not in the same IP format as server address \"{serverAddress}\"");
using (BlockAllocatedMemoryStream membershipAddresses = new BlockAllocatedMemoryStream())
{
byte[] serverAddressBytes = serverAddress.GetAddressBytes();
byte[] sourceAddressBytes = sourceAddress.GetAddressBytes();
byte[] localAddressBytes = localAddress.GetAddressBytes();
membershipAddresses.Write(serverAddressBytes, 0, serverAddressBytes.Length);
membershipAddresses.Write(sourceAddressBytes, 0, sourceAddressBytes.Length);
membershipAddresses.Write(localAddressBytes, 0, localAddressBytes.Length);
multicastMembershipAddresses = membershipAddresses.ToArray();
}
}
// Execute multicast unsubscribe for specific source
m_udpClient.Provider.SetSocketOption(serverAddress.AddressFamily == AddressFamily.InterNetworkV6 ? SocketOptionLevel.IPv6 : SocketOptionLevel.IP, SocketOptionName.DropSourceMembership, multicastMembershipAddresses);
}
}
catch (SocketException ex)
{
if (ex.SocketErrorCode != SocketError.InvalidArgument)
throw;
}
}
///
/// Dumps payloads from the send queue when the send queue grows too large.
///
private void DumpPayloads()
{
// Check to see if the client has reached the maximum send queue size.
if (MaxSendQueueSize <= 0 || m_sendQueue.Count < MaxSendQueueSize)
return;
for (int i = 0; i < MaxSendQueueSize; i++)
{
if (m_sendQueue.TryDequeue(out UdpClientPayload payload))
{
payload.WaitHandle.Set();
payload.WaitHandle.Dispose();
payload.WaitHandle = null;
}
}
throw new InvalidOperationException($"UDP client reached maximum send queue size. {MaxSendQueueSize} payloads dumped from the queue.");
}
///
/// Processes the termination of client.
///
private void TerminateConnection(bool raiseEvent)
{
if (CurrentState != ClientState.Disconnected)
m_udpClient.Reset();
Interlocked.Exchange(ref m_receiving, 0);
if (raiseEvent)
OnConnectionTerminated();
}
///
/// Calls all the receive handlers in sequence.
///
private void OnReceive(EndPoint remoteEndPoint, IPPacketInformation packetInformation, byte[] data, int size)
{
OnReceiveDataFrom(remoteEndPoint, packetInformation, size);
OnReceiveDataFromComplete(remoteEndPoint, packetInformation, data, size);
OnReceiveDataComplete(data, size);
}
///
/// Raises the event.
///
/// End-point from which data has been received.
/// Packet information for received data, if available.
/// Number of bytes received from the client.
private void OnReceiveDataFrom(EndPoint remoteEndPoint, IPPacketInformation packetInformation, int size)
{
try
{
ReceiveDataFrom?.Invoke(this, new EventArgs(remoteEndPoint, packetInformation, size));
}
catch (Exception ex)
{
OnUnhandledUserException(ex);
}
}
///
/// Raises the event.
///
/// End-point from which data has been received.
/// Packet information for received data, if available.
/// Data received from the client.
/// Number of bytes received from the client.
private void OnReceiveDataFromComplete(EndPoint remoteEndPoint, IPPacketInformation packetInformation, byte[] data, int size)
{
try
{
ReceiveDataFromComplete?.Invoke(this, new EventArgs(remoteEndPoint, packetInformation, data, size));
}
catch (Exception ex)
{
OnUnhandledUserException(ex);
}
}
#endregion
}
}