using System; using System.Buffers; using System.Collections.Generic; using System.Diagnostics; using System.IO.Pipelines; using System.Net.Sockets; namespace MQTTnet.AspNetCore.Client.Tcp { public class SocketSender { private readonly Socket _socket; private readonly SocketAsyncEventArgs _eventArgs = new SocketAsyncEventArgs(); private readonly SocketAwaitable _awaitable; private List> _bufferList; public SocketSender(Socket socket, PipeScheduler scheduler) { _socket = socket; _awaitable = new SocketAwaitable(scheduler); _eventArgs.UserToken = _awaitable; _eventArgs.Completed += (_, e) => ((SocketAwaitable)e.UserToken).Complete(e.BytesTransferred, e.SocketError); } public SocketAwaitable SendAsync(in ReadOnlySequence buffers) { if (buffers.IsSingleSegment) { return SendAsync(buffers.First); } #if NETCOREAPP2_1 if (!_eventArgs.MemoryBuffer.Equals(Memory.Empty)) #else if (_eventArgs.Buffer != null) #endif { _eventArgs.SetBuffer(null, 0, 0); } _eventArgs.BufferList = GetBufferList(buffers); if (!_socket.SendAsync(_eventArgs)) { _awaitable.Complete(_eventArgs.BytesTransferred, _eventArgs.SocketError); } return _awaitable; } private SocketAwaitable SendAsync(ReadOnlyMemory memory) { // The BufferList getter is much less expensive then the setter. if (_eventArgs.BufferList != null) { _eventArgs.BufferList = null; } #if NETCOREAPP2_1 _eventArgs.SetBuffer(MemoryMarshal.AsMemory(memory)); #else var segment = memory.GetArray(); _eventArgs.SetBuffer(segment.Array, segment.Offset, segment.Count); #endif if (!_socket.SendAsync(_eventArgs)) { _awaitable.Complete(_eventArgs.BytesTransferred, _eventArgs.SocketError); } return _awaitable; } private List> GetBufferList(in ReadOnlySequence buffer) { Debug.Assert(!buffer.IsEmpty); Debug.Assert(!buffer.IsSingleSegment); if (_bufferList == null) { _bufferList = new List>(); } else { // Buffers are pooled, so it's OK to root them until the next multi-buffer write. _bufferList.Clear(); } foreach (var b in buffer) { _bufferList.Add(b.GetArray()); } return _bufferList; } } }