Support reading from last snapshot event

SimpleEventStore.AzureDocumentDb/AzureDocumentDbStorageEngine.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Collections.ObjectModel;
 using System.Linq;
 using System.Threading.Tasks;
 using Microsoft.Azure.Documents;
@@ -94,6 +95,40 @@ public async Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string s
             return events.AsReadOnly();
         }
 
+        public async Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwardsFromLast(string streamId, Predicate<StorageEvent> readFromHere)
+        {
+            var eventsQuery = client.CreateDocumentQuery<DocumentDbStorageEvent>(commitsLink)
+                .Where(x => x.StreamId == streamId)
+                .OrderByDescending(x => x.EventNumber)
+                .AsDocumentQuery();
+
+            var eventsInReverseOrder = new List<StorageEvent>();
+            var finished = false;
+
+            while (!finished && eventsQuery.HasMoreResults)
+            {
+                var response = await eventsQuery.ExecuteNextAsync<DocumentDbStorageEvent>();
+                loggingOptions.OnSuccess(ResponseInformation.FromReadResponse(nameof(ReadStreamForwardsFromLast), response));
+
+                foreach (var e in response)
+                {
+                    var storageEvent = e.ToStorageEvent(typeMap);
+                    eventsInReverseOrder.Add(storageEvent);
+
+                    if (readFromHere(storageEvent))
+                    {
+                        finished = true;
+                        break;
+                    }
+                }
+            }
+
+            return eventsInReverseOrder
+                .Reverse<StorageEvent>()
+                .ToList()
+                .AsReadOnly();
+        }
+
         public async Task DeleteStream(string streamId)
         {
             while (true)

SimpleEventStore.Tests/EventStoreReading.cs

@@ -63,5 +63,24 @@ public async Task when_reading_a_stream_only_the_required_events_are_returned()
             Assert.That(events.Count, Is.EqualTo(1));
             Assert.That(events.First().EventBody, Is.InstanceOf<OrderDispatched>());
         }
+
+        [Test]
+        public async Task when_reading_a_stream_from_snapshot_only_subsequent_events_are_returned()
+        {
+            var streamId = Guid.NewGuid().ToString();
+            var subject = await GetEventStore();
+
+            await subject.AppendToStream(streamId, 0, new EventData(Guid.NewGuid(), new OrderCreated(streamId)));
+            await subject.AppendToStream(streamId, 1, new EventData(Guid.NewGuid(), new OrderSnapshot(streamId)));
+            await subject.AppendToStream(streamId, 2, new EventData(Guid.NewGuid(), new OrderDispatched(streamId)));
+
+            var events = await subject.ReadStreamForwardsFromLast(
+                streamId,
+                storageEvent => storageEvent.EventBody is OrderSnapshot);
+
+            Assert.That(events.Count, Is.EqualTo(2));
+            Assert.That(events.First().EventBody, Is.InstanceOf<OrderSnapshot>());
+            Assert.That(events.Skip(1).Single().EventBody, Is.InstanceOf<OrderDispatched>());
+        }
     }
 }

SimpleEventStore.Tests/Events/OrderSnapshot.cs

@@ -0,0 +1,12 @@
+namespace SimpleEventStore.Tests.Events
+{
+    public class OrderSnapshot
+    {
+        public string OrderId { get; private set; }
+
+        public OrderSnapshot(string orderId)
+        {
+            OrderId = orderId;
+        }
+    }
+}

SimpleEventStore/EventStore.cs

@@ -32,7 +32,7 @@ public Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamI
         {
             Guard.IsNotNullOrEmpty(nameof(streamId), streamId);
 
-            return engine.ReadStreamForwards(streamId, 1, Int32.MaxValue);
+            return engine.ReadStreamForwards(streamId, 1, int.MaxValue);
         }
 
         public Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamId, int startPosition, int numberOfEventsToRead)
@@ -42,6 +42,21 @@ public Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamI
             return engine.ReadStreamForwards(streamId, startPosition, numberOfEventsToRead);
         }
 
+        public Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwardsFromLast(string streamId, Predicate<StorageEvent> readFromHere)
+        {
+            Guard.IsNotNullOrEmpty(nameof(streamId), streamId);
+            Guard.IsNotNull(nameof(readFromHere), readFromHere);
+
+            return engine.ReadStreamForwardsFromLast(streamId, readFromHere);
+        }
+
+        public Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwardsFromLast<TSnapshot>(string streamId)
+        {
+            return ReadStreamForwardsFromLast(
+                streamId,
+                storageEvent => storageEvent.EventBody is TSnapshot);
+        }
+
         public Task DeleteStream(string streamId)
         {
             Guard.IsNotNullOrEmpty(nameof(streamId), streamId);

SimpleEventStore/IStorageEngine.cs

@@ -10,6 +10,8 @@ public interface IStorageEngine
 
         Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamId, int startPosition, int numberOfEventsToRead);
 
+        Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwardsFromLast(string streamId, Predicate<StorageEvent> readFromHere);
+
         Task<IStorageEngine> Initialise();
 
         Task DeleteStream(string streamId);

SimpleEventStore/InMemory/EnumerableExtensions.cs

@@ -0,0 +1,21 @@
+using System;
+using System.Collections.Generic;
+
+namespace SimpleEventStore.InMemory
+{
+    static class EnumerableExtensions
+    {
+        public static IEnumerable<T> TakeUntilImmediatelyAfter<T>(this IEnumerable<T> source, Predicate<T> stopImmediatelyAfter)
+        {
+            foreach (var item in source)
+            {
+                yield return item;
+
+                if (stopImmediatelyAfter(item))
+                {
+                    yield break;
+                }
+            }
+        }
+    }
+}

SimpleEventStore/InMemory/InMemoryStorageEngine.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
@@ -65,7 +66,29 @@ public Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwards(string streamI
                 return Task.FromResult(EmptyStream);
             }
 
-            IReadOnlyCollection<StorageEvent> stream = streams[streamId].Skip(startPosition - 1).Take(numberOfEventsToRead).ToList().AsReadOnly();
+            IReadOnlyCollection<StorageEvent> stream = streams[streamId]
+                .Skip(startPosition - 1)
+                .Take(numberOfEventsToRead)
+                .ToList()
+                .AsReadOnly();
+
+            return Task.FromResult(stream);
+        }
+
+        public Task<IReadOnlyCollection<StorageEvent>> ReadStreamForwardsFromLast(string streamId, Predicate<StorageEvent> readFromHere)
+        {
+            if (!streams.ContainsKey(streamId))
+            {
+                return Task.FromResult(EmptyStream);
+            }
+
+            IReadOnlyCollection<StorageEvent> stream = streams[streamId]
+                .Reverse<StorageEvent>()
+                .TakeUntilImmediatelyAfter(readFromHere)
+                .Reverse()
+                .ToList()
+                .AsReadOnly();
+
             return Task.FromResult(stream);
         }