using System; using System.Net; using System.Net.Sockets; using System.Text; using System.Threading; namespace NetCoreServer { /// /// Unix Domain Socket client is used to read/write data from/into the connected Unix Domain Socket server /// /// Thread-safe public class UdsClient : IDisposable { /// /// Initialize Unix Domain Socket client with a given socket path /// /// Socket path public UdsClient(string path) : this(new UnixDomainSocketEndPoint(path)) {} /// /// Initialize Unix Domain Socket client with a given Unix Domain Socket endpoint /// /// Unix Domain Socket endpoint public UdsClient(UnixDomainSocketEndPoint endpoint) { Id = Guid.NewGuid(); Endpoint = endpoint; } /// /// Client Id /// public Guid Id { get; } /// /// Endpoint /// public EndPoint Endpoint { get; private set; } /// /// Socket /// public Socket Socket { get; private set; } /// /// Number of bytes pending sent by the client /// public long BytesPending { get; private set; } /// /// Number of bytes sending by the client /// public long BytesSending { get; private set; } /// /// Number of bytes sent by the client /// public long BytesSent { get; private set; } /// /// Number of bytes received by the client /// public long BytesReceived { get; private set; } /// /// Option: receive buffer limit /// public int OptionReceiveBufferLimit { get; set; } = 0; /// /// Option: receive buffer size /// public int OptionReceiveBufferSize { get; set; } = 8192; /// /// Option: send buffer limit /// public int OptionSendBufferLimit { get; set; } = 0; /// /// Option: send buffer size /// public int OptionSendBufferSize { get; set; } = 8192; #region Connect/Disconnect client private SocketAsyncEventArgs _connectEventArg; /// /// Is the client connecting? /// public bool IsConnecting { get; private set; } /// /// Is the client connected? /// public bool IsConnected { get; private set; } /// /// Create a new socket object /// /// /// Method may be override if you need to prepare some specific socket object in your implementation. /// /// Socket object protected virtual Socket CreateSocket() { return new Socket(Endpoint.AddressFamily, SocketType.Stream, ProtocolType.IP); } /// /// Connect the client (synchronous) /// /// /// Please note that synchronous connect will not receive data automatically! /// You should use Receive() or ReceiveAsync() method manually after successful connection. /// /// 'true' if the client was successfully connected, 'false' if the client failed to connect 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; } /// /// Disconnect the client (synchronous) /// /// 'true' if the client was successfully disconnected, 'false' if the client is already disconnected 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; } /// /// Reconnect the client (synchronous) /// /// 'true' if the client was successfully reconnected, 'false' if the client is already reconnected public virtual bool Reconnect() { if (!Disconnect()) return false; return Connect(); } /// /// Connect the client (asynchronous) /// /// 'true' if the client was successfully connected, 'false' if the client failed to connect 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; } /// /// Disconnect the client (asynchronous) /// /// 'true' if the client was successfully disconnected, 'false' if the client is already disconnected public virtual bool DisconnectAsync() => Disconnect(); /// /// Reconnect the client (asynchronous) /// /// 'true' if the client was successfully reconnected, 'false' if the client is already reconnected 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; /// /// Send data to the server (synchronous) /// /// Buffer to send /// Size of sent data public virtual long Send(byte[] buffer) => Send(buffer.AsSpan()); /// /// Send data to the server (synchronous) /// /// Buffer to send /// Buffer offset /// Buffer size /// Size of sent data public virtual long Send(byte[] buffer, long offset, long size) => Send(buffer.AsSpan((int)offset, (int)size)); /// /// Send data to the server (synchronous) /// /// Buffer to send as a span of bytes /// Size of sent data public virtual long Send(ReadOnlySpan 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; } /// /// Send text to the server (synchronous) /// /// Text string to send /// Size of sent text public virtual long Send(string text) => Send(Encoding.UTF8.GetBytes(text)); /// /// Send text to the server (synchronous) /// /// Text to send as a span of characters /// Size of sent text public virtual long Send(ReadOnlySpan text) => Send(Encoding.UTF8.GetBytes(text.ToArray())); /// /// Send data to the server (asynchronous) /// /// Buffer to send /// 'true' if the data was successfully sent, 'false' if the client is not connected public virtual bool SendAsync(byte[] buffer) => SendAsync(buffer.AsSpan()); /// /// Send data to the server (asynchronous) /// /// Buffer to send /// Buffer offset /// Buffer size /// 'true' if the data was successfully sent, 'false' if the client is not connected public virtual bool SendAsync(byte[] buffer, long offset, long size) => SendAsync(buffer.AsSpan((int)offset, (int)size)); /// /// Send data to the server (asynchronous) /// /// Buffer to send as a span of bytes /// 'true' if the data was successfully sent, 'false' if the client is not connected public virtual bool SendAsync(ReadOnlySpan 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; } /// /// Send text to the server (asynchronous) /// /// Text string to send /// 'true' if the text was successfully sent, 'false' if the client is not connected public virtual bool SendAsync(string text) => SendAsync(Encoding.UTF8.GetBytes(text)); /// /// Send text to the server (asynchronous) /// /// Text to send as a span of characters /// 'true' if the text was successfully sent, 'false' if the client is not connected public virtual bool SendAsync(ReadOnlySpan text) => SendAsync(Encoding.UTF8.GetBytes(text.ToArray())); /// /// Receive data from the server (synchronous) /// /// Buffer to receive /// Size of received data public virtual long Receive(byte[] buffer) { return Receive(buffer, 0, buffer.Length); } /// /// Receive data from the server (synchronous) /// /// Buffer to receive /// Buffer offset /// Buffer size /// Size of received data 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; } /// /// Receive text from the server (synchronous) /// /// Text size to receive /// Received text public virtual string Receive(long size) { var buffer = new byte[size]; var length = Receive(buffer); return Encoding.UTF8.GetString(buffer, 0, (int)length); } /// /// Receive data from the server (asynchronous) /// public virtual void ReceiveAsync() { // Try to receive data from the server TryReceive(); } /// /// Try to receive new data /// 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) {} } } /// /// Try to send pending data /// 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) {} } } /// /// Clear send/receive buffers /// private void ClearBuffers() { lock (_sendLock) { // Clear send buffers _sendBufferMain.Clear(); _sendBufferFlush.Clear(); _sendBufferFlushOffset= 0; // Update statistic BytesPending = 0; BytesSending = 0; } } #endregion #region IO processing /// /// This method is called whenever a receive or send operation is completed on a socket /// 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"); } } /// /// This method is invoked when an asynchronous connect operation completes /// 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(); } } /// /// This method is invoked when an asynchronous receive operation completes /// 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; } /// /// This method is invoked when an asynchronous send operation completes /// 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 /// /// Handle client connecting notification /// protected virtual void OnConnecting() {} /// /// Handle client connected notification /// protected virtual void OnConnected() {} /// /// Handle client disconnecting notification /// protected virtual void OnDisconnecting() {} /// /// Handle client disconnected notification /// protected virtual void OnDisconnected() {} /// /// Handle buffer received notification /// /// Received buffer /// Received buffer offset /// Received buffer size /// /// Notification is called when another part of buffer was received from the server /// protected virtual void OnReceived(byte[] buffer, long offset, long size) {} /// /// Handle buffer sent notification /// /// Size of sent buffer /// Size of pending buffer /// /// 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. /// protected virtual void OnSent(long sent, long pending) {} /// /// Handle empty send buffer notification /// /// /// 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. /// protected virtual void OnEmpty() {} /// /// Handle error notification /// /// Socket error code protected virtual void OnError(SocketError error) {} #endregion #region Error handling /// /// Send error notification /// /// Socket error code 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 /// /// Disposed flag /// public bool IsDisposed { get; private set; } /// /// Client socket disposed flag /// 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 } }