//******************************************************************************************************
// TcpServer.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/02/2006 - Pinal C. Patel
// Original version of source code generated.
// 09/06/2006 - J. Ritchie Carroll
// Added bypass optimizations for high-speed socket access.
// 12/01/2006 - Pinal C. Patel
// Modified code for handling "PayloadAware" transmissions.
// 01/28/3008 - J. Ritchie Carroll
// Placed accepted TCP socket connections on their own threads instead of thread pool.
// 09/29/2008 - J. Ritchie Carroll
// Converted to C#.
// 07/17/2009 - Pinal C. Patel
// Added support to specify a specific interface address on a multiple interface machine.
// 09/14/2009 - Stephen C. Wills
// Added new header and license agreement.
// 10/14/2009 - Pinal C. Patel
// Added null reference check to DisconnectOne() for safety.
// 02/11/2011 - Pinal C. Patel
// Added IntegratedSecurity property to enable integrated windows authentication.
// 09/21/2011 - J. Ritchie Carroll
// Added Mono implementation exception regions.
// 12/04/2011 - J. Ritchie Carroll
// Modified to use concurrent dictionary.
// 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.
//
//******************************************************************************************************
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.ComponentModel;
using System.Diagnostics.CodeAnalysis;
using System.Linq;
using System.Net;
using System.Net.Sockets;
using System.Security.Principal;
using System.Text;
using System.Threading;
#if !MONO
using System.Net.Security;
using System.Security.Authentication;
#endif
using GSF.Configuration;
using GSF.Threading;
namespace GSF.Communication
{
///
/// Represents a TCP-based communication server.
///
///
/// The socket can be bound to a specified interface on a machine with multiple interfaces by
/// specifying the interface in the (Example: "Port=8888; Interface=127.0.0.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 TcpServer m_server;
///
/// static void Main(string[] args)
/// {
/// // Initialize the server.
/// m_server = new TcpServer("Port=8888");
/// m_server.Handshake = false;
/// m_server.PayloadAware = false;
/// m_server.ReceiveTimeout = -1;
/// m_server.Encryption = CipherStrength.None;
/// m_server.Compression = CompressionStrength.NoCompression;
/// m_server.SecureSession = false;
/// m_server.Initialize();
/// // Register event handlers.
/// m_server.ServerStarted += m_server_ServerStarted;
/// m_server.ServerStopped += m_server_ServerStopped;
/// m_server.ClientConnected += m_server_ClientConnected;
/// m_server.ClientDisconnected += m_server_ClientDisconnected;
/// m_server.ReceiveClientDataComplete += m_server_ReceiveClientDataComplete;
/// // Start the server.
/// m_server.Start();
///
/// // Multicast user input to all connected clients.
/// string input;
/// while (string.Compare(input = Console.ReadLine(), "Exit", true) != 0)
/// {
/// m_server.Multicast(input);
/// }
///
/// // Stop the server on shutdown.
/// m_server.Stop();
/// }
///
/// static void m_server_ServerStarted(object sender, EventArgs e)
/// {
/// Console.WriteLine("Server has been started!");
/// }
///
/// static void m_server_ServerStopped(object sender, EventArgs e)
/// {
/// Console.WriteLine("Server has been stopped!");
/// }
///
/// static void m_server_ClientConnected(object sender, EventArgs<Guid> e)
/// {
/// Console.WriteLine(string.Format("Client connected - {0}.", e.Argument));
/// }
///
/// static void m_server_ClientDisconnected(object sender, EventArgs<Guid> e)
/// {
/// Console.WriteLine(string.Format("Client disconnected - {0}.", e.Argument));
/// }
///
/// static void m_server_ReceiveClientDataComplete(object sender, EventArgs<Guid, byte[], int> e)
/// {
/// Console.WriteLine(string.Format("Received data from {0} - {1}.", e.Argument1, m_server.TextEncoding.GetString(e.Argument2, 0, e.Argument3)));
/// }
/// }
///
///
public class TcpServer : ServerBase
{
#region [ Members ]
// Nested Types
private class TcpClientInfo
{
public TransportProvider Client;
public SocketAsyncEventArgs SendArgs;
public object SendLock;
public ConcurrentQueue SendQueue;
public ShortSynchronizedOperation DumpPayloadsOperation;
public int Sending;
public WindowsPrincipal ClientPrincipal;
}
private class TcpServerPayload
{
// Per payload state
public byte[] Data;
public int Offset;
public int Length;
public ManualResetEventSlim WaitHandle;
// Per client state
public TcpClientInfo ClientInfo;
}
// Constants
///
/// Specifies the default value for the property.
///
public const bool DefaultPayloadAware = false;
///
/// Specifies the default value for the property.
///
public const bool DefaultIntegratedSecurity = false;
///
/// Specifies the default value for the property.
///
public const bool DefaultIgnoreInvalidCredentials = false;
///
/// 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 bool DefaultNoDelay = false;
///
/// Specifies the default value for the property.
///
public const string DefaultConfigurationString = "Port=8888";
// Fields
private byte[] m_payloadMarker;
private EndianOrder m_payloadEndianOrder;
private IPStack m_ipStack;
private SocketAsyncEventArgs m_acceptArgs;
private readonly ConcurrentDictionary m_clientInfoLookup;
private Dictionary m_configData;
private readonly EventHandler m_acceptHandler;
private readonly EventHandler m_sendHandler;
private readonly EventHandler m_receivePayloadAwareHandler;
private readonly EventHandler m_receivePayloadUnawareHandler;
#endregion
#region [ Constructors ]
///
/// Initializes a new instance of the class.
///
public TcpServer() : this(DefaultConfigurationString)
{
}
///
/// Initializes a new instance of the class.
///
/// Config string of the . See for format.
public TcpServer(string configString) : base(TransportProtocol.Tcp, configString)
{
PayloadAware = DefaultPayloadAware;
m_payloadMarker = Payload.DefaultMarker;
m_payloadEndianOrder = EndianOrder.LittleEndian;
IntegratedSecurity = DefaultIntegratedSecurity;
IgnoreInvalidCredentials = DefaultIgnoreInvalidCredentials;
AllowDualStackSocket = DefaultAllowDualStackSocket;
MaxSendQueueSize = DefaultMaxSendQueueSize;
NoDelay = DefaultNoDelay;
m_clientInfoLookup = new ConcurrentDictionary();
m_acceptHandler = (sender, args) => ProcessAccept(args);
m_sendHandler = (sender, args) => ProcessSend(args);
m_receivePayloadAwareHandler = (sender, args) => ProcessReceivePayloadAware(args);
m_receivePayloadUnawareHandler = (sender, args) => ProcessReceivePayloadUnaware(args);
}
///
/// Initializes a new instance of the class.
///
/// object that contains the .
public TcpServer(IContainer container) : this() => container?.Add(this);
#endregion
#region [ Properties ]
///
/// Gets or sets a boolean value that indicates whether the payload boundaries are to be preserved during transmission.
///
[Category("Data")]
[DefaultValue(DefaultPayloadAware)]
[Description("Indicates whether the payload boundaries are to be preserved during transmission.")]
public bool PayloadAware { get; set; }
///
/// Gets or sets the byte sequence used to mark the beginning of a payload in a transmission.
///
///
/// Setting property to null will create a zero-length payload marker.
///
[Browsable(false)]
[DesignerSerializationVisibility(DesignerSerializationVisibility.Hidden)]
public byte[] PayloadMarker
{
get => m_payloadMarker;
set => m_payloadMarker = value ?? Array.Empty();
}
///
/// Gets or sets the endian order to apply for encoding and decoding payload size in a transmission.
///
///
/// Setting property to null will force use of little-endian encoding.
///
[Browsable(false)]
[DesignerSerializationVisibility(DesignerSerializationVisibility.Hidden)]
public EndianOrder PayloadEndianOrder
{
get => m_payloadEndianOrder;
set => m_payloadEndianOrder = value ?? EndianOrder.LittleEndian;
}
///
/// Gets or sets a boolean value that indicates whether the client Windows account credentials are used for authentication.
///
///
/// This option is ignored under Mono deployments.
///
[Category("Security")]
[DefaultValue(DefaultIntegratedSecurity)]
[Description("Indicates whether the client Windows account credentials are used for authentication.")]
public bool IntegratedSecurity { get; set; }
///
/// Gets or sets a boolean value that indicates whether the server
/// should ignore errors when the client's credentials are invalid.
///
///
/// This property should only be set to true if there is an alternative by which
/// to authenticate the client when integrated security fails. When this is set
/// to true, if the client's credentials are invalid, the
/// method will return true for that client, but the principal will still be null.
///
[Category("Security")]
[DefaultValue(DefaultIgnoreInvalidCredentials)]
[Description("Indicates whether the client Windows account credentials are validated during authentication.")]
public bool IgnoreInvalidCredentials { get; set; }
///
/// 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 or sets a boolean value that determines if small packets are delivered to the remote host without delay.
///
[Category("Settings")]
[DefaultValue(DefaultNoDelay)]
[Description("Determines if small packets are delivered to the remote host without delay.")]
public bool NoDelay { get; set; }
///
/// Gets the object for the .
///
[Browsable(false)]
public Socket Server { get; private set; }
///
/// Gets the receive handler used to handle data received on the connected sockets.
///
private EventHandler ReceiveHandler => PayloadAware ? m_receivePayloadAwareHandler : m_receivePayloadUnawareHandler;
///
/// Gets the descriptive status of the server.
///
public override string Status
{
get
{
StringBuilder statusBuilder = new StringBuilder(base.Status);
int count = 0;
foreach (ConcurrentQueue sendQueue in m_clientInfoLookup.Values.Select(clientInfo => clientInfo.SendQueue))
{
statusBuilder.AppendFormat(" Queued payloads: {0} for client {1}", sendQueue.Count, ++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.
///
/// ID of the client from which data buffer should be read.
/// 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 -or-
/// Specified does not exist, cannot read buffer.
///
/// is null.
///
/// or is less than 0 -or-
/// and will exceed length.
///
public override int Read(Guid clientID, byte[] buffer, int startIndex, int length)
{
buffer.ValidateParameters(startIndex, length);
if (!m_clientInfoLookup.TryGetValue(clientID, out TcpClientInfo clientInfo))
throw new InvalidOperationException("Specified client ID does not exist, cannot read buffer.");
TransportProvider tcpClient = clientInfo.Client;
if (tcpClient.ReceiveBuffer == null)
throw new InvalidOperationException("No received data buffer has been defined to read.");
int readIndex = ReadIndicies[clientID];
int sourceLength = tcpClient.BytesReceived - readIndex;
int readBytes = length > sourceLength ? sourceLength : length;
Buffer.BlockCopy(tcpClient.ReceiveBuffer, readIndex, buffer, startIndex, readBytes);
// Update read index for next call
readIndex += readBytes;
if (readIndex >= tcpClient.BytesReceived)
readIndex = 0;
ReadIndicies[clientID] = readIndex;
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["PayloadAware", true].Update(PayloadAware);
settings["IntegratedSecurity", true].Update(IntegratedSecurity);
settings["AllowDualStackSocket", true].Update(AllowDualStackSocket);
settings["MaxSendQueueSize", true].Update(MaxSendQueueSize);
settings["NoDelay", true].Update(NoDelay);
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("PayloadAware", PayloadAware, "True if payload boundaries are to be preserved during transmission, otherwise False.");
settings.Add("IntegratedSecurity", IntegratedSecurity, "True if the client Windows account credentials are used for authentication, otherwise False.");
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.");
settings.Add("NoDelay", NoDelay, "True to disable Nagle so that small packets are delivered to the remote host without delay, otherwise False.");
PayloadAware = settings["PayloadAware"].ValueAs(PayloadAware);
IntegratedSecurity = settings["IntegratedSecurity"].ValueAs(IntegratedSecurity);
AllowDualStackSocket = settings["AllowDualStackSocket"].ValueAs(AllowDualStackSocket);
MaxSendQueueSize = settings["MaxSendQueueSize"].ValueAs(MaxSendQueueSize);
NoDelay = settings["NoDelay"].ValueAs(NoDelay);
}
///
/// Stops the synchronously and disconnects all connected clients.
///
public override void Stop()
{
SocketAsyncEventArgs acceptArgs = m_acceptArgs;
m_acceptArgs = null;
if (CurrentState != ServerState.Running)
return;
DisconnectAll(); // Disconnection all clients.
Server.Close(); // Stop accepting new connections.
// Clean up accept args.
acceptArgs?.Dispose();
OnServerStopped();
}
///
/// Starts the synchronously and begins accepting client connections asynchronously.
///
/// Attempt is made to the when it is running.
public override void Start()
{
if (CurrentState != ServerState.NotRunning)
return;
// Initialize if unitialized.
if (!Initialized)
Initialize();
// Overwrite config file if integrated security exists in connection string.
if (m_configData.TryGetValue("integratedSecurity", out string integratedSecuritySetting))
IntegratedSecurity = integratedSecuritySetting.ParseBoolean();
#if MONO
// Force integrated security to be False under Mono since it's not supported
m_integratedSecurity = false;
#endif
// Overwrite config file if max client connections exists in connection string.
if (m_configData.ContainsKey("maxClientConnections") && int.TryParse(m_configData["maxClientConnections"], out int maxClientConnections))
MaxClientConnections = maxClientConnections;
// Overwrite config file if max send queue size exists in connection string.
if (m_configData.ContainsKey("maxSendQueueSize") && int.TryParse(m_configData["maxSendQueueSize"], out int maxSendQueueSize))
MaxSendQueueSize = maxSendQueueSize;
// Overwrite config file if no delay exists in connection string.
if (m_configData.TryGetValue("noDelay", out string noDelaySetting))
NoDelay = noDelaySetting.ParseBoolean();
// Bind server socket to local end-point and listen.
Server = Transport.CreateSocket(m_configData["interface"], int.Parse(m_configData["port"]), ProtocolType.Tcp, m_ipStack, AllowDualStackSocket);
Server.NoDelay = NoDelay;
Server.Listen(1);
// Begin accepting incoming connection asynchronously.
m_acceptArgs = FastObjectFactory.CreateObjectFunction();
m_acceptArgs.AcceptSocket = null;
m_acceptArgs.SetBuffer(null, 0, 0);
m_acceptArgs.SocketFlags = SocketFlags.None;
m_acceptArgs.Completed += m_acceptHandler;
if (!Server.AcceptAsync(m_acceptArgs))
ThreadPool.QueueUserWorkItem(state => ProcessAccept((SocketAsyncEventArgs)state), m_acceptArgs);
// Notify that the server has been started successfully.
OnServerStarted();
}
///
/// Disconnects the specified connected client.
///
/// ID of the client to be disconnected.
/// Client does not exist for the specified .
public override void DisconnectOne(Guid clientID)
{
if (!TryGetClient(clientID, out TransportProvider client))
return;
try
{
if (client.Provider != null && client.Provider.Connected)
client.Provider.Disconnect(false);
OnClientDisconnected(clientID);
client.Reset();
}
catch (Exception ex)
{
OnSendClientDataException(client.ID, new InvalidOperationException($"Client disconnection exception: {ex.Message}", ex));
}
}
///
/// Gets the object associated with the specified client ID.
///
/// ID of the client.
/// The TCP client.
/// A object.
public bool TryGetClient(Guid clientID, out TransportProvider tcpClient)
{
bool clientExists = m_clientInfoLookup.TryGetValue(clientID, out TcpClientInfo clientInfo);
tcpClient = clientExists ? clientInfo.Client : null;
return clientExists;
}
///
/// Gets the object associated with the specified client ID.
///
/// ID of the client.
/// The principal of the client.
/// A object.
public bool TryGetClientPrincipal(Guid clientID, out WindowsPrincipal clientPrincipal)
{
bool clientExists = m_clientInfoLookup.TryGetValue(clientID, out TcpClientInfo clientInfo);
clientPrincipal = clientExists ? clientInfo.ClientPrincipal : null;
return clientExists;
}
///
/// Validates the specified .
///
/// Configuration string to be validated.
/// Port property is missing.
/// Port property value is not between and .
protected override void ValidateConfigurationString(string configurationString)
{
m_configData = configurationString.ParseKeyValuePairs();
// Derive desired IP stack based on specified "interface" setting, adding setting if it's not defined
m_ipStack = Transport.GetInterfaceIPStack(m_configData);
if (!m_configData.ContainsKey("port"))
throw new ArgumentException($"Port property is missing (Example: {DefaultConfigurationString})");
if (!Transport.IsPortNumberValid(m_configData["port"]))
throw new ArgumentOutOfRangeException(nameof(configurationString), $"Port number must be between {Transport.PortRangeLow} and {Transport.PortRangeHigh}");
}
///
/// Sends data to the specified client asynchronously.
///
/// ID of the client to which the data is to be sent.
/// 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 SendDataToAsync(Guid clientID, byte[] data, int offset, int length)
{
if (!m_clientInfoLookup.TryGetValue(clientID, out TcpClientInfo clientInfo))
throw new InvalidOperationException($"No client found for ID {clientID}.");
ConcurrentQueue sendQueue = clientInfo.SendQueue;
// Execute operation to see if the client has reached the maximum send queue size.
clientInfo.DumpPayloadsOperation.TryRun();
// Prepare for payload-aware transmission.
if (PayloadAware)
Payload.AddHeader(ref data, ref offset, ref length, m_payloadMarker, m_payloadEndianOrder);
// Create payload and wait handle.
TcpServerPayload payload = FastObjectFactory.CreateObjectFunction();
ManualResetEventSlim handle = FastObjectFactory.CreateObjectFunction();
payload.Data = data;
payload.Offset = offset;
payload.Length = length;
payload.WaitHandle = handle;
payload.ClientInfo = clientInfo;
handle.Reset();
// Queue payload for sending.
sendQueue.Enqueue(payload);
lock (clientInfo.SendLock)
{
// Send next queued payload.
if (Interlocked.CompareExchange(ref clientInfo.Sending, 1, 0) == 0)
{
if (sendQueue.TryDequeue(out TcpServerPayload dequeuedPayload))
ThreadPool.QueueUserWorkItem(state => SendPayload((TcpServerPayload)state), dequeuedPayload);
else
Interlocked.Exchange(ref clientInfo.Sending, 0);
}
}
// Notify that the send operation has started.
OnSendClientDataStart(clientID);
// Return the async handle that can be used to wait for the async operation to complete.
return handle.WaitHandle;
}
///
/// Raises the event.
///
/// ID of client to send to event.
/// Exception to send to event.
protected override void OnSendClientDataException(Guid clientID, Exception ex)
{
if (m_clientInfoLookup.ContainsKey(clientID) && CurrentState == ServerState.Running)
base.OnSendClientDataException(clientID, ex);
}
///
/// Raises the event.
///
/// ID of client to send to event.
/// Exception to send to event.
protected virtual void OnReceiveClientDataException(Guid clientID, SocketException ex)
{
if (m_clientInfoLookup.ContainsKey(clientID) && ex.SocketErrorCode != SocketError.Disconnecting)
OnReceiveClientDataException(clientID, (Exception)ex);
}
///
/// Raises the event.
///
/// ID of client to send to event.
/// Exception to send to event.
protected override void OnReceiveClientDataException(Guid clientID, Exception ex)
{
if (m_clientInfoLookup.ContainsKey(clientID) && CurrentState == ServerState.Running)
base.OnReceiveClientDataException(clientID, ex);
}
///
/// Callback method for asynchronous accept operation.
///
[SuppressMessage("Microsoft.Usage", "CA2202:Do not dispose objects multiple times")]
private void ProcessAccept(SocketAsyncEventArgs acceptArgs)
{
TransportProvider client = new TransportProvider();
SocketAsyncEventArgs receiveArgs = null;
WindowsPrincipal clientPrincipal = null;
IPEndPoint remoteEndPoint = null;
TcpClientInfo clientInfo;
try
{
if (CurrentState == ServerState.NotRunning)
return;
// If acceptArgs was disposed, m_acceptArgs will either
// be null or another instance of SocketAsyncEventArgs.
// This check will tell us whether it's been disposed.
if (acceptArgs != m_acceptArgs)
return;
SocketError error = acceptArgs.SocketError;
client.Provider = acceptArgs.AcceptSocket;
if (error == SocketError.Success || error == SocketError.ConnectionReset)
{
// Return to accepting new connections.
acceptArgs.AcceptSocket = null;
if (!Server.AcceptAsync(acceptArgs))
ThreadPool.QueueUserWorkItem(state => ProcessAccept(acceptArgs));
}
else
{
// For unrecoverable errors, we need to ensure the server
// will be restarted before we can throw the error.
if (error != SocketError.ConnectionReset)
ThreadPool.QueueUserWorkItem(state => ReStart());
}
if (error != SocketError.Success)
throw new SocketException((int)error);
// Process the newly connected client.
client.Provider.ReceiveBufferSize = ReceiveBufferSize;
remoteEndPoint = client.Provider.RemoteEndPoint as IPEndPoint;
// Set up SocketAsyncEventArgs for receive operations.
receiveArgs = FastObjectFactory.CreateObjectFunction();
receiveArgs.Completed += ReceiveHandler;
#if !MONO
// Authenticate the connected client Windows credentials.
if (IntegratedSecurity)
{
NetworkStream socketStream = null;
NegotiateStream authenticationStream = null;
ICancellationToken timeoutToken = null;
try
{
socketStream = new NetworkStream(client.Provider, false);
authenticationStream = new NegotiateStream(socketStream, true);
timeoutToken = new Action(() => client.Provider?.Dispose()).DelayAndExecute(15000);
authenticationStream.AuthenticateAsServer();
if (!timeoutToken.Cancel())
throw new SocketException((int)SocketError.TimedOut);
if (authenticationStream.RemoteIdentity is WindowsIdentity)
clientPrincipal = new WindowsPrincipal((WindowsIdentity)authenticationStream.RemoteIdentity);
}
catch (InvalidCredentialException)
{
if (!IgnoreInvalidCredentials)
throw;
}
finally
{
timeoutToken?.Cancel();
socketStream?.Dispose();
authenticationStream?.Dispose();
}
}
#endif
if (MaxClientConnections != -1 && ClientIDs.Length >= MaxClientConnections)
{
// Reject client connection since limit has been reached.
TerminateConnection(client, receiveArgs, false);
}
else
{
// We can proceed further with receiving data from the client.
clientInfo = new TcpClientInfo
{
Client = client,
SendArgs = FastObjectFactory.CreateObjectFunction(),
SendLock = new object(),
SendQueue = new ConcurrentQueue(),
ClientPrincipal = clientPrincipal
};
// Create operation to dump send queue payloads when the queue grows too large.
clientInfo.DumpPayloadsOperation = new ShortSynchronizedOperation(() =>
{
// Check to see if the client has reached the maximum send queue size.
if (MaxSendQueueSize <= 0 || clientInfo.SendQueue.Count < MaxSendQueueSize)
return;
for (int i = 0; i < MaxSendQueueSize; i++)
{
if (clientInfo.SendQueue.TryDequeue(out TcpServerPayload payload))
{
payload.WaitHandle.Set();
payload.WaitHandle.Dispose();
payload.WaitHandle = null;
}
}
throw new InvalidOperationException($"Client {clientInfo.Client.ID} connected to TCP server reached maximum send queue size. {MaxSendQueueSize} payloads dumped from the queue.");
}, ex => OnSendClientDataException(clientInfo.Client.ID, ex));
// Set up socket args.
client.SetSendBuffer(SendBufferSize);
clientInfo.SendArgs.Completed += m_sendHandler;
clientInfo.SendArgs.SetBuffer(client.SendBuffer, 0, client.SendBufferSize);
m_clientInfoLookup.TryAdd(client.ID, clientInfo);
OnClientConnected(client.ID);
if (!PayloadAware)
{
receiveArgs.UserToken = client;
}
else
{
EventArgs, bool> userToken = FastObjectFactory, bool>>.CreateObjectFunction();
userToken.Argument1 = client;
receiveArgs.UserToken = userToken;
}
ReceivePayloadAsync(client, receiveArgs);
}
}
catch (ObjectDisposedException)
{
// m_acceptArgs may be disposed while in the middle of accepting a connection
}
catch (Exception ex)
{
// Notify of the exception.
string clientAddress = remoteEndPoint?.Address.ToString() ?? "UNKNOWN";
string errorMessage = $"Unable to accept connection to client [{clientAddress}]: {ex.Message}";
OnClientConnectingException(new Exception(errorMessage, ex));
if (receiveArgs != null)
TerminateConnection(client, receiveArgs, false);
}
}
///
/// Asynchronous loop sends payloads on the socket.
///
private void SendPayload(TcpServerPayload payload)
{
TcpClientInfo clientInfo = null;
TransportProvider client = null;
//ManualResetEventSlim handle;
try
{
clientInfo = payload.ClientInfo;
client = clientInfo.Client;
SocketAsyncEventArgs args = clientInfo.SendArgs;
args.UserToken = payload;
//handle = payload.WaitHandle;
// Copy payload into send buffer.
int copyLength = Math.Min(payload.Length, client.SendBufferSize);
Buffer.BlockCopy(payload.Data, payload.Offset, client.SendBuffer, 0, copyLength);
// Set buffer and user token of send args.
args.SetBuffer(0, copyLength);
// Update offset and length.
payload.Offset += copyLength;
payload.Length -= copyLength;
// Send data over socket.
if (!client.Provider.SendAsync(args))
ProcessSend(args);
}
catch (Exception ex)
{
if (client != null)
OnSendClientDataException(client.ID, ex);
if (clientInfo != null)
{
// Assume process send was not able
// to continue the asynchronous loop.
Interlocked.Exchange(ref clientInfo.Sending, 0);
}
}
}
///
/// Callback method for asynchronous send operation.
///
private void ProcessSend(SocketAsyncEventArgs args)
{
TcpServerPayload payload = null;
TcpClientInfo clientInfo = null;
TransportProvider client = null;
ConcurrentQueue sendQueue = null;
try
{
payload = (TcpServerPayload)args.UserToken;
clientInfo = payload.ClientInfo;
client = clientInfo.Client;
sendQueue = clientInfo.SendQueue;
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 (args.SocketError != SocketError.Success)
throw new SocketException((int)args.SocketError);
// Update statistics on the client.
client.Statistics.UpdateBytesSent(args.BytesTransferred);
// Send operation is complete.
if (payload.Length <= 0)
OnSendClientDataComplete(client.ID);
}
catch (Exception ex)
{
// Send operation failed to complete.
if (client != null)
OnSendClientDataException(client.ID, ex);
}
finally
{
if (payload != null)
{
try
{
if (payload.Length > 0)
{
// Still more to send for this payload.
ThreadPool.QueueUserWorkItem(state => SendPayload((TcpServerPayload)state), payload);
}
else if (sendQueue != null)
{
payload.WaitHandle = null;
payload.ClientInfo = null;
// Begin sending next client payload.
if (sendQueue.TryDequeue(out payload))
{
ThreadPool.QueueUserWorkItem(state => SendPayload((TcpServerPayload)state), payload);
}
else if (clientInfo != null)
{
lock (clientInfo.SendLock)
{
if (sendQueue.TryDequeue(out payload))
ThreadPool.QueueUserWorkItem(state => SendPayload((TcpServerPayload)state), payload);
else
Interlocked.Exchange(ref clientInfo.Sending, 0);
}
}
}
}
catch (Exception ex)
{
string errorMessage = $"Exception encountered while attempting to send next payload: {ex.Message}";
if (client != null)
OnSendClientDataException(client.ID, new Exception(errorMessage, ex));
if (clientInfo != null)
Interlocked.Exchange(ref clientInfo.Sending, 0);
}
}
}
}
///
/// Initiate method for asynchronous receive operation of payload data.
///
private void ReceivePayloadAsync(TransportProvider client, SocketAsyncEventArgs args)
{
// Initialize bytes received.
client.BytesReceived = 0;
// Initiate receiving.
if (PayloadAware)
{
// Set user token to indicate we are waiting for payload header.
EventArgs, bool> userToken = (EventArgs, bool>)args.UserToken;
userToken.Argument2 = true;
// Payload boundaries are to be preserved.
client.SetReceiveBuffer(m_payloadMarker.Length + Payload.LengthSegment);
ReceivePayloadAwareAsync(client, args);
}
else
{
// Payload boundaries are not to be preserved.
client.SetReceiveBuffer(ReceiveBufferSize);
ReceivePayloadUnawareAsync(client, args);
}
}
///
/// Initiate method for asynchronous receive operation of payload data in "payload-aware" mode.
///
private void ReceivePayloadAwareAsync(TransportProvider client, SocketAsyncEventArgs args)
{
byte[] buffer = client.ReceiveBuffer;
int offset = client.BytesReceived;
int length = client.ReceiveBufferSize - offset;
args.SetBuffer(buffer, offset, length);
if (!client.Provider.ReceiveAsync(args))
ThreadPool.QueueUserWorkItem(state => ProcessReceivePayloadAware((SocketAsyncEventArgs)state), args);
}
///
/// Callback method for asynchronous receive operation of payload data in "payload-aware" mode.
///
private void ProcessReceivePayloadAware(SocketAsyncEventArgs args)
{
EventArgs, bool> userToken = (EventArgs, bool>)args.UserToken;
TransportProvider client = userToken.Argument1;
try
{
if (args.SocketError != SocketError.Success)
throw new SocketException((int)args.SocketError);
if (args.BytesTransferred == 0)
throw new SocketException((int)SocketError.Disconnecting);
// Update statistics and pointers.
client.Statistics.UpdateBytesReceived(args.BytesTransferred);
client.BytesReceived += args.BytesTransferred;
if (userToken.Argument2)
{
// We're waiting on the payload length, so we'll check if the received data has this information.
int payloadLength = Payload.ExtractLength(client.ReceiveBuffer, client.BytesReceived, m_payloadMarker, m_payloadEndianOrder);
// We have the payload length.
// If it is set to zero, there is no payload; wait for another header.
// Otherwise we'll create a buffer that's big enough to hold the entire payload.
if (payloadLength == 0)
{
client.BytesReceived = 0;
}
else if (payloadLength != -1)
{
client.BytesReceived = 0;
client.SetReceiveBuffer(payloadLength);
userToken.Argument2 = false;
}
ReceivePayloadAwareAsync(client, args);
}
else
{
// We're accumulating the payload in the receive buffer until the entire payload is received.
if (client.BytesReceived == client.ReceiveBufferSize)
{
// We've received the entire payload.
OnReceiveClientDataComplete(client.ID, client.ReceiveBuffer, client.BytesReceived);
ReceivePayloadAsync(client, args);
}
else
{
// We've not yet received the entire payload.
ReceivePayloadAwareAsync(client, args);
}
}
}
catch (ObjectDisposedException)
{
// Make sure connection is terminated when server is disposed.
TerminateConnection(client, args, true);
}
catch (SocketException ex)
{
// Terminate connection when socket exception is encountered.
OnReceiveClientDataException(client.ID, ex);
TerminateConnection(client, args, true);
}
catch (Exception ex)
{
try
{
// For any other exception, notify and resume receive.
OnReceiveClientDataException(client.ID, ex);
ReceivePayloadAsync(client, args);
}
catch
{
// Terminate connection if resuming receiving fails.
TerminateConnection(client, args, true);
}
}
}
///
/// Initiate method for asynchronous receive operation of payload data in "payload-unaware" mode.
///
private void ReceivePayloadUnawareAsync(TransportProvider client, SocketAsyncEventArgs args)
{
byte[] buffer = client.ReceiveBuffer;
int length = client.ReceiveBufferSize;
args.SetBuffer(buffer, 0, length);
if (!client.Provider.ReceiveAsync(args))
ThreadPool.QueueUserWorkItem(state => ProcessReceivePayloadUnaware((SocketAsyncEventArgs)state), args);
}
///
/// Callback method for asynchronous receive operation of payload data in "payload-unaware" mode.
///
private void ProcessReceivePayloadUnaware(SocketAsyncEventArgs args)
{
TransportProvider client = (TransportProvider)args.UserToken;
try
{
if (args.SocketError != SocketError.Success)
throw new SocketException((int)args.SocketError);
if (args.BytesTransferred == 0)
throw new SocketException((int)SocketError.Disconnecting);
// Update statistics and pointers.
client.Statistics.UpdateBytesReceived(args.BytesTransferred);
client.BytesReceived = args.BytesTransferred;
// Notify of received data and resume receive operation.
OnReceiveClientDataComplete(client.ID, client.ReceiveBuffer, client.BytesReceived);
ReceivePayloadUnawareAsync(client, args);
}
catch (ObjectDisposedException)
{
// Make sure connection is terminated when server is disposed.
TerminateConnection(client, args, true);
}
catch (SocketException ex)
{
// Terminate connection when socket exception is encountered.
OnReceiveClientDataException(client.ID, ex);
TerminateConnection(client, args, true);
}
catch (Exception ex)
{
try
{
// For any other exception, notify and resume receive.
OnReceiveClientDataException(client.ID, ex);
ReceivePayloadAsync(client, args);
}
catch
{
// Terminate connection if resuming receiving fails.
TerminateConnection(client, args, true);
}
}
}
///
/// Processes the termination of client.
///
private void TerminateConnection(TransportProvider client, SocketAsyncEventArgs args, bool raiseEvent)
{
try
{
if (m_clientInfoLookup.TryRemove(client.ID, out TcpClientInfo clientInfo))
{
client.Reset();
clientInfo.SendArgs.Dispose();
}
if (raiseEvent)
OnClientDisconnected(client.ID);
}
finally
{
args.Dispose();
}
}
#endregion
}
}