add CancellationToken (#13)

* add CancellationToken

Author: zivillian

Graphite/GraphiteClient.cs

@@ -4,6 +4,7 @@
 using System.Net.Http;
 using System.Net.Sockets;
 using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 using ahd.Graphite.Base;
 using ahd.Graphite.Exceptions;
@@ -81,10 +82,11 @@ public GraphiteClient()
         /// </summary>
         /// <param name="series">metric path</param>
         /// <param name="value">metric value</param>
+        /// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="CancellationToken.None"/>.</param>
         /// <returns></returns>
-        public Task SendAsync(string series, double value)
+        public Task SendAsync(string series, double value, CancellationToken cancellationToken = default(CancellationToken))
         {
-            return SendAsync(series, value, DateTime.Now);
+            return SendAsync(series, value, DateTime.Now, cancellationToken);
         }
 
         /// <summary>
@@ -93,10 +95,11 @@ public Task SendAsync(string series, double value)
         /// <param name="series">metric path</param>
         /// <param name="value">metric value</param>
         /// <param name="timestamp">metric timestamp</param>
+        /// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="CancellationToken.None"/>.</param>
         /// <returns></returns>
-        public Task SendAsync(string series, double value, DateTime timestamp)
+        public Task SendAsync(string series, double value, DateTime timestamp, CancellationToken cancellationToken = default(CancellationToken))
         {
-            return SendAsync(new Datapoint(series, value, timestamp));
+            return SendAsync(new []{new Datapoint(series, value, timestamp)}, cancellationToken);
         }
 
         /// <summary>
@@ -105,35 +108,59 @@ public Task SendAsync(string series, double value, DateTime timestamp)
         /// <param name="datapoints"></param>
         /// <returns></returns>
         public Task SendAsync(params Datapoint[] datapoints)
+        {
+            return SendAsync(datapoints, CancellationToken.None);
+        }
+
+        /// <summary>
+        /// Send a list of datapoints in up to <see cref="BatchSize"/> batches
+        /// </summary>
+        /// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="CancellationToken.None"/>.</param>
+        /// <param name="datapoints"></param>
+        /// <returns></returns>
+        public Task SendAsync(CancellationToken cancellationToken, params Datapoint[] datapoints)
+        {
+            return SendAsync(datapoints, cancellationToken);
+        }
+
+        /// <summary>
+        /// Send a list of datapoints in up to <see cref="BatchSize"/> batches
+        /// </summary>
+        /// <param name="datapoints"></param>
+        /// <param name="cancellationToken">The token to monitor for cancellation requests.</param>
+        /// <returns></returns>
+        public Task SendAsync(Datapoint[] datapoints, CancellationToken cancellationToken)
         {
             ICollection<Datapoint> points = datapoints;
-            return SendAsync(points);
+            return SendAsync(points, cancellationToken);
         }
 
         /// <summary>
         /// Send a list of datapoints in up to <see cref="BatchSize"/> batches
         /// </summary>
         /// <param name="datapoints"></param>
+        /// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="CancellationToken.None"/>.</param>
         /// <returns></returns>
-        public async Task SendAsync(ICollection<Datapoint> datapoints)
+        public async Task SendAsync(ICollection<Datapoint> datapoints, CancellationToken cancellationToken = default(CancellationToken))
         {
             if (datapoints == null || datapoints.Count == 0) throw new ArgumentNullException(nameof(datapoints));
             var batches = GetBatches(datapoints);
             foreach (var batch in batches)
             {
-                await SendInternalAsync(batch).ConfigureAwait(false);
+                await SendInternalAsync(batch, cancellationToken).ConfigureAwait(false);
             }
         }
 
-        private async Task SendInternalAsync(ICollection<Datapoint> datapoints)
+        private async Task SendInternalAsync(ICollection<Datapoint> datapoints, CancellationToken cancellationToken)
         {
             using (var client = new TcpClient(AddressFamily.InterNetworkV6))
             {
                 client.Client.DualMode = true;
                 await client.ConnectAsync(Host, Formatter.Port).ConfigureAwait(false);
+                cancellationToken.ThrowIfCancellationRequested();
                 using (var stream = client.GetStream())
                 {
-                    await Formatter.WriteAsync(stream, datapoints).ConfigureAwait(false);
+                    await Formatter.WriteAsync(stream, datapoints, cancellationToken).ConfigureAwait(false);
                 }
             }
         }

Graphite/IGraphiteFormatter.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.IO;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace ahd.Graphite
@@ -15,7 +16,8 @@ public interface IGraphiteFormatter
         /// </summary>
         /// <param name="stream">target stream to write the data to</param>
         /// <param name="datapoints">list of datapoints to send</param>
-        Task WriteAsync(Stream stream, ICollection<Datapoint> datapoints);
+        /// <param name="cancellationToken">The token to monitor for cancellation requests. The default value is <see cref="CancellationToken.None"/>.</param>
+        Task WriteAsync(Stream stream, ICollection<Datapoint> datapoints, CancellationToken cancellationToken = default(CancellationToken));
 
         /// <summary>
         /// Format the datapoints and write to the target stream

Graphite/PickleGraphiteFormatter.cs

@@ -2,6 +2,7 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Razorvine.Pickle;
 
@@ -33,7 +34,7 @@ public PickleGraphiteFormatter()
         }
 
         /// <inheritdoc/>
-        public async Task WriteAsync(Stream stream, ICollection<Datapoint> datapoints)
+        public async Task WriteAsync(Stream stream, ICollection<Datapoint> datapoints, CancellationToken cancellationToken = default(CancellationToken))
         {
             using (var pickler = new Pickler())
             {
@@ -45,10 +46,10 @@ public async Task WriteAsync(Stream stream, ICollection<Datapoint> datapoints)
                 var sizeBytes = BitConverter.GetBytes(size);
                 if (BitConverter.IsLittleEndian)
                     Array.Reverse(sizeBytes);
-                await stream.WriteAsync(sizeBytes, 0, sizeBytes.Length).ConfigureAwait(false);
+                await stream.WriteAsync(sizeBytes, 0, sizeBytes.Length, cancellationToken).ConfigureAwait(false);
 
-                await stream.WriteAsync(pickled, 0, pickled.Length).ConfigureAwait(false);
-                await stream.FlushAsync().ConfigureAwait(false);
+                await stream.WriteAsync(pickled, 0, pickled.Length, cancellationToken).ConfigureAwait(false);
+                await stream.FlushAsync(cancellationToken).ConfigureAwait(false);
             }
         }
 

Graphite/PlaintextGraphiteFormatter.cs

@@ -1,6 +1,7 @@
 using System.Collections.Generic;
 using System.Globalization;
 using System.IO;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace ahd.Graphite
@@ -31,14 +32,16 @@ public PlaintextGraphiteFormatter(ushort port):this()
         public ushort Port { get; }
 
         /// <inheritdoc/>
-        public async Task WriteAsync(Stream stream, ICollection<Datapoint> datapoints)
+        public async Task WriteAsync(Stream stream, ICollection<Datapoint> datapoints, CancellationToken cancellationToken = default(CancellationToken))
         {
             using (var writer = new StreamWriter(stream) {NewLine = "\n"})
             {
                 foreach (var datapoint in datapoints)
                 {
+                    cancellationToken.ThrowIfCancellationRequested();
                     await writer.WriteLineAsync($"{datapoint.Series} {datapoint.Value.ToString(CultureInfo.InvariantCulture)} {datapoint.UnixTimestamp}").ConfigureAwait(false);
                 }
+                cancellationToken.ThrowIfCancellationRequested();
                 await writer.FlushAsync().ConfigureAwait(false);
             }
         }