Added CancellationToken support to Event Store (#45)

Added support for passing Cancellation Tokens down to the Cosmos SDK to
allow for better control over timeouts.

Due to the addition of optional parameters, this is a runtime breaking
change, but not a compile time one.

Author: afscrome

src/SimpleEventStore/SimpleEventStore.AzureDocumentDb/AzureDocumentDbStorageEngine.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Azure.Documents;
 using Microsoft.Azure.Documents.Client;
@@ -39,19 +40,28 @@ internal AzureDocumentDbStorageEngine(DocumentClient client, string databaseName
             this.databaseUri = UriFactory.CreateDatabaseUri(databaseName);
         }
 
-        public async Task<IStorageEngine> Initialise()
+        public async Task<IStorageEngine> Initialise(CancellationToken cancellationToken = default)
         {
+            cancellationToken.ThrowIfCancellationRequested();
             await CreateDatabaseIfItDoesNotExist();
+
+            cancellationToken.ThrowIfCancellationRequested();
             await CreateCollectionIfItDoesNotExist();
+            
+            cancellationToken.ThrowIfCancellationRequested();
             await CreateAppendStoredProcedureIfItDoesNotExist();
 
+
+            cancellationToken.ThrowIfCancellationRequested();
             await SetDatabaseOfferThroughput();
+            
+            cancellationToken.ThrowIfCancellationRequested();
             await SetCollectionOfferThroughput();
 
             return this;
         }
 
-        public async Task AppendToStream(string streamId, IEnumerable<StorageEvent> events)
+        public async Task AppendToStream(string streamId, IEnumerable<StorageEvent> events, CancellationToken cancellationToken = default)
         {
             var docs = events.Select(d => DocumentDbStorageEvent.FromStorageEvent(d, this.typeMap, this.jsonSerializer)).ToList();
 
@@ -60,6 +70,7 @@ public async Task AppendToStream(string streamId, IEnumerable<StorageEvent> even
                 var result = await this.client.ExecuteStoredProcedureAsync<dynamic>(
                     storedProcLink,
                     new RequestOptions { PartitionKey = new PartitionKey(streamId), ConsistencyLevel = this.collectionOptions.ConsistencyLevel },
+                    cancellationToken,
                     docs);
 
                 loggingOptions.OnSuccess(ResponseInformation.FromWriteResponse(nameof(AppendToStream), result));
@@ -75,7 +86,7 @@ public async Task AppendToStream(string streamId, IEnumerable<StorageEvent> even
             }
         }
 
-        public async Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamId, int startPosition, int numberOfEventsToRead)
+        public async Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamId, int startPosition, int numberOfEventsToRead, CancellationToken cancellationToken = default)
         {
             int endPosition = numberOfEventsToRead == int.MaxValue ? int.MaxValue : startPosition + numberOfEventsToRead;
 
@@ -88,7 +99,7 @@ public async Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string s
 
             while (eventsQuery.HasMoreResults)
             {
-                var response = await eventsQuery.ExecuteNextAsync<DocumentDbStorageEvent>();
+                var response = await eventsQuery.ExecuteNextAsync<DocumentDbStorageEvent>(cancellationToken);
                 loggingOptions.OnSuccess(ResponseInformation.FromReadResponse(nameof(ReadStreamForwards), response));
 
                 foreach (var e in response)

src/SimpleEventStore/SimpleEventStore/IStorageEngine.cs

@@ -1,15 +1,16 @@
 using System;
 using System.Collections.Generic;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace SimpleEventStore
 {
     public interface IStorageEngine
     {
-        Task AppendToStream(string streamId, IEnumerable<StorageEvent> events);
+        Task AppendToStream(string streamId, IEnumerable<StorageEvent> events, CancellationToken cancellationToken = default);
 
-        Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamId, int startPosition, int numberOfEventsToRead);
+        Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamId, int startPosition, int numberOfEventsToRead, CancellationToken cancellationToken = default);
 
-        Task<IStorageEngine> Initialise();
+        Task<IStorageEngine> Initialise(CancellationToken cancellationToken = default);
     }
 }

src/SimpleEventStore/SimpleEventStore/InMemory/InMemoryStorageEngine.cs

@@ -1,6 +1,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 
 namespace SimpleEventStore.InMemory
@@ -12,7 +13,7 @@ public class InMemoryStorageEngine : IStorageEngine
         private readonly ConcurrentDictionary<string, List<StorageEvent>> streams = new ConcurrentDictionary<string, List<StorageEvent>>();
         private readonly List<StorageEvent> allEvents = new List<StorageEvent>();
 
-        public Task AppendToStream(string streamId, IEnumerable<StorageEvent> events)
+        public Task AppendToStream(string streamId, IEnumerable<StorageEvent> events, CancellationToken cancellationToken = default)
         {
             return Task.Run(() =>
             {
@@ -41,7 +42,7 @@ private void AddEventsToAllStream(IEnumerable<StorageEvent> events)
             }
         }
 
-        public Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamId, int startPosition, int numberOfEventsToRead)
+        public Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamId, int startPosition, int numberOfEventsToRead, CancellationToken cancellationToken = default)
         {
             if (!streams.ContainsKey(streamId))
             {
@@ -52,7 +53,7 @@ public Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamI
             return Task.FromResult(stream);
         }
 
-        public Task<IStorageEngine> Initialise()
+        public Task<IStorageEngine> Initialise(CancellationToken cancellationToken = default)
         {
             return Task.FromResult<IStorageEngine>(this);
         }