/// <summary>

/// Unix Domain Socket client is used to read/write data from/into the connected Unix Domain Socket server

/// </summary>

/// <remarks>Thread-safe</remarks>

public class UdsClient : IDisposable

{

/// <summary>

/// Initialize Unix Domain Socket client with a given socket path

/// </summary>

/// <param name="path">Socket path</param>

public UdsClient(string path) : this(new UnixDomainSocketEndPoint(path)) {}

/// <summary>

/// Initialize Unix Domain Socket client with a given Unix Domain Socket endpoint

/// </summary>

/// <param name="endpoint">Unix Domain Socket endpoint</param>

public UdsClient(UnixDomainSocketEndPoint endpoint)

{

Id = Guid.NewGuid();

Endpoint = endpoint;

}

/// <summary>

/// Client Id

/// </summary>

public Guid Id { get; }

/// <summary>

/// Endpoint

/// </summary>

public EndPoint Endpoint { get; private set; }

/// <summary>

/// Socket

/// </summary>

public Socket Socket { get; private set; }

/// <summary>

/// Number of bytes pending sent by the client

/// </summary>

public long BytesPending { get; private set; }

/// <summary>

/// Number of bytes sending by the client

/// </summary>

public long BytesSending { get; private set; }

/// <summary>

/// Number of bytes sent by the client

/// </summary>

public long BytesSent { get; private set; }

/// <summary>

/// Number of bytes received by the client

/// </summary>

public long BytesReceived { get; private set; }

/// <summary>

/// Option: receive buffer limit

/// </summary>

public int OptionReceiveBufferLimit { get; set; } = 0;

/// <summary>

/// Option: receive buffer size

/// </summary>

public int OptionReceiveBufferSize { get; set; } = 8192;

/// <summary>

/// Option: send buffer limit

/// </summary>

public int OptionSendBufferLimit { get; set; } = 0;

/// <summary>

/// Option: send buffer size

/// </summary>

public int OptionSendBufferSize { get; set; } = 8192;

#region Connect/Disconnect client

private SocketAsyncEventArgs _connectEventArg;

/// <summary>

/// Is the client connecting?

/// </summary>

public bool IsConnecting { get; private set; }

/// <summary>

/// Is the client connected?

/// </summary>

public bool IsConnected { get; private set; }

/// <summary>

/// Create a new socket object

/// </summary>

/// <remarks>

/// Method may be override if you need to prepare some specific socket object in your implementation.

/// </remarks>

/// <returns>Socket object</returns>

protected virtual Socket CreateSocket()

{

return new Socket(Endpoint.AddressFamily, SocketType.Stream, ProtocolType.IP);

}

/// <summary>

/// Connect the client (synchronous)

/// </summary>

/// <remarks>

/// Please note that synchronous connect will not receive data automatically!

/// You should use Receive() or ReceiveAsync() method manually after successful connection.

/// </remarks>

/// <returns>'true' if the client was successfully connected, 'false' if the client failed to connect</returns>

public virtual bool Connect()

{

if (IsConnected || IsConnecting)

return false;

// Setup buffers

_receiveBuffer = new Buffer();

_sendBufferMain = new Buffer();

_sendBufferFlush = new Buffer();

// Setup event args

_connectEventArg = new SocketAsyncEventArgs();

_connectEventArg.RemoteEndPoint = Endpoint;

_connectEventArg.Completed += OnAsyncCompleted;

_receiveEventArg = new SocketAsyncEventArgs();

_receiveEventArg.Completed += OnAsyncCompleted;

_sendEventArg = new SocketAsyncEventArgs();

_sendEventArg.Completed += OnAsyncCompleted;

// Create a new client socket

Socket = CreateSocket();

// Update the client socket disposed flag

IsSocketDisposed = false;

// Call the client connecting handler

OnConnecting();

try

{

// Connect to the server

Socket.Connect(Endpoint);

}

catch (SocketException ex)

{

// Call the client error handler

SendError(ex.SocketErrorCode);

// Reset event args

_connectEventArg.Completed -= OnAsyncCompleted;

_receiveEventArg.Completed -= OnAsyncCompleted;

_sendEventArg.Completed -= OnAsyncCompleted;

// Call the client disconnecting handler

OnDisconnecting();

// Close the client socket

Socket.Close();

// Dispose the client socket

Socket.Dispose();

// Dispose event arguments

_connectEventArg.Dispose();

_receiveEventArg.Dispose();

_sendEventArg.Dispose();

// Call the client disconnected handler

OnDisconnected();

return false;

}

// Prepare receive & send buffers

_receiveBuffer.Reserve(OptionReceiveBufferSize);

_sendBufferMain.Reserve(OptionSendBufferSize);

_sendBufferFlush.Reserve(OptionSendBufferSize);

// Reset statistic

BytesPending = 0;

BytesSending = 0;

BytesSent = 0;

BytesReceived = 0;

// Update the connected flag

IsConnected = true;

// Call the client connected handler

OnConnected();

// Call the empty send buffer handler

if (_sendBufferMain.IsEmpty)

OnEmpty();

return true;

}

/// <summary>

/// Disconnect the client (synchronous)

/// </summary>

/// <returns>'true' if the client was successfully disconnected, 'false' if the client is already disconnected</returns>

public virtual bool Disconnect()

{

if (!IsConnected && !IsConnecting)

return false;

// Cancel connecting operation

if (IsConnecting)

Socket.CancelConnectAsync(_connectEventArg);

// Reset event args

_connectEventArg.Completed -= OnAsyncCompleted;

_receiveEventArg.Completed -= OnAsyncCompleted;

_sendEventArg.Completed -= OnAsyncCompleted;

// Call the client disconnecting handler

OnDisconnecting();

try

{

try

{

// Shutdown the socket associated with the client

Socket.Shutdown(SocketShutdown.Both);

}

catch (SocketException) {}

// Close the client socket

Socket.Close();

// Dispose the client socket

Socket.Dispose();

// Dispose event arguments

_connectEventArg.Dispose();

_receiveEventArg.Dispose();

_sendEventArg.Dispose();

// Update the client socket disposed flag

IsSocketDisposed = true;

}

catch (ObjectDisposedException) {}

// Update the connected flag

IsConnected = false;

// Update sending/receiving flags

_receiving = false;

_sending = false;

// Clear send/receive buffers

ClearBuffers();

// Call the client disconnected handler

OnDisconnected();

return true;

}

/// <summary>

/// Reconnect the client (synchronous)

/// </summary>

/// <returns>'true' if the client was successfully reconnected, 'false' if the client is already reconnected</returns>

public virtual bool Reconnect()

{

if (!Disconnect())

return false;

return Connect();

}

/// <summary>

/// Connect the client (asynchronous)

/// </summary>

/// <returns>'true' if the client was successfully connected, 'false' if the client failed to connect</returns>

public virtual bool ConnectAsync()

{

if (IsConnected || IsConnecting)

return false;

// Setup buffers

_receiveBuffer = new Buffer();

_sendBufferMain = new Buffer();

_sendBufferFlush = new Buffer();

// Setup event args

_connectEventArg = new SocketAsyncEventArgs();

_connectEventArg.RemoteEndPoint = Endpoint;

_connectEventArg.Completed += OnAsyncCompleted;

_receiveEventArg = new SocketAsyncEventArgs();

_receiveEventArg.Completed += OnAsyncCompleted;

_sendEventArg = new SocketAsyncEventArgs();

_sendEventArg.Completed += OnAsyncCompleted;

// Create a new client socket

Socket = CreateSocket();

// Update the client socket disposed flag

IsSocketDisposed = false;

// Update the connecting flag

IsConnecting = true;

// Call the client connecting handler

OnConnecting();

// Async connect to the server

if (!Socket.ConnectAsync(_connectEventArg))

ProcessConnect(_connectEventArg);

return true;

}

/// <summary>

/// Disconnect the client (asynchronous)

/// </summary>

/// <returns>'true' if the client was successfully disconnected, 'false' if the client is already disconnected</returns>

public virtual bool DisconnectAsync() => Disconnect();

/// <summary>

/// Reconnect the client (asynchronous)

/// </summary>

/// <returns>'true' if the client was successfully reconnected, 'false' if the client is already reconnected</returns>

public virtual bool ReconnectAsync()

{

if (!DisconnectAsync())

return false;

while (IsConnected)

Thread.Yield();

return ConnectAsync();

}

#endregion

#region Send/Receive data

// Receive buffer

private bool _receiving;

private Buffer _receiveBuffer;

private SocketAsyncEventArgs _receiveEventArg;

// Send buffer

private readonly object _sendLock = new object();

private bool _sending;

private Buffer _sendBufferMain;

private Buffer _sendBufferFlush;

private SocketAsyncEventArgs _sendEventArg;

private long _sendBufferFlushOffset;

/// <summary>

/// Send data to the server (synchronous)

/// </summary>

/// <param name="buffer">Buffer to send</param>

/// <returns>Size of sent data</returns>

public virtual long Send(byte[] buffer) => Send(buffer.AsSpan());

/// <summary>

/// Send data to the server (synchronous)

/// </summary>

/// <param name="buffer">Buffer to send</param>

/// <param name="offset">Buffer offset</param>

/// <param name="size">Buffer size</param>

/// <returns>Size of sent data</returns>

public virtual long Send(byte[] buffer, long offset, long size) => Send(buffer.AsSpan((int)offset, (int)size));

/// <summary>

/// Send data to the server (synchronous)

/// </summary>

/// <param name="buffer">Buffer to send as a span of bytes</param>

/// <returns>Size of sent data</returns>

public virtual long Send(ReadOnlySpan<byte> buffer)

{

if (!IsConnected)

return 0;

if (buffer.IsEmpty)

return 0;

// Sent data to the server

long sent = Socket.Send(buffer, SocketFlags.None, out SocketError ec);

if (sent > 0)

{

// Update statistic

BytesSent += sent;

// Call the buffer sent handler

OnSent(sent, BytesPending + BytesSending);

}

// Check for socket error

if (ec != SocketError.Success)

{

SendError(ec);

Disconnect();

}

return sent;

}

/// <summary>

/// Send text to the server (synchronous)

/// </summary>

/// <param name="text">Text string to send</param>

/// <returns>Size of sent text</returns>

public virtual long Send(string text) => Send(Encoding.UTF8.GetBytes(text));

/// <summary>

/// Send text to the server (synchronous)

/// </summary>

/// <param name="text">Text to send as a span of characters</param>

/// <returns>Size of sent text</returns>

public virtual long Send(ReadOnlySpan<char> text) => Send(Encoding.UTF8.GetBytes(text.ToArray()));

/// <summary>

/// Send data to the server (asynchronous)

/// </summary>

/// <param name="buffer">Buffer to send</param>

/// <returns>'true' if the data was successfully sent, 'false' if the client is not connected</returns>

public virtual bool SendAsync(byte[] buffer) => SendAsync(buffer.AsSpan());

/// <summary>

/// Send data to the server (asynchronous)

/// </summary>

/// <param name="buffer">Buffer to send</param>

/// <param name="offset">Buffer offset</param>

/// <param name="size">Buffer size</param>

/// <returns>'true' if the data was successfully sent, 'false' if the client is not connected</returns>

public virtual bool SendAsync(byte[] buffer, long offset, long size) => SendAsync(buffer.AsSpan((int)offset, (int)size));

/// <summary>

/// Send data to the server (asynchronous)

/// </summary>

/// <param name="buffer">Buffer to send as a span of bytes</param>

/// <returns>'true' if the data was successfully sent, 'false' if the client is not connected</returns>

public virtual bool SendAsync(ReadOnlySpan<byte> buffer)

{

if (!IsConnected)

return false;

if (buffer.IsEmpty)

return true;

lock (_sendLock)

{

// Check the send buffer limit

if (((_sendBufferMain.Size + buffer.Length) > OptionSendBufferLimit) && (OptionSendBufferLimit > 0))

{

SendError(SocketError.NoBufferSpaceAvailable);

return false;

}

// Fill the main send buffer

_sendBufferMain.Append(buffer);

// Update statistic

BytesPending = _sendBufferMain.Size;

// Avoid multiple send handlers

if (_sending)

return true;

else

_sending = true;

// Try to send the main buffer

TrySend();

}

return true;

}

/// <summary>

/// Send text to the server (asynchronous)

/// </summary>

/// <param name="text">Text string to send</param>

/// <returns>'true' if the text was successfully sent, 'false' if the client is not connected</returns>

public virtual bool SendAsync(string text) => SendAsync(Encoding.UTF8.GetBytes(text));

/// <summary>

/// Send text to the server (asynchronous)

/// </summary>

/// <param name="text">Text to send as a span of characters</param>

/// <returns>'true' if the text was successfully sent, 'false' if the client is not connected</returns>

public virtual bool SendAsync(ReadOnlySpan<char> text) => SendAsync(Encoding.UTF8.GetBytes(text.ToArray()));

/// <summary>

/// Receive data from the server (synchronous)

/// </summary>

/// <param name="buffer">Buffer to receive</param>

/// <returns>Size of received data</returns>

public virtual long Receive(byte[] buffer) { return Receive(buffer, 0, buffer.Length); }

/// <summary>

/// Receive data from the server (synchronous)

/// </summary>

/// <param name="buffer">Buffer to receive</param>

/// <param name="offset">Buffer offset</param>

/// <param name="size">Buffer size</param>

/// <returns>Size of received data</returns>

public virtual long Receive(byte[] buffer, long offset, long size)

{

if (!IsConnected)

return 0;

if (size == 0)

return 0;

// Receive data from the server

long received = Socket.Receive(buffer, (int)offset, (int)size, SocketFlags.None, out SocketError ec);

if (received > 0)

{

// Update statistic

BytesReceived += received;

// Call the buffer received handler

OnReceived(buffer, 0, received);

}

// Check for socket error

if (ec != SocketError.Success)

{

SendError(ec);

Disconnect();

}

return received;

}

/// <summary>

/// Receive text from the server (synchronous)

/// </summary>

/// <param name="size">Text size to receive</param>

/// <returns>Received text</returns>

public virtual string Receive(long size)

{

var buffer = new byte[size];

var length = Receive(buffer);

return Encoding.UTF8.GetString(buffer, 0, (int)length);

}

/// <summary>

/// Receive data from the server (asynchronous)

/// </summary>

public virtual void ReceiveAsync()

{

// Try to receive data from the server

TryReceive();

}

/// <summary>

/// Try to receive new data

/// </summary>

private void TryReceive()

{

if (_receiving)

return;

if (!IsConnected)

return;

bool process = true;

while (process)

{

process = false;

try

{

// Async receive with the receive handler

_receiving = true;

_receiveEventArg.SetBuffer(_receiveBuffer.Data, 0, (int)_receiveBuffer.Capacity);

if (!Socket.ReceiveAsync(_receiveEventArg))

process = ProcessReceive(_receiveEventArg);

}

catch (ObjectDisposedException) {}

}

}

/// <summary>

/// Try to send pending data

/// </summary>

private void TrySend()

{

if (!IsConnected)

return;

bool empty = false;

bool process = true;

while (process)

{

process = false;

lock (_sendLock)

{

// Is previous socket send in progress?

if (_sendBufferFlush.IsEmpty)

{

// Swap flush and main buffers

_sendBufferFlush = Interlocked.Exchange(ref _sendBufferMain, _sendBufferFlush);

_sendBufferFlushOffset = 0;

// Update statistic

BytesPending = 0;

BytesSending += _sendBufferFlush.Size;

// Check if the flush buffer is empty

if (_sendBufferFlush.IsEmpty)

{

// Need to call empty send buffer handler

empty = true;

// End sending process

_sending = false;

}

}

else

return;

}

// Call the empty send buffer handler

if (empty)

{

OnEmpty();

return;

}

try

{

// Async write with the write handler

_sendEventArg.SetBuffer(_sendBufferFlush.Data, (int)_sendBufferFlushOffset, (int)(_sendBufferFlush.Size - _sendBufferFlushOffset));

if (!Socket.SendAsync(_sendEventArg))

process = ProcessSend(_sendEventArg);

}

catch (ObjectDisposedException) {}

}

}

/// <summary>

/// Clear send/receive buffers

/// </summary>

private void ClearBuffers()

{

lock (_sendLock)

{

// Clear send buffers

_sendBufferMain.Clear();

_sendBufferFlush.Clear();

_sendBufferFlushOffset= 0;

// Update statistic

BytesPending = 0;

BytesSending = 0;

}

}

#endregion

#region IO processing

/// <summary>

/// This method is called whenever a receive or send operation is completed on a socket

/// </summary>

private void OnAsyncCompleted(object sender, SocketAsyncEventArgs e)

{

if (IsSocketDisposed)

return;

// Determine which type of operation just completed and call the associated handler

switch (e.LastOperation)

{

case SocketAsyncOperation.Connect:

ProcessConnect(e);

break;

case SocketAsyncOperation.Receive:

if (ProcessReceive(e))

TryReceive();

break;

case SocketAsyncOperation.Send:

if (ProcessSend(e))

TrySend();

break;

default:

throw new ArgumentException("The last operation completed on the socket was not a receive or send");

}

}

/// <summary>

/// This method is invoked when an asynchronous connect operation completes

/// </summary>

private void ProcessConnect(SocketAsyncEventArgs e)

{

IsConnecting = false;

if (e.SocketError == SocketError.Success)

{

// Prepare receive & send buffers

_receiveBuffer.Reserve(OptionReceiveBufferSize);

_sendBufferMain.Reserve(OptionSendBufferSize);

_sendBufferFlush.Reserve(OptionSendBufferSize);

// Reset statistic

BytesPending = 0;

BytesSending = 0;

BytesSent = 0;

BytesReceived = 0;

// Update the connected flag

IsConnected = true;

// Try to receive something from the server

TryReceive();

// Check the socket disposed state: in some rare cases it might be disconnected while receiving!

if (IsSocketDisposed)

return;

// Call the client connected handler

OnConnected();

// Call the empty send buffer handler

if (_sendBufferMain.IsEmpty)

OnEmpty();

}

else

{

// Call the client disconnected handler

SendError(e.SocketError);

OnDisconnected();

}

}

/// <summary>

/// This method is invoked when an asynchronous receive operation completes

/// </summary>

private bool ProcessReceive(SocketAsyncEventArgs e)

{

if (!IsConnected)

return false;

long size = e.BytesTransferred;

// Received some data from the server

if (size > 0)

{

// Update statistic

BytesReceived += size;

// Call the buffer received handler

OnReceived(_receiveBuffer.Data, 0, size);

// If the receive buffer is full increase its size

if (_receiveBuffer.Capacity == size)

{

// Check the receive buffer limit

if (((2 * size) > OptionReceiveBufferLimit) && (OptionReceiveBufferLimit > 0))

{

SendError(SocketError.NoBufferSpaceAvailable);

DisconnectAsync();

return false;

}

_receiveBuffer.Reserve(2 * size);

}

}

_receiving = false;

// Try to receive again if the client is valid

if (e.SocketError == SocketError.Success)

{

// If zero is returned from a read operation, the remote end has closed the connection

if (size > 0)

return true;

else

DisconnectAsync();

}

else

{

SendError(e.SocketError);

DisconnectAsync();

}

return false;

}

/// <summary>

/// This method is invoked when an asynchronous send operation completes

/// </summary>

private bool ProcessSend(SocketAsyncEventArgs e)

{

if (!IsConnected)

return false;

long size = e.BytesTransferred;

// Send some data to the server

if (size > 0)

{

// Update statistic

BytesSending -= size;

BytesSent += size;

// Increase the flush buffer offset

_sendBufferFlushOffset += size;

// Successfully send the whole flush buffer

if (_sendBufferFlushOffset == _sendBufferFlush.Size)

{

// Clear the flush buffer

_sendBufferFlush.Clear();

_sendBufferFlushOffset = 0;

}

// Call the buffer sent handler

OnSent(size, BytesPending + BytesSending);

}

// Try to send again if the client is valid

if (e.SocketError == SocketError.Success)

return true;

else

{

SendError(e.SocketError);

DisconnectAsync();

return false;

}

}

#endregion

#region Session handlers

/// <summary>

/// Handle client connecting notification

/// </summary>

protected virtual void OnConnecting() {}

/// <summary>

/// Handle client connected notification

/// </summary>

protected virtual void OnConnected() {}

/// <summary>

/// Handle client disconnecting notification

/// </summary>

protected virtual void OnDisconnecting() {}

/// <summary>

/// Handle client disconnected notification

/// </summary>

protected virtual void OnDisconnected() {}

/// <summary>

/// Handle buffer received notification

/// </summary>

/// <param name="buffer">Received buffer</param>

/// <param name="offset">Received buffer offset</param>

/// <param name="size">Received buffer size</param>

/// <remarks>

/// Notification is called when another part of buffer was received from the server

/// </remarks>

protected virtual void OnReceived(byte[] buffer, long offset, long size) {}

/// <summary>

/// Handle buffer sent notification

/// </summary>

/// <param name="sent">Size of sent buffer</param>

/// <param name="pending">Size of pending buffer</param>

/// <remarks>

/// Notification is called when another part of buffer was sent to the server.

/// This handler could be used to send another buffer to the server for instance when the pending size is zero.

/// </remarks>

protected virtual void OnSent(long sent, long pending) {}

/// <summary>

/// Handle empty send buffer notification

/// </summary>

/// <remarks>

/// Notification is called when the send buffer is empty and ready for a new data to send.

/// This handler could be used to send another buffer to the server.

/// </remarks>

protected virtual void OnEmpty() {}

/// <summary>

/// Handle error notification

/// </summary>

/// <param name="error">Socket error code</param>

protected virtual void OnError(SocketError error) {}

#endregion

#region Error handling

/// <summary>

/// Send error notification

/// </summary>

/// <param name="error">Socket error code</param>

private void SendError(SocketError error)

{

// Skip disconnect errors

if ((error == SocketError.ConnectionAborted) ||

(error == SocketError.ConnectionRefused) ||

(error == SocketError.ConnectionReset) ||

(error == SocketError.OperationAborted) ||

(error == SocketError.Shutdown))

return;

OnError(error);

}

#endregion

#region IDisposable implementation

/// <summary>

/// Disposed flag

/// </summary>

public bool IsDisposed { get; private set; }

/// <summary>

/// Client socket disposed flag

/// </summary>

public bool IsSocketDisposed { get; private set; } = true;

// Implement IDisposable.

public void Dispose()

{

Dispose(true);

GC.SuppressFinalize(this);

}

protected virtual void Dispose(bool disposingManagedResources)

{

// The idea here is that Dispose(Boolean) knows whether it is

// being called to do explicit cleanup (the Boolean is true)

// versus being called due to a garbage collection (the Boolean

// is false). This distinction is useful because, when being

// disposed explicitly, the Dispose(Boolean) method can safely

// execute code using reference type fields that refer to other

// objects knowing for sure that these other objects have not been

// finalized or disposed of yet. When the Boolean is false,

// the Dispose(Boolean) method should not execute code that

// refer to reference type fields because those objects may

// have already been finalized."

if (!IsDisposed)

{

if (disposingManagedResources)

{

// Dispose managed resources here...

DisconnectAsync();

}

// Dispose unmanaged resources here...

// Set large fields to null here...

// Mark as disposed.

IsDisposed = true;

}

}

#endregion

}