initial commit
This commit is contained in:
i2p
@@ -0,0 +1,612 @@
|
||||
using Quasar.Common.Extensions;
|
||||
using Quasar.Common.Messages;
|
||||
using Quasar.Common.Networking;
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.Linq;
|
||||
using System.Net;
|
||||
using System.Net.Security;
|
||||
using System.Net.Sockets;
|
||||
using System.Security.Authentication;
|
||||
using System.Security.Cryptography;
|
||||
using System.Security.Cryptography.X509Certificates;
|
||||
using System.Threading;
|
||||
|
||||
namespace Quasar.Client.Networking
|
||||
{
|
||||
public class Client : ISender
|
||||
{
|
||||
/// <summary>
|
||||
/// Occurs as a result of an unrecoverable issue with the client.
|
||||
/// </summary>
|
||||
public event ClientFailEventHandler ClientFail;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a method that will handle failure of the client.
|
||||
/// </summary>
|
||||
/// <param name="s">The client that has failed.</param>
|
||||
/// <param name="ex">The exception containing information about the cause of the client's failure.</param>
|
||||
public delegate void ClientFailEventHandler(Client s, Exception ex);
|
||||
|
||||
/// <summary>
|
||||
/// Fires an event that informs subscribers that the client has failed.
|
||||
/// </summary>
|
||||
/// <param name="ex">The exception containing information about the cause of the client's failure.</param>
|
||||
private void OnClientFail(Exception ex)
|
||||
{
|
||||
var handler = ClientFail;
|
||||
handler?.Invoke(this, ex);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Occurs when the state of the client has changed.
|
||||
/// </summary>
|
||||
public event ClientStateEventHandler ClientState;
|
||||
|
||||
/// <summary>
|
||||
/// Represents the method that will handle a change in the client's state
|
||||
/// </summary>
|
||||
/// <param name="s">The client which changed its state.</param>
|
||||
/// <param name="connected">The new connection state of the client.</param>
|
||||
public delegate void ClientStateEventHandler(Client s, bool connected);
|
||||
|
||||
/// <summary>
|
||||
/// Fires an event that informs subscribers that the state of the client has changed.
|
||||
/// </summary>
|
||||
/// <param name="connected">The new connection state of the client.</param>
|
||||
private void OnClientState(bool connected)
|
||||
{
|
||||
if (Connected == connected) return;
|
||||
|
||||
Connected = connected;
|
||||
|
||||
var handler = ClientState;
|
||||
handler?.Invoke(this, connected);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Occurs when a message is received from the server.
|
||||
/// </summary>
|
||||
public event ClientReadEventHandler ClientRead;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a method that will handle a message from the server.
|
||||
/// </summary>
|
||||
/// <param name="s">The client that has received the message.</param>
|
||||
/// <param name="message">The message that has been received by the server.</param>
|
||||
/// <param name="messageLength">The length of the message.</param>
|
||||
public delegate void ClientReadEventHandler(Client s, IMessage message, int messageLength);
|
||||
|
||||
/// <summary>
|
||||
/// Fires an event that informs subscribers that a message has been received by the server.
|
||||
/// </summary>
|
||||
/// <param name="message">The message that has been received by the server.</param>
|
||||
/// <param name="messageLength">The length of the message.</param>
|
||||
private void OnClientRead(IMessage message, int messageLength)
|
||||
{
|
||||
var handler = ClientRead;
|
||||
handler?.Invoke(this, message, messageLength);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Occurs when a message is sent by the client.
|
||||
/// </summary>
|
||||
public event ClientWriteEventHandler ClientWrite;
|
||||
|
||||
/// <summary>
|
||||
/// Represents the method that will handle the sent message.
|
||||
/// </summary>
|
||||
/// <param name="s">The client that has sent the message.</param>
|
||||
/// <param name="message">The message that has been sent by the client.</param>
|
||||
/// <param name="messageLength">The length of the message.</param>
|
||||
public delegate void ClientWriteEventHandler(Client s, IMessage message, int messageLength);
|
||||
|
||||
/// <summary>
|
||||
/// Fires an event that informs subscribers that the client has sent a message.
|
||||
/// </summary>
|
||||
/// <param name="message">The message that has been sent by the client.</param>
|
||||
/// <param name="messageLength">The length of the message.</param>
|
||||
private void OnClientWrite(IMessage message, int messageLength)
|
||||
{
|
||||
var handler = ClientWrite;
|
||||
handler?.Invoke(this, message, messageLength);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The type of the message received.
|
||||
/// </summary>
|
||||
public enum ReceiveType
|
||||
{
|
||||
Header,
|
||||
Payload
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The buffer size for receiving data in bytes.
|
||||
/// </summary>
|
||||
public int BUFFER_SIZE { get { return 1024 * 16; } } // 16KB
|
||||
|
||||
/// <summary>
|
||||
/// The keep-alive time in ms.
|
||||
/// </summary>
|
||||
public uint KEEP_ALIVE_TIME { get { return 25000; } } // 25s
|
||||
|
||||
/// <summary>
|
||||
/// The keep-alive interval in ms.
|
||||
/// </summary>
|
||||
public uint KEEP_ALIVE_INTERVAL { get { return 25000; } } // 25s
|
||||
|
||||
/// <summary>
|
||||
/// The header size in bytes.
|
||||
/// </summary>
|
||||
public int HEADER_SIZE { get { return 4; } } // 4B
|
||||
|
||||
/// <summary>
|
||||
/// The maximum size of a message in bytes.
|
||||
/// </summary>
|
||||
public int MAX_MESSAGE_SIZE { get { return (1024 * 1024) * 50; } } // 50MB
|
||||
|
||||
/// <summary>
|
||||
/// Gets if the client is currently connected to a server.
|
||||
/// </summary>
|
||||
public bool Connected { get; private set; }
|
||||
|
||||
/// <summary>
|
||||
/// The stream used for communication.
|
||||
/// </summary>
|
||||
private SslStream _stream;
|
||||
|
||||
/// <summary>
|
||||
/// The server certificate.
|
||||
/// </summary>
|
||||
private readonly X509Certificate2 _serverCertificate;
|
||||
|
||||
/// <summary>
|
||||
/// The buffer for incoming messages.
|
||||
/// </summary>
|
||||
private byte[] _readBuffer;
|
||||
|
||||
/// <summary>
|
||||
/// The buffer for the client's incoming payload.
|
||||
/// </summary>
|
||||
private byte[] _payloadBuffer;
|
||||
|
||||
/// <summary>
|
||||
/// The queue which holds messages to send.
|
||||
/// </summary>
|
||||
private readonly Queue<IMessage> _sendBuffers = new Queue<IMessage>();
|
||||
|
||||
/// <summary>
|
||||
/// Determines if the client is currently sending messages.
|
||||
/// </summary>
|
||||
private bool _sendingMessages;
|
||||
|
||||
/// <summary>
|
||||
/// Lock object for the sending messages boolean.
|
||||
/// </summary>
|
||||
private readonly object _sendingMessagesLock = new object();
|
||||
|
||||
/// <summary>
|
||||
/// The queue which holds buffers to read.
|
||||
/// </summary>
|
||||
private readonly Queue<byte[]> _readBuffers = new Queue<byte[]>();
|
||||
|
||||
/// <summary>
|
||||
/// Determines if the client is currently reading messages.
|
||||
/// </summary>
|
||||
private bool _readingMessages;
|
||||
|
||||
/// <summary>
|
||||
/// Lock object for the reading messages boolean.
|
||||
/// </summary>
|
||||
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;
|
||||
|
||||
/// <summary>
|
||||
/// The mutex prevents multiple simultaneous write operations on the <see cref="_stream"/>.
|
||||
/// </summary>
|
||||
private readonly Mutex _singleWriteMutex = new Mutex();
|
||||
|
||||
/// <summary>
|
||||
/// Constructor of the client, initializes serializer types.
|
||||
/// </summary>
|
||||
/// <param name="serverCertificate">The server certificate.</param>
|
||||
protected Client(X509Certificate2 serverCertificate)
|
||||
{
|
||||
_serverCertificate = serverCertificate;
|
||||
_readBuffer = new byte[BUFFER_SIZE];
|
||||
TypeRegistry.AddTypesToSerializer(typeof(IMessage), TypeRegistry.GetPacketTypes(typeof(IMessage)).ToArray());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Attempts to connect to the specified ip address on the specified port.
|
||||
/// </summary>
|
||||
/// <param name="ip">The ip address to connect to.</param>
|
||||
/// <param name="port">The port of the host.</param>
|
||||
protected void Connect(IPAddress ip, ushort port)
|
||||
{
|
||||
Socket handle = null;
|
||||
try
|
||||
{
|
||||
Disconnect();
|
||||
|
||||
handle = new Socket(ip.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
|
||||
handle.SetKeepAliveEx(KEEP_ALIVE_INTERVAL, KEEP_ALIVE_TIME);
|
||||
handle.Connect(ip, port);
|
||||
|
||||
if (handle.Connected)
|
||||
{
|
||||
_stream = new SslStream(new NetworkStream(handle, true), false, ValidateServerCertificate);
|
||||
_stream.AuthenticateAsClient(ip.ToString(), null, SslProtocols.Tls12, false);
|
||||
_stream.BeginRead(_readBuffer, 0, _readBuffer.Length, AsyncReceive, null);
|
||||
OnClientState(true);
|
||||
}
|
||||
else
|
||||
{
|
||||
handle.Dispose();
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
handle?.Dispose();
|
||||
OnClientFail(ex);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Validates the server certificate by comparing it with the included server certificate.
|
||||
/// </summary>
|
||||
/// <param name="sender">The sender of the callback.</param>
|
||||
/// <param name="certificate">The server certificate to validate.</param>
|
||||
/// <param name="chain">The X.509 chain.</param>
|
||||
/// <param name="sslPolicyErrors">The SSL policy errors.</param>
|
||||
/// <returns>Returns <value>true</value> when the validation was successful, otherwise <value>false</value>.</returns>
|
||||
private bool ValidateServerCertificate(object sender, X509Certificate certificate, X509Chain chain, SslPolicyErrors sslPolicyErrors)
|
||||
{
|
||||
#if DEBUG
|
||||
// for debugging don't validate server certificate
|
||||
return true;
|
||||
#else
|
||||
var serverCsp = (RSACryptoServiceProvider)_serverCertificate.PublicKey.Key;
|
||||
var connectedCsp = (RSACryptoServiceProvider)new X509Certificate2(certificate).PublicKey.Key;
|
||||
// compare the received server certificate with the included server certificate to validate we are connected to the correct server
|
||||
return _serverCertificate.Equals(certificate);
|
||||
#endif
|
||||
}
|
||||
|
||||
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 ex)
|
||||
{
|
||||
OnClientFail(ex);
|
||||
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 ex)
|
||||
{
|
||||
OnClientFail(ex);
|
||||
}
|
||||
}
|
||||
|
||||
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[HEADER_SIZE];
|
||||
|
||||
if (_readableDataLen + _writeOffset >= HEADER_SIZE)
|
||||
{
|
||||
// completely received header
|
||||
int headerLength = HEADER_SIZE - _writeOffset;
|
||||
|
||||
try
|
||||
{
|
||||
Array.Copy(readBuffer, _readOffset, _payloadBuffer, _writeOffset, headerLength);
|
||||
|
||||
_payloadLen = BitConverter.ToInt32(_payloadBuffer, _readOffset);
|
||||
|
||||
if (_payloadLen <= 0 || _payloadLen > MAX_MESSAGE_SIZE)
|
||||
throw new Exception("invalid header");
|
||||
|
||||
// try to re-use old payload buffers which fit
|
||||
if (_payloadBuffer.Length <= _payloadLen + HEADER_SIZE)
|
||||
Array.Resize(ref _payloadBuffer, _payloadLen + HEADER_SIZE);
|
||||
}
|
||||
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 - HEADER_SIZE + _readableDataLen) >= _payloadLen
|
||||
? _payloadLen - (_writeOffset - HEADER_SIZE)
|
||||
: _readableDataLen;
|
||||
|
||||
try
|
||||
{
|
||||
Array.Copy(readBuffer, _readOffset, _payloadBuffer, _writeOffset, length);
|
||||
}
|
||||
catch (Exception)
|
||||
{
|
||||
process = false;
|
||||
Disconnect();
|
||||
break;
|
||||
}
|
||||
|
||||
_writeOffset += length;
|
||||
_readOffset += length;
|
||||
_readableDataLen -= length;
|
||||
|
||||
if (_writeOffset - HEADER_SIZE == _payloadLen)
|
||||
{
|
||||
// completely received payload
|
||||
try
|
||||
{
|
||||
using (PayloadReader pr = new PayloadReader(_payloadBuffer, _payloadLen + HEADER_SIZE, 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;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends a message to the connected server.
|
||||
/// </summary>
|
||||
/// <typeparam name="T">The type of the message.</typeparam>
|
||||
/// <param name="message">The message to be sent.</param>
|
||||
public void Send<T>(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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends a message to the connected server.
|
||||
/// Blocks the thread until the message has been sent.
|
||||
/// </summary>
|
||||
/// <typeparam name="T">The type of the message.</typeparam>
|
||||
/// <param name="message">The message to be sent.</param>
|
||||
public void SendBlocking<T>(T message) where T : IMessage
|
||||
{
|
||||
if (!Connected || message == null) return;
|
||||
|
||||
SafeSendMessage(message);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Safely sends a message and prevents multiple simultaneous
|
||||
/// write operations on the <see cref="_stream"/>.
|
||||
/// </summary>
|
||||
/// <param name="message">The message to send.</param>
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Disconnect the client from the server, disconnect all proxies that
|
||||
/// are held by this client, and dispose of other resources associated
|
||||
/// with this client.
|
||||
/// </summary>
|
||||
public void Disconnect()
|
||||
{
|
||||
if (_stream != null)
|
||||
{
|
||||
_stream.Close();
|
||||
_readOffset = 0;
|
||||
_writeOffset = 0;
|
||||
_readableDataLen = 0;
|
||||
_payloadLen = 0;
|
||||
_payloadBuffer = null;
|
||||
_receiveState = ReceiveType.Header;
|
||||
//_singleWriteMutex.Dispose(); TODO: fix socket re-use by creating new client on disconnect
|
||||
|
||||
}
|
||||
|
||||
OnClientState(false);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user