using Quasar.Common.Messages; using Quasar.Common.Networking; using System; using System.Collections.Generic; using System.Net; using System.Net.Security; using System.Threading; namespace Quasar.Server.Networking { public class Client : IEquatable, ISender { /// /// Occurs when the state of the client changes. /// public event ClientStateEventHandler ClientState; /// /// Represents the method that will handle a change in a client's state. /// /// The client which changed its state. /// The new connection state of the client. public delegate void ClientStateEventHandler(Client s, bool connected); /// /// Fires an event that informs subscribers that the state of the client has changed. /// /// The new connection state of the client. private void OnClientState(bool connected) { if (Connected == connected) return; Connected = connected; var handler = ClientState; handler?.Invoke(this, connected); } /// /// Occurs when a message is received from the client. /// public event ClientReadEventHandler ClientRead; /// /// Represents the method that will handle a message received from the client. /// /// The client that has received the message. /// The message that received by the client. /// The length of the message. public delegate void ClientReadEventHandler(Client s, IMessage message, int messageLength); /// /// Fires an event that informs subscribers that a message has been /// received from the client. /// /// The message that received by the client. /// The length of the message. private void OnClientRead(IMessage message, int messageLength) { var handler = ClientRead; handler?.Invoke(this, message, messageLength); } /// /// Occurs when a message is sent by the client. /// public event ClientWriteEventHandler ClientWrite; /// /// Represents the method that will handle the sent message. /// /// The client that has sent the message. /// The message that has been sent by the client. /// The length of the message. public delegate void ClientWriteEventHandler(Client s, IMessage message, int messageLength); /// /// Fires an event that informs subscribers that the client has sent a message. /// /// The message that has been sent by the client. /// The length of the message. private void OnClientWrite(IMessage message, int messageLength) { var handler = ClientWrite; handler?.Invoke(this, message, messageLength); } public static bool operator ==(Client c1, Client c2) { if (ReferenceEquals(c1, null)) return ReferenceEquals(c2, null); return c1.Equals(c2); } public static bool operator !=(Client c1, Client c2) { return !(c1 == c2); } /// /// Checks whether the clients are equal. /// /// Client to compare with. /// True if equal, else False. public bool Equals(Client other) { if (ReferenceEquals(null, other)) return false; if (ReferenceEquals(this, other)) return true; try { // the port is always unique for each client return this.EndPoint.Port.Equals(other.EndPoint.Port); } catch (Exception) { return false; } } public override bool Equals(object obj) { return this.Equals(obj as Client); } /// /// Returns the hashcode for this instance. /// /// A hash code for the current instance. public override int GetHashCode() { return this.EndPoint.GetHashCode(); } /// /// The type of the message received. /// public enum ReceiveType { Header, Payload } /// /// The stream used for communication. /// private readonly SslStream _stream; /// /// The buffer pool to hold the receive-buffers for the clients. /// private readonly BufferPool _bufferPool; /// /// The queue which holds messages to send. /// private readonly Queue _sendBuffers = new Queue(); /// /// Determines if the client is currently sending messages. /// private bool _sendingMessages; /// /// Lock object for the sending messages boolean. /// private readonly object _sendingMessagesLock = new object(); /// /// The queue which holds buffers to read. /// private readonly Queue _readBuffers = new Queue(); /// /// Determines if the client is currently reading messages. /// private bool _readingMessages; /// /// Lock object for the reading messages boolean. /// private readonly object _readingMessagesLock = new object(); // receive info private int _readOffset; private int _writeOffset; private int _readableDataLen; private int _payloadLen; private ReceiveType _receiveState = ReceiveType.Header; /// /// The time when the client connected. /// public DateTime ConnectedTime { get; } /// /// The connection state of the client. /// public bool Connected { get; private set; } /// /// Determines if the client is identified. /// public bool Identified { get; set; } /// /// Stores values of the user. /// public UserState Value { get; set; } /// /// The Endpoint which the client is connected to. /// public IPEndPoint EndPoint { get; } /// /// The buffer for the client's incoming messages. /// private readonly byte[] _readBuffer; /// /// The buffer for the client's incoming payload. /// private byte[] _payloadBuffer; /// /// The header size in bytes. /// private const int HeaderSize = 4; // 4 B /// /// The maximum size of a message in bytes. /// private const int MaxMessageSize = (1024 * 1024) * 5; // 5 MB /// /// The mutex prevents multiple simultaneous write operations on the . /// private readonly Mutex _singleWriteMutex = new Mutex(); public Client(BufferPool bufferPool, SslStream stream, IPEndPoint endPoint) { try { Identified = false; Value = new UserState(); EndPoint = endPoint; ConnectedTime = DateTime.UtcNow; _stream = stream; _bufferPool = bufferPool; _readBuffer = _bufferPool.GetBuffer(); _stream.BeginRead(_readBuffer, 0, _readBuffer.Length, AsyncReceive, null); OnClientState(true); } catch (Exception) { Disconnect(); } } private void AsyncReceive(IAsyncResult result) { int bytesTransferred; try { bytesTransferred = _stream.EndRead(result); if (bytesTransferred <= 0) throw new Exception("no bytes transferred"); } catch (NullReferenceException) { return; } catch (ObjectDisposedException) { return; } catch (Exception) { Disconnect(); return; } byte[] received = new byte[bytesTransferred]; try { Array.Copy(_readBuffer, received, received.Length); } catch (Exception) { Disconnect(); return; } lock (_readBuffers) { _readBuffers.Enqueue(received); } lock (_readingMessagesLock) { if (!_readingMessages) { _readingMessages = true; ThreadPool.QueueUserWorkItem(AsyncReceive); } } try { _stream.BeginRead(_readBuffer, 0, _readBuffer.Length, AsyncReceive, null); } catch (ObjectDisposedException) { } catch (Exception) { Disconnect(); } } private void AsyncReceive(object state) { while (true) { byte[] readBuffer; lock (_readBuffers) { if (_readBuffers.Count == 0) { lock (_readingMessagesLock) { _readingMessages = false; } return; } readBuffer = _readBuffers.Dequeue(); } _readableDataLen += readBuffer.Length; bool process = true; while (process) { switch (_receiveState) { case ReceiveType.Header: { if (_payloadBuffer == null) _payloadBuffer = new byte[HeaderSize]; if (_readableDataLen + _writeOffset >= HeaderSize) { // completely received header int headerLength = HeaderSize - _writeOffset; try { Array.Copy(readBuffer, _readOffset, _payloadBuffer, _writeOffset, headerLength); _payloadLen = BitConverter.ToInt32(_payloadBuffer, _readOffset); if (_payloadLen <= 0 || _payloadLen > MaxMessageSize) throw new Exception("invalid header"); // try to re-use old payload buffers which fit if (_payloadBuffer.Length <= _payloadLen + HeaderSize) Array.Resize(ref _payloadBuffer, _payloadLen + HeaderSize); } catch (Exception) { process = false; Disconnect(); break; } _readableDataLen -= headerLength; _writeOffset += headerLength; _readOffset += headerLength; _receiveState = ReceiveType.Payload; } else // _readableDataLen + _writeOffset < HeaderSize { // received only a part of the header try { Array.Copy(readBuffer, _readOffset, _payloadBuffer, _writeOffset, _readableDataLen); } catch (Exception) { process = false; Disconnect(); break; } _readOffset += _readableDataLen; _writeOffset += _readableDataLen; process = false; // nothing left to process } break; } case ReceiveType.Payload: { int length = (_writeOffset - HeaderSize + _readableDataLen) >= _payloadLen ? _payloadLen - (_writeOffset - HeaderSize) : _readableDataLen; try { Array.Copy(readBuffer, _readOffset, _payloadBuffer, _writeOffset, length); } catch (Exception) { process = false; Disconnect(); break; } _writeOffset += length; _readOffset += length; _readableDataLen -= length; if (_writeOffset - HeaderSize == _payloadLen) { // completely received payload try { using (PayloadReader pr = new PayloadReader(_payloadBuffer, _payloadLen + HeaderSize, false)) { IMessage message = pr.ReadMessage(); OnClientRead(message, _payloadBuffer.Length); } } catch (Exception) { process = false; Disconnect(); break; } _receiveState = ReceiveType.Header; _payloadLen = 0; _writeOffset = 0; } if (_readableDataLen == 0) process = false; break; } } } _readOffset = 0; _readableDataLen = 0; } } /// /// Sends a message to the connected client. /// /// The type of the message. /// The message to be sent. public void Send(T message) where T : IMessage { if (!Connected || message == null) return; lock (_sendBuffers) { _sendBuffers.Enqueue(message); lock (_sendingMessagesLock) { if (_sendingMessages) return; _sendingMessages = true; ThreadPool.QueueUserWorkItem(ProcessSendBuffers); } } } /// /// Sends a message to the connected client. /// Blocks the thread until the message has been sent. /// /// The type of the message. /// The message to be sent. public void SendBlocking(T message) where T : IMessage { if (!Connected || message == null) return; SafeSendMessage(message); } /// /// Safely sends a message and prevents multiple simultaneous /// write operations on the . /// /// The message to send. private void SafeSendMessage(IMessage message) { try { _singleWriteMutex.WaitOne(); using (PayloadWriter pw = new PayloadWriter(_stream, true)) { OnClientWrite(message, pw.WriteMessage(message)); } } catch (Exception) { Disconnect(); SendCleanup(true); } finally { _singleWriteMutex.ReleaseMutex(); } } private void ProcessSendBuffers(object state) { while (true) { if (!Connected) { SendCleanup(true); return; } IMessage message; lock (_sendBuffers) { if (_sendBuffers.Count == 0) { SendCleanup(); return; } message = _sendBuffers.Dequeue(); } SafeSendMessage(message); } } private void SendCleanup(bool clear = false) { lock (_sendingMessagesLock) { _sendingMessages = false; } if (!clear) return; lock (_sendBuffers) { _sendBuffers.Clear(); } } /// /// Disconnect the client from the server and dispose of /// resources associated with the client. /// public void Disconnect() { if (_stream != null) { _stream.Close(); _readOffset = 0; _writeOffset = 0; _readableDataLen = 0; _payloadLen = 0; _payloadBuffer = null; _receiveState = ReceiveType.Header; _singleWriteMutex.Dispose(); _bufferPool.ReturnBuffer(_readBuffer); } OnClientState(false); } } }