Async methods in TcpSocket

SharpAdbClient.Tests/DummyTcpSocket.cs

@@ -26,6 +26,8 @@ public bool Connected
 
         public int ReceiveBufferSize
         { get; set; } = 1024;
+        public int ReceiveTimeout 
+        { get; set; } = -1;
 
         public void Close()
         {
@@ -37,6 +39,23 @@ public void Connect(EndPoint endPoint)
             this.Connected = true;
         }
 
+        public Task ConnectAsync(EndPoint endPoint, CancellationToken cancellationToken)
+        {
+            Connect(endPoint);
+            return Task.CompletedTask;
+        }
+
+        public void Reconnect()
+        {
+            throw new NotImplementedException();
+        }
+
+        public Task ReconnectAsync(CancellationToken cancellationToken)
+        {
+            Reconnect();
+            return Task.CompletedTask;
+        }
+
         public void Dispose()
         {
             this.Connected = false;
@@ -64,14 +83,15 @@ public int Send(byte[] buffer, int offset, int size, SocketFlags socketFlags)
             return size;
         }
 
-        public byte[] GetBytesSent()
+        public Task<int> SendAsync(byte[] buffer, int offset, int size, SocketFlags socketFlags, CancellationToken cancellationToken)
         {
-            return this.OutputStream.ToArray();
+            this.OutputStream.Write(buffer, offset, size);
+            return Task.FromResult(size);
         }
 
-        public void Reconnect()
+        public byte[] GetBytesSent()
         {
-            throw new NotImplementedException();
+            return this.OutputStream.ToArray();
         }
     }
 }

SharpAdbClient/ITcpSocket.cs

@@ -36,6 +36,16 @@ int ReceiveBufferSize
             set;
         }
 
+        /// <summary>
+        ///  Gets or sets a value that specifies the amount of time after which a synchronous
+        //   Receive call will time out.
+        /// </summary>
+        int ReceiveTimeout
+        {
+            get;
+            set;
+        }
+
         /// <summary>
         /// Establishes a connection to a remote host.
         /// </summary>
@@ -44,12 +54,32 @@ int ReceiveBufferSize
         /// </param>
         void Connect(EndPoint endPoint);
 
+        /// <summary>
+        /// Establishes a connection to a remote host asynchronously.
+        /// </summary>
+        /// <param name="endPoint">
+        /// An <see cref="EndPoint"/> that represents the remote device.
+        /// </param>
+        /// <param name="cancellationToken">
+        /// A <see cref="CancellationToken"/> which can be used to cancel the asynchronous task.
+        /// </param>
+        Task ConnectAsync(EndPoint endPoint, CancellationToken cancellationToken);
+
         /// <summary>
         /// Re-establishes the connection to a remote host. Assumes you have resolved the reason that caused the
         /// socket to disconnect.
         /// </summary>
         void Reconnect();
 
+        /// <summary>
+        /// Re-establishes the connection to a remote host asynchronously. Assumes you have resolved the reason 
+        /// that caused the socket to disconnect.
+        /// </summary>
+        /// <param name="cancellationToken">
+        /// A <see cref="CancellationToken"/> which can be used to cancel the asynchronous task.
+        /// </param>
+        Task ReconnectAsync(CancellationToken cancellationToken);
+
         /// <summary>
         /// Sends the specified number of bytes of data to a connected
         /// <see cref="ITcpSocket"/>, starting at the specified <paramref name="offset"/>,
@@ -72,6 +102,32 @@ int ReceiveBufferSize
         /// </returns>
         int Send(byte[] buffer, int offset, int size, SocketFlags socketFlags);
 
+
+        /// <summary>
+        /// Sends the specified number of bytes of data asynchronously to a connected
+        /// <see cref="ITcpSocket"/>, starting at the specified <paramref name="offset"/>,
+        /// and using the specified <paramref name="socketFlags"/>.
+        /// </summary>
+        /// <param name="buffer">
+        /// An array of type Byte that contains the data to be sent.
+        /// </param>
+        /// <param name="offset">
+        /// The position in the data buffer at which to begin sending data.
+        /// </param>
+        /// <param name="size">
+        /// The number of bytes to send.
+        /// </param>
+        /// <param name="socketFlags">
+        /// A bitwise combination of the SocketFlags values.
+        /// </param>
+        /// <param name="cancellationToken">
+        /// A <see cref="CancellationToken"/> which can be used to cancel the asynchronous task.
+        /// </param>
+        /// <returns>
+        /// The number of bytes sent to the Socket.
+        /// </returns>
+        Task<int> SendAsync(byte[] buffer, int offset, int size, SocketFlags socketFlags, CancellationToken cancellationToken);
+
         /// <summary>
         /// Receives the specified number of bytes from a bound <see cref="ITcpSocket"/> into the specified offset position of the
         /// receive buffer, using the specified SocketFlags.

SharpAdbClient/SocketExtensions.cs

@@ -59,5 +59,51 @@ public static Task<int> ReceiveAsync(
             ArraySegment<byte> array = new ArraySegment<byte>(buffer, offset, size);
             return socket.ReceiveAsync(array, socketFlags);
         }
+
+        /// <summary>
+        /// Asynchronously sends data to a connected socket.
+        /// </summary>
+        /// <param name="socket">
+        /// The socket from which to read data.
+        /// </param>
+        /// <param name="buffer">
+        /// An array of type <see cref="byte"/> that is the storage location for
+        /// the received data.
+        /// </param>
+        /// <param name="offset">
+        /// The zero-based position in the <paramref name="buffer"/> parameter at which to
+        /// start storing data.
+        /// </param>
+        /// <param name="size">
+        /// The number of bytes to receive.
+        /// </param>
+        /// <param name="socketFlags">
+        /// A bitwise combination of the <see cref="SocketFlags"/> values.
+        /// </param>
+        /// <param name="cancellationToken">
+        /// A <see cref="CancellationToken"/> which can be used to cancel the asynchronous task.
+        /// </param>
+        /// <remarks>
+        /// Cancelling the task will also close the socket.
+        /// </remarks>
+        /// <returns>
+        /// The number of bytes received.
+        /// </returns>
+        public static Task<int> SendAsync(
+            this Socket socket,
+            byte[] buffer,
+            int offset,
+            int size,
+            SocketFlags socketFlags,
+            CancellationToken cancellationToken)
+        {
+            // Register a callback so that when a cancellation is requested, the socket is closed.
+            // This will cause an ObjectDisposedException to bubble up via TrySetResult, which we can catch
+            // and convert to a TaskCancelledException - which is the exception we expect.
+            var cancellationTokenRegistration = cancellationToken.Register(() => socket.Dispose());
+
+            ArraySegment<byte> array = new ArraySegment<byte>(buffer, offset, size);
+            return socket.SendAsync(array, socketFlags);
+        }
     }
 }

SharpAdbClient/TcpSocket.cs

@@ -51,6 +51,20 @@ public int ReceiveBufferSize
             }
         }
 
+        /// <inheritdoc/>
+        public int ReceiveTimeout
+        {
+            get
+            {
+                return this.socket.ReceiveTimeout;
+            }
+
+            set
+            {
+                this.socket.ReceiveTimeout = value;
+            }
+        }
+
         /// <inheritdoc/>
         public void Connect(EndPoint endPoint)
         {
@@ -64,6 +78,56 @@ public void Connect(EndPoint endPoint)
             this.endPoint = endPoint;
         }
 
+        /// <inheritdoc/>
+        public Task ConnectAsync(EndPoint endPoint, CancellationToken cancellationToken)
+        {
+            if (!(endPoint is IPEndPoint || endPoint is DnsEndPoint))
+            {
+                throw new NotSupportedException();
+            }
+
+            return Task.Run(() =>
+            {
+                var completedEvent = new ManualResetEvent(false);
+                bool successful = false;
+                var asyncEvent = new SocketAsyncEventArgs();
+                asyncEvent.RemoteEndPoint = endPoint;
+                asyncEvent.Completed += (sender, args) =>
+                {
+                    successful = true;
+                    completedEvent.Set();
+                };
+                cancellationToken.Register(() =>
+                {
+                    Socket.CancelConnectAsync(asyncEvent);
+                    completedEvent.Set();
+                });
+
+                this.socket.ConnectAsync(asyncEvent);
+                completedEvent.WaitOne();
+                if (successful)
+                {
+                    this.socket.Blocking = true;
+                    this.endPoint = endPoint;
+                }
+                if (cancellationToken.IsCancellationRequested)
+                    throw new TaskCanceledException();
+            });
+        }
+
+        /// <inheritdoc/>
+        public Task ReconnectAsync(CancellationToken cancellationToken)
+        {
+            if (this.socket.Connected)
+            {
+                // Already connected - nothing to do.
+                return Task.CompletedTask;
+            }
+
+            this.socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
+            return this.ConnectAsync(this.endPoint, cancellationToken);
+        }
+
         /// <inheritdoc/>
         public void Reconnect()
         {
@@ -90,9 +154,9 @@ public int Send(byte[] buffer, int offset, int size, SocketFlags socketFlags)
         }
 
         /// <inheritdoc/>
-        public Stream GetStream()
+        public Task<int> SendAsync(byte[] buffer, int offset, int size, SocketFlags socketFlags, CancellationToken cancellationToken)
         {
-            return new NetworkStream(this.socket);
+            return this.socket.SendAsync(buffer, offset, size, socketFlags, cancellationToken);
         }
 
         /// <inheritdoc/>
@@ -106,5 +170,11 @@ public Task<int> ReceiveAsync(byte[] buffer, int offset, int size, SocketFlags s
         {
             return this.socket.ReceiveAsync(buffer, offset, size, socketFlags, cancellationToken);
         }
+
+        /// <inheritdoc/>
+        public Stream GetStream()
+        {
+            return new NetworkStream(this.socket);
+        }
     }
 }