From 376804a956a902c8b75f776267abed708d780b84 Mon Sep 17 00:00:00 2001
From: Dennis Doomen <dennis.doomen@avivasolutions.nl>
Date: Tue, 23 May 2017 18:04:18 +0200
Subject: [PATCH 1/3] Updated some of the packages.

---
 Samples/ExampleHost/App.config                |  6 ++-
 Samples/ExampleHost/ExampleHost.csproj        | 25 +++++------
 Samples/ExampleHost/JsonFileEventStore.cs     |  2 +-
 Samples/ExampleHost/packages.config           | 12 +++---
 .../LiquidProjections.Specs.csproj            | 41 ++++++++-----------
 Tests/LiquidProjections.Specs/packages.config | 16 ++++----
 6 files changed, 48 insertions(+), 54 deletions(-)

diff --git a/Samples/ExampleHost/App.config b/Samples/ExampleHost/App.config
index 97530a9..7c164fe 100644
--- a/Samples/ExampleHost/App.config
+++ b/Samples/ExampleHost/App.config
@@ -15,7 +15,7 @@
       </dependentAssembly>
       <dependentAssembly>
         <assemblyIdentity name="Microsoft.Owin" publicKeyToken="31bf3856ad364e35" culture="neutral" />
-        <bindingRedirect oldVersion="0.0.0.0-3.0.1.0" newVersion="3.0.1.0" />
+        <bindingRedirect oldVersion="0.0.0.0-3.1.0.0" newVersion="3.1.0.0" />
       </dependentAssembly>
       <dependentAssembly>
         <assemblyIdentity name="System.Web.Http.Owin" publicKeyToken="31bf3856ad364e35" culture="neutral" />
@@ -25,6 +25,10 @@
         <assemblyIdentity name="Microsoft.Owin.Hosting" publicKeyToken="31bf3856ad364e35" culture="neutral" />
         <bindingRedirect oldVersion="0.0.0.0-3.0.1.0" newVersion="3.0.1.0" />
       </dependentAssembly>
+      <dependentAssembly>
+        <assemblyIdentity name="Newtonsoft.Json" publicKeyToken="30ad4fe6b2a6aeed" culture="neutral" />
+        <bindingRedirect oldVersion="0.0.0.0-10.0.0.0" newVersion="10.0.0.0" />
+      </dependentAssembly>
     </assemblyBinding>
   </runtime>
 </configuration>
diff --git a/Samples/ExampleHost/ExampleHost.csproj b/Samples/ExampleHost/ExampleHost.csproj
index f8d37ac..c03f633 100644
--- a/Samples/ExampleHost/ExampleHost.csproj
+++ b/Samples/ExampleHost/ExampleHost.csproj
@@ -39,25 +39,20 @@
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
   </PropertyGroup>
   <ItemGroup>
-    <Reference Include="Microsoft.Owin, Version=3.0.1.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\Microsoft.Owin.3.0.1\lib\net45\Microsoft.Owin.dll</HintPath>
-      <Private>True</Private>
+    <Reference Include="Microsoft.Owin, Version=3.1.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\Microsoft.Owin.3.1.0\lib\net45\Microsoft.Owin.dll</HintPath>
     </Reference>
-    <Reference Include="Microsoft.Owin.Diagnostics, Version=3.0.1.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\Microsoft.Owin.Diagnostics.3.0.1\lib\net45\Microsoft.Owin.Diagnostics.dll</HintPath>
-      <Private>True</Private>
+    <Reference Include="Microsoft.Owin.Diagnostics, Version=3.1.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\Microsoft.Owin.Diagnostics.3.1.0\lib\net45\Microsoft.Owin.Diagnostics.dll</HintPath>
     </Reference>
-    <Reference Include="Microsoft.Owin.Host.HttpListener, Version=3.0.1.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\Microsoft.Owin.Host.HttpListener.3.0.1\lib\net45\Microsoft.Owin.Host.HttpListener.dll</HintPath>
-      <Private>True</Private>
+    <Reference Include="Microsoft.Owin.Host.HttpListener, Version=3.1.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\Microsoft.Owin.Host.HttpListener.3.1.0\lib\net45\Microsoft.Owin.Host.HttpListener.dll</HintPath>
     </Reference>
-    <Reference Include="Microsoft.Owin.Hosting, Version=3.0.1.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\Microsoft.Owin.Hosting.3.0.1\lib\net45\Microsoft.Owin.Hosting.dll</HintPath>
-      <Private>True</Private>
+    <Reference Include="Microsoft.Owin.Hosting, Version=3.1.0.0, Culture=neutral, PublicKeyToken=31bf3856ad364e35, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\Microsoft.Owin.Hosting.3.1.0\lib\net45\Microsoft.Owin.Hosting.dll</HintPath>
     </Reference>
-    <Reference Include="Newtonsoft.Json, Version=6.0.0.0, Culture=neutral, PublicKeyToken=30ad4fe6b2a6aeed, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\Newtonsoft.Json.6.0.4\lib\net45\Newtonsoft.Json.dll</HintPath>
-      <Private>True</Private>
+    <Reference Include="Newtonsoft.Json, Version=10.0.0.0, Culture=neutral, PublicKeyToken=30ad4fe6b2a6aeed, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\Newtonsoft.Json.10.0.2\lib\net45\Newtonsoft.Json.dll</HintPath>
     </Reference>
     <Reference Include="Owin, Version=1.0.0.0, Culture=neutral, PublicKeyToken=f0ebd12fd5e55cc5, processorArchitecture=MSIL">
       <HintPath>..\..\packages\Owin.1.0\lib\net40\Owin.dll</HintPath>
diff --git a/Samples/ExampleHost/JsonFileEventStore.cs b/Samples/ExampleHost/JsonFileEventStore.cs
index 3e7e1fa..068d3b1 100644
--- a/Samples/ExampleHost/JsonFileEventStore.cs
+++ b/Samples/ExampleHost/JsonFileEventStore.cs
@@ -72,7 +72,7 @@ private Task<Transaction[]> LoadNextPageAsync()
                             Body = JsonConvert.DeserializeObject(json, new JsonSerializerSettings
                             {
                                 TypeNameHandling = TypeNameHandling.All,
-                                TypeNameAssemblyFormat = FormatterAssemblyStyle.Full
+                                TypeNameAssemblyFormatHandling = TypeNameAssemblyFormatHandling.Full
                             })
                         });
                     }
diff --git a/Samples/ExampleHost/packages.config b/Samples/ExampleHost/packages.config
index fafe704..fa8d503 100644
--- a/Samples/ExampleHost/packages.config
+++ b/Samples/ExampleHost/packages.config
@@ -4,12 +4,12 @@
   <package id="Microsoft.AspNet.WebApi.Client" version="5.2.3" targetFramework="net45" />
   <package id="Microsoft.AspNet.WebApi.Core" version="5.2.3" targetFramework="net45" />
   <package id="Microsoft.AspNet.WebApi.Owin" version="5.2.3" targetFramework="net452" />
-  <package id="Microsoft.Owin" version="3.0.1" targetFramework="net452" />
-  <package id="Microsoft.Owin.Diagnostics" version="3.0.1" targetFramework="net452" />
-  <package id="Microsoft.Owin.Host.HttpListener" version="3.0.1" targetFramework="net452" />
-  <package id="Microsoft.Owin.Hosting" version="3.0.1" targetFramework="net452" />
-  <package id="Microsoft.Owin.SelfHost" version="3.0.1" targetFramework="net452" />
-  <package id="Newtonsoft.Json" version="6.0.4" targetFramework="net45" />
+  <package id="Microsoft.Owin" version="3.1.0" targetFramework="net45" />
+  <package id="Microsoft.Owin.Diagnostics" version="3.1.0" targetFramework="net45" />
+  <package id="Microsoft.Owin.Host.HttpListener" version="3.1.0" targetFramework="net45" />
+  <package id="Microsoft.Owin.Hosting" version="3.1.0" targetFramework="net45" />
+  <package id="Microsoft.Owin.SelfHost" version="3.1.0" targetFramework="net45" />
+  <package id="Newtonsoft.Json" version="10.0.2" targetFramework="net45" />
   <package id="Owin" version="1.0" targetFramework="net452" />
   <package id="TinyIoC" version="1.3" targetFramework="net452" developmentDependency="true" />
 </packages>
\ No newline at end of file
diff --git a/Tests/LiquidProjections.Specs/LiquidProjections.Specs.csproj b/Tests/LiquidProjections.Specs/LiquidProjections.Specs.csproj
index 4839cf3..f9b3bcc 100644
--- a/Tests/LiquidProjections.Specs/LiquidProjections.Specs.csproj
+++ b/Tests/LiquidProjections.Specs/LiquidProjections.Specs.csproj
@@ -34,17 +34,14 @@
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
   </PropertyGroup>
   <ItemGroup>
-    <Reference Include="Chill, Version=2.4.1.0, Culture=neutral, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\Chill.2.4.1\Lib\net45\Chill.dll</HintPath>
-      <Private>True</Private>
+    <Reference Include="Chill, Version=3.0.0.0, Culture=neutral, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\Chill.3.0.0\lib\net45\Chill.dll</HintPath>
     </Reference>
-    <Reference Include="FluentAssertions, Version=4.18.0.0, Culture=neutral, PublicKeyToken=33f2691a05b67b6a, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\FluentAssertions.4.18.0\lib\net45\FluentAssertions.dll</HintPath>
-      <Private>True</Private>
+    <Reference Include="FluentAssertions, Version=4.19.2.0, Culture=neutral, PublicKeyToken=33f2691a05b67b6a, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\FluentAssertions.4.19.2\lib\net45\FluentAssertions.dll</HintPath>
     </Reference>
-    <Reference Include="FluentAssertions.Core, Version=4.18.0.0, Culture=neutral, PublicKeyToken=33f2691a05b67b6a, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\FluentAssertions.4.18.0\lib\net45\FluentAssertions.Core.dll</HintPath>
-      <Private>True</Private>
+    <Reference Include="FluentAssertions.Core, Version=4.19.2.0, Culture=neutral, PublicKeyToken=33f2691a05b67b6a, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\FluentAssertions.4.19.2\lib\net45\FluentAssertions.Core.dll</HintPath>
     </Reference>
     <Reference Include="System" />
     <Reference Include="System.Core" />
@@ -55,20 +52,16 @@
     <Reference Include="System.Net.Http" />
     <Reference Include="System.Xml" />
     <Reference Include="xunit.abstractions, Version=2.0.0.0, Culture=neutral, PublicKeyToken=8d05b1bb7a6fdb6c, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\xunit.abstractions.2.0.0\lib\net35\xunit.abstractions.dll</HintPath>
-      <Private>True</Private>
+      <HintPath>..\..\packages\xunit.abstractions.2.0.1\lib\net35\xunit.abstractions.dll</HintPath>
     </Reference>
-    <Reference Include="xunit.assert, Version=2.1.0.3179, Culture=neutral, PublicKeyToken=8d05b1bb7a6fdb6c, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\xunit.assert.2.1.0\lib\dotnet\xunit.assert.dll</HintPath>
-      <Private>True</Private>
+    <Reference Include="xunit.assert, Version=2.2.0.3545, Culture=neutral, PublicKeyToken=8d05b1bb7a6fdb6c, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\xunit.assert.2.2.0\lib\netstandard1.1\xunit.assert.dll</HintPath>
     </Reference>
-    <Reference Include="xunit.core, Version=2.1.0.3179, Culture=neutral, PublicKeyToken=8d05b1bb7a6fdb6c, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\xunit.extensibility.core.2.1.0\lib\dotnet\xunit.core.dll</HintPath>
-      <Private>True</Private>
+    <Reference Include="xunit.core, Version=2.2.0.3545, Culture=neutral, PublicKeyToken=8d05b1bb7a6fdb6c, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\xunit.extensibility.core.2.2.0\lib\netstandard1.1\xunit.core.dll</HintPath>
     </Reference>
-    <Reference Include="xunit.execution.desktop, Version=2.1.0.3179, Culture=neutral, PublicKeyToken=8d05b1bb7a6fdb6c, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\xunit.extensibility.execution.2.1.0\lib\net45\xunit.execution.desktop.dll</HintPath>
-      <Private>True</Private>
+    <Reference Include="xunit.execution.desktop, Version=2.2.0.3545, Culture=neutral, PublicKeyToken=8d05b1bb7a6fdb6c, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\xunit.extensibility.execution.2.2.0\lib\net452\xunit.execution.desktop.dll</HintPath>
     </Reference>
   </ItemGroup>
   <ItemGroup>
@@ -77,9 +70,8 @@
     <Compile Include="ProjectionExceptionSpecs.cs" />
     <Compile Include="ProjectorSpecs.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
-  </ItemGroup>
-  <ItemGroup>
-    <None Include="packages.config" />
+    <Compile Include="TaskExtensions.cs" />
+    <Compile Include="TransactionBuilder.cs" />
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\..\Src\LiquidProjections.Abstractions\LiquidProjections.Abstractions.csproj">
@@ -98,6 +90,9 @@
   <ItemGroup>
     <Service Include="{82A7F48D-3B50-4B1E-B82E-3ADA8210C358}" />
   </ItemGroup>
+  <ItemGroup>
+    <None Include="packages.config" />
+  </ItemGroup>
   <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
   <!-- To modify your build process, add your task inside one of the targets below and uncomment it. 
        Other similar extension points exist, see Microsoft.Common.targets.
diff --git a/Tests/LiquidProjections.Specs/packages.config b/Tests/LiquidProjections.Specs/packages.config
index 8a8ba7d..99ec5eb 100644
--- a/Tests/LiquidProjections.Specs/packages.config
+++ b/Tests/LiquidProjections.Specs/packages.config
@@ -1,11 +1,11 @@
 <?xml version="1.0" encoding="utf-8"?>
 <packages>
-  <package id="Chill" version="2.4.1" targetFramework="net452" />
-  <package id="FluentAssertions" version="4.18.0" targetFramework="net452" />
-  <package id="xunit" version="2.1.0" targetFramework="net452" />
-  <package id="xunit.abstractions" version="2.0.0" targetFramework="net452" />
-  <package id="xunit.assert" version="2.1.0" targetFramework="net452" />
-  <package id="xunit.core" version="2.1.0" targetFramework="net452" />
-  <package id="xunit.extensibility.core" version="2.1.0" targetFramework="net452" />
-  <package id="xunit.extensibility.execution" version="2.1.0" targetFramework="net452" />
+  <package id="Chill" version="3.0.0" targetFramework="net452" />
+  <package id="FluentAssertions" version="4.19.2" targetFramework="net452" />
+  <package id="xunit" version="2.2.0" targetFramework="net452" />
+  <package id="xunit.abstractions" version="2.0.1" targetFramework="net452" />
+  <package id="xunit.assert" version="2.2.0" targetFramework="net452" />
+  <package id="xunit.core" version="2.2.0" targetFramework="net452" />
+  <package id="xunit.extensibility.core" version="2.2.0" targetFramework="net452" />
+  <package id="xunit.extensibility.execution" version="2.2.0" targetFramework="net452" />
 </packages>
\ No newline at end of file

From e575e9f668566c4897f4c6e229c5a0a721b7ad8d Mon Sep 17 00:00:00 2001
From: Dennis Doomen <dennis.doomen@avivasolutions.nl>
Date: Tue, 23 May 2017 18:04:41 +0200
Subject: [PATCH 2/3] Added NCrunch settings

---
 .../LiquidProjections.Abstractions.v3.ncrunchproject         | 5 +++++
 .../LiquidProjections.Testing.v3.ncrunchproject              | 5 +++++
 Src/LiquidProjections/LiquidProjections.v3.ncrunchproject    | 5 +++++
 3 files changed, 15 insertions(+)
 create mode 100644 Src/LiquidProjections.Abstractions/LiquidProjections.Abstractions.v3.ncrunchproject
 create mode 100644 Src/LiquidProjections.Testing/LiquidProjections.Testing.v3.ncrunchproject
 create mode 100644 Src/LiquidProjections/LiquidProjections.v3.ncrunchproject

diff --git a/Src/LiquidProjections.Abstractions/LiquidProjections.Abstractions.v3.ncrunchproject b/Src/LiquidProjections.Abstractions/LiquidProjections.Abstractions.v3.ncrunchproject
new file mode 100644
index 0000000..0e3a2cf
--- /dev/null
+++ b/Src/LiquidProjections.Abstractions/LiquidProjections.Abstractions.v3.ncrunchproject
@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <PreviouslyBuiltSuccessfully>True</PreviouslyBuiltSuccessfully>
+  </Settings>
+</ProjectConfiguration>
\ No newline at end of file
diff --git a/Src/LiquidProjections.Testing/LiquidProjections.Testing.v3.ncrunchproject b/Src/LiquidProjections.Testing/LiquidProjections.Testing.v3.ncrunchproject
new file mode 100644
index 0000000..0e3a2cf
--- /dev/null
+++ b/Src/LiquidProjections.Testing/LiquidProjections.Testing.v3.ncrunchproject
@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <PreviouslyBuiltSuccessfully>True</PreviouslyBuiltSuccessfully>
+  </Settings>
+</ProjectConfiguration>
\ No newline at end of file
diff --git a/Src/LiquidProjections/LiquidProjections.v3.ncrunchproject b/Src/LiquidProjections/LiquidProjections.v3.ncrunchproject
new file mode 100644
index 0000000..0e3a2cf
--- /dev/null
+++ b/Src/LiquidProjections/LiquidProjections.v3.ncrunchproject
@@ -0,0 +1,5 @@
+<ProjectConfiguration>
+  <Settings>
+    <PreviouslyBuiltSuccessfully>True</PreviouslyBuiltSuccessfully>
+  </Settings>
+</ProjectConfiguration>
\ No newline at end of file

From f06507d3dbe65515ad95dcf24d9e3112e0dba0f4 Mon Sep 17 00:00:00 2001
From: Dennis Doomen <dennis.doomen@avivasolutions.nl>
Date: Tue, 23 May 2017 18:05:26 +0200
Subject: [PATCH 3/3] Added the possibility to restart a subscription if the
 projector is ahead of the event source.

---
 Samples/ExampleHost/CountsProjector.cs        |   2 +-
 Samples/ExampleHost/JsonFileEventStore.cs     |  15 +-
 .../CreateSubscription.cs                     |  41 +++-
 .../MemoryEventSource.cs                      |  66 +++++--
 Src/LiquidProjections/Dispatcher.cs           | 101 ++++++----
 Src/LiquidProjections/EnumerableExtensions.cs |  16 ++
 .../DispatcherSpecs.cs                        | 177 ++++++++++++++++--
 .../LiquidProjections.Specs/ProjectorSpecs.cs |  11 +-
 .../LiquidProjections.Specs/TaskExtensions.cs |  25 +++
 .../TransactionBuilder.cs                     |  26 +++
 10 files changed, 395 insertions(+), 85 deletions(-)
 create mode 100644 Src/LiquidProjections/EnumerableExtensions.cs
 create mode 100644 Tests/LiquidProjections.Specs/TaskExtensions.cs
 create mode 100644 Tests/LiquidProjections.Specs/TransactionBuilder.cs

diff --git a/Samples/ExampleHost/CountsProjector.cs b/Samples/ExampleHost/CountsProjector.cs
index 1a49cb1..7da5c41 100644
--- a/Samples/ExampleHost/CountsProjector.cs
+++ b/Samples/ExampleHost/CountsProjector.cs
@@ -23,7 +23,7 @@ public CountsProjector(Dispatcher dispatcher, InMemoryDatabase store)
 
         public void Start()
         {
-            dispatcher.Subscribe(null, async transactions =>
+            dispatcher.Subscribe(null, async (transactions, info) =>
             {
                 await documentCountProjector.Handle(transactions);
             });
diff --git a/Samples/ExampleHost/JsonFileEventStore.cs b/Samples/ExampleHost/JsonFileEventStore.cs
index 068d3b1..8e9b4b9 100644
--- a/Samples/ExampleHost/JsonFileEventStore.cs
+++ b/Samples/ExampleHost/JsonFileEventStore.cs
@@ -5,6 +5,7 @@
 using System.Linq;
 using System.Runtime.Serialization.Formatters;
 using System.Threading.Tasks;
+using LiquidProjections.Abstractions;
 using Newtonsoft.Json;
 
 namespace LiquidProjections.ExampleHost
@@ -25,9 +26,11 @@ public JsonFileEventStore(string filePath, int pageSize)
             entryQueue = new Queue<ZipArchiveEntry>(zip.Entries.Where(e => e.Name.EndsWith(".json")));
         }
 
-        public IDisposable Subscribe(long? lastProcessedCheckpoint, Func<IReadOnlyList<Transaction>, Task> handler, string subscriptionId)
+        public IDisposable Subscribe(long? lastProcessedCheckpoint, Subscriber subscriber, string subscriptionId)
         {
-            var subscriber = new Subscriber(lastProcessedCheckpoint ?? 0, handler);
+            var subscription = new Subscription(
+                lastProcessedCheckpoint ?? 0, 
+                transactions => subscriber.HandleTransactions(transactions, null));
             
             Task.Run(async () =>
             {
@@ -39,13 +42,13 @@ public IDisposable Subscribe(long? lastProcessedCheckpoint, Func<IReadOnlyList<T
                     // Start loading the next page on a separate thread while we have the subscriber handle the previous transactions.
                     loader = LoadNextPageAsync();
 
-                    await subscriber.Send(transactions);
+                    await subscription.Send(transactions);
 
                     transactions = await loader;
                 }
             });
 
-            return subscriber;
+            return subscription;
         }
 
         private Task<Transaction[]> LoadNextPageAsync()
@@ -105,13 +108,13 @@ public void Dispose()
             zip = null;
         }
 
-        internal class Subscriber : IDisposable
+        internal class Subscription : IDisposable
         {
             private readonly long lastProcessedCheckpoint;
             private readonly Func<IReadOnlyList<Transaction>, Task> handler;
             private bool disposed;
 
-            public Subscriber(long lastProcessedCheckpoint, Func<IReadOnlyList<Transaction>, Task> handler)
+            public Subscription(long lastProcessedCheckpoint, Func<IReadOnlyList<Transaction>, Task> handler)
             {
                 this.lastProcessedCheckpoint = lastProcessedCheckpoint;
                 this.handler = handler;
diff --git a/Src/LiquidProjections.Abstractions/CreateSubscription.cs b/Src/LiquidProjections.Abstractions/CreateSubscription.cs
index b1bac50..a81ad31 100644
--- a/Src/LiquidProjections.Abstractions/CreateSubscription.cs
+++ b/Src/LiquidProjections.Abstractions/CreateSubscription.cs
@@ -7,13 +7,40 @@ namespace LiquidProjections.Abstractions
     /// <summary>
     /// Creates a subscription on an event store, starting at the transaction following 
     /// <paramref name="lastProcessedCheckpoint"/>, identified by <paramref name="subscriptionId"/>, and which 
-    /// passed transactions to the provided <paramref name="handler"/>.
+    /// passes any transactions to the provided <paramref name="subscriber"/>.
     /// </summary>
-    /// <param name="lastProcessedCheckpoint"></param>
-    /// <param name="handler"></param>
-    /// <param name="subscriptionId"></param>
-    /// <returns></returns>
+    /// <param name="lastProcessedCheckpoint"> 
+    /// The checkpoint of the transaction the subscriber has last seen, or <c>null</c> to start from the beginning.
+    /// </param>
+    /// <param name="subscriber">
+    /// An object wrapping the various handlers that the subscription will use.
+    /// </param>
+    /// <param name="subscriptionId">
+    /// Identifies this subscription and helps distinct multiple subscriptions. 
+    /// </param>
+    /// <returns>
+    /// A disposable object that will cancel the subscription.
+    /// </returns>
     public delegate IDisposable CreateSubscription(long? lastProcessedCheckpoint,
-        Func<IReadOnlyList<Transaction>, Task> handler,
-        string subscriptionId);
+        Subscriber subscriber, string subscriptionId);
+
+    public class Subscriber
+    {
+        /// <summary>
+        /// Represents a handler that will receive the transactions that the event store pushes to the subscriber.
+        /// </summary>
+        public Func<IReadOnlyList<Transaction>, SubscriptionInfo, Task> HandleTransactions { get; set; }
+
+        /// <summary>
+        /// Represents a handler that the event store will use if the requested checkpoint does not exist. 
+        /// </summary>
+        public Func<SubscriptionInfo, Task> NoSuchCheckpoint { get; set; }
+    }
+
+    public class SubscriptionInfo
+    {
+        public string Id { get; set; }
+
+        public IDisposable Subscription { get; set; }
+    }
 }
\ No newline at end of file
diff --git a/Src/LiquidProjections.Testing/MemoryEventSource.cs b/Src/LiquidProjections.Testing/MemoryEventSource.cs
index 38d0843..288aa6f 100644
--- a/Src/LiquidProjections.Testing/MemoryEventSource.cs
+++ b/Src/LiquidProjections.Testing/MemoryEventSource.cs
@@ -4,16 +4,16 @@
 using System.Globalization;
 using System.Linq;
 using System.Threading.Tasks;
-using LiquidProjections.Testing;
+using LiquidProjections.Abstractions;
 
-namespace LiquidProjections
+namespace LiquidProjections.Testing
 {
     public class MemoryEventSource
     {
         private readonly int batchSize;
         private long lastCheckpoint;
 
-        private readonly List<Subscriber> subscribers = new List<Subscriber>();
+        private readonly List<Subscription> subscribers = new List<Subscription>();
         private readonly List<Transaction> history = new List<Transaction>();
 
         public MemoryEventSource(int batchSize = 10)
@@ -21,24 +21,33 @@ public MemoryEventSource(int batchSize = 10)
             this.batchSize = batchSize;
         }
 
-        public IDisposable Subscribe(long? lastProcessedCheckpoint, Func<IReadOnlyList<Transaction>, Task> handler, string subscriptionId)
+        public IDisposable Subscribe(long? lastProcessedCheckpoint, Subscriber subscriber, string subscriptionId)
         {
             lastCheckpoint = lastProcessedCheckpoint ?? 0;
-            var subscriber = new Subscriber(lastCheckpoint, batchSize, handler);
+            var subscription = new Subscription(lastCheckpoint, batchSize, subscriber, subscriptionId);
 
-            subscribers.Add(subscriber);
+            subscribers.Add(subscription);
 
-            Func<Task> asyncAction = async () =>
+            async Task AsyncAction()
             {
+                if (history.LastOrDefault()?.Checkpoint < lastProcessedCheckpoint)
+                {
+                    await subscriber.NoSuchCheckpoint(new SubscriptionInfo
+                    {
+                        Id = subscriptionId,
+                        Subscription = subscription
+                    });
+                }
+
                 foreach (Transaction transaction in history)
                 {
-                    await subscriber.Send(new[] { transaction }).ConfigureAwait(false);
+                    await subscription.Send(new[] {transaction}).ConfigureAwait(false);
                 }
-            };
+            }
 
-            asyncAction().ConfigureAwait(false).GetAwaiter().GetResult();
+            AsyncAction().ConfigureAwait(false).GetAwaiter().GetResult();
 
-            return subscriber;
+            return subscription;
         }
 
 
@@ -102,29 +111,46 @@ public async Task<Transaction> WriteWithHeaders(object anEvent, IDictionary<stri
 
             return transaction;
         }
+
+        public bool HasSubscriptionForId(string subscriptionId)
+        {
+            Subscription subscription = subscribers.SingleOrDefault(s => s.Id == subscriptionId);
+            return (subscription != null) && !subscription.IsDisposed;
+        }
     }
 
-    internal class Subscriber : IDisposable
+    internal class Subscription : IDisposable
     {
         private readonly long lastProcessedCheckpoint;
         private readonly int batchSize;
-        private readonly Func<IReadOnlyList<Transaction>, Task> handler;
+        private readonly Subscriber subscriber;
+        private readonly string subscriptionId;
         private bool disposed = false;
-        
-        public Subscriber(long lastProcessedCheckpoint, int batchSize, Func<IReadOnlyList<Transaction>, Task> handler)
+
+        public Subscription(long lastProcessedCheckpoint, int batchSize,
+            Subscriber subscriber, string subscriptionId)
         {
             this.lastProcessedCheckpoint = lastProcessedCheckpoint;
             this.batchSize = batchSize;
-            this.handler = handler;
+            this.subscriber = subscriber;
+            this.subscriptionId = subscriptionId;
         }
 
         public async Task Send(IEnumerable<Transaction> transactions)
         {
             if (!disposed)
             {
-                foreach (var batch in transactions.Where(t => t.Checkpoint > lastProcessedCheckpoint).InBatchesOf(batchSize))
+                var subscriptionInfo = new SubscriptionInfo
                 {
-                    await handler(new ReadOnlyCollection<Transaction>(batch.ToList())).ConfigureAwait(false);
+                    Id = subscriptionId,
+                    Subscription = this
+                };
+
+                Transaction[] requestedTransactions = transactions.Where(t => t.Checkpoint > lastProcessedCheckpoint).ToArray();
+                foreach (var batch in requestedTransactions.InBatchesOf(batchSize))
+                {
+                    await subscriber.HandleTransactions(new ReadOnlyCollection<Transaction>(batch.ToList()), subscriptionInfo)
+                        .ConfigureAwait(false);
                 }
             }
         }
@@ -133,5 +159,9 @@ public void Dispose()
         {
             disposed = true;
         }
+
+        public bool IsDisposed => disposed;
+
+        public string Id => subscriptionId;
     }
 }
diff --git a/Src/LiquidProjections/Dispatcher.cs b/Src/LiquidProjections/Dispatcher.cs
index 8c4c8a5..dd35afe 100644
--- a/Src/LiquidProjections/Dispatcher.cs
+++ b/Src/LiquidProjections/Dispatcher.cs
@@ -10,62 +10,87 @@ public class Dispatcher
     {
         private readonly CreateSubscription createSubscription;
 
-        [Obsolete("The IEventStore interface has been replaced by the SubscribeToEvents delegate")]
-        public Dispatcher(IEventStore eventStore)
-        {
-            if (eventStore == null)
-            {
-                throw new ArgumentNullException(nameof(eventStore));
-            }
-
-            this.createSubscription = eventStore.Subscribe;
-        }
-
         public Dispatcher(CreateSubscription createSubscription)
         {
             this.createSubscription = createSubscription;
         }
 
-        public IDisposable Subscribe(long? checkpoint, Func<IReadOnlyList<Transaction>, Task> handler, string subscriptionId)
+        public IDisposable Subscribe(long? lastProcessedCheckpoint,
+            Func<IReadOnlyList<Transaction>, SubscriptionInfo, Task> handler,
+            SubscriptionOptions options = null)
         {
+            if (options == null)
+            {
+                options = new SubscriptionOptions();
+            }
+
             if (handler == null)
             {
                 throw new ArgumentNullException(nameof(handler));
             }
 
-            var subscriptionMonitor = new object();
-            IDisposable subscription = null;
-
-            lock (subscriptionMonitor)
+            return createSubscription(lastProcessedCheckpoint, new Subscriber
             {
-                subscription = createSubscription(checkpoint,
-                    async transactions =>
-                    {
-                        try
-                        {
-                            await handler(transactions);
-                        }
-                        catch (Exception exception)
-                        {
-                            LogProvider.GetLogger(typeof(Dispatcher)).FatalException(
-                                "Projector exception was not handled. Event subscription has been cancelled.",
-                                exception);
+                HandleTransactions = async (transactions, info) => await HandleTransactions(transactions, handler, info),
+                NoSuchCheckpoint = async info => await HandleUnknownCheckpoint(info, handler, options)
+            }, options.Id);
+        }
 
-                            lock (subscriptionMonitor)
-                            {
-                                subscription.Dispose();
-                            }
-                        }
-                    },
-                    subscriptionId);
+        private static async Task HandleTransactions(IReadOnlyList<Transaction> transactions, Func<IReadOnlyList<Transaction>, SubscriptionInfo, Task> handler, SubscriptionInfo info)
+        {
+            try
+            {
+                await handler(transactions, info);
             }
+            catch (Exception exception)
+            {
+                LogProvider.GetLogger(typeof(Dispatcher)).FatalException(
+                    "Projector exception was not handled. Event subscription has been cancelled.",
+                    exception);
 
-            return subscription;
+                info.Subscription?.Dispose();
+            }
         }
 
-        public IDisposable Subscribe(long? checkpoint, Func<IReadOnlyList<Transaction>, Task> handler)
+        private async Task HandleUnknownCheckpoint(SubscriptionInfo info, Func<IReadOnlyList<Transaction>, SubscriptionInfo, Task> handler, SubscriptionOptions options)
         {
-            return Subscribe(checkpoint, handler, null);
+            if (options.RestartWhenAhead)
+            {
+                try
+                {
+                    info.Subscription?.Dispose();
+
+                    await options.BeforeRestarting();
+
+                    Subscribe(null, handler, options);
+                }
+                catch (Exception exception)
+                {
+                    LogProvider.GetLogger(typeof(Dispatcher)).FatalException(
+                        "Failed to restart the projector.",
+                        exception);
+                }
+            }
         }
     }
+
+    public class SubscriptionOptions
+    {
+        /// <summary>
+        /// Can be used by subscribers to understand which is which.
+        /// </summary>
+        public string Id { get; set; }
+
+        /// <summary>
+        /// If set to <c>true</c>, the dispatcher will automatically restart at the first transaction if it detects
+        /// that the subscriber is ahead of the event store (e.g. because it got restored to an earlier time).
+        /// </summary>
+        public bool RestartWhenAhead { get; set; }
+
+        /// <summary>
+        /// If restarting is enabled through <see cref="RestartWhenAhead"/>, this property can be used to run some
+        /// clean-up code before the dispatcher will restart at the first transaction.
+        /// </summary>
+        public Func<Task> BeforeRestarting { get; set; } = () => Task.FromResult(0);
+    }
 }
\ No newline at end of file
diff --git a/Src/LiquidProjections/EnumerableExtensions.cs b/Src/LiquidProjections/EnumerableExtensions.cs
new file mode 100644
index 0000000..78adad2
--- /dev/null
+++ b/Src/LiquidProjections/EnumerableExtensions.cs
@@ -0,0 +1,16 @@
+using System.Collections.Generic;
+using System.Collections.ObjectModel;
+using System.Diagnostics;
+using System.Linq;
+
+namespace LiquidProjections
+{
+    [DebuggerNonUserCode]
+    internal static class EnumerableExtensions
+    {
+        public static IReadOnlyList<T> ToReadOnly<T>(this IEnumerable<T> items)
+        {
+            return new ReadOnlyCollection<T>(items.ToList());
+        }
+    }
+}
\ No newline at end of file
diff --git a/Tests/LiquidProjections.Specs/DispatcherSpecs.cs b/Tests/LiquidProjections.Specs/DispatcherSpecs.cs
index 3b5a781..c3bf4e7 100644
--- a/Tests/LiquidProjections.Specs/DispatcherSpecs.cs
+++ b/Tests/LiquidProjections.Specs/DispatcherSpecs.cs
@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
@@ -6,10 +7,11 @@
 using Chill;
 
 using FluentAssertions;
-
+using LiquidProjections.Abstractions;
 using LiquidProjections.Logging;
-
+using LiquidProjections.Testing;
 using Xunit;
+// ReSharper disable ConvertToLambdaExpression
 
 namespace LiquidProjections.Specs
 {
@@ -21,27 +23,34 @@ public When_a_projector_throws_an_exception()
             {
                 Given(() =>
                 {
-                    UseThe(new FakeEventStore());
-                    WithSubject(_ => new Dispatcher(The<FakeEventStore>().Subscribe));
+                    UseThe(new MemoryEventSource());
+                    WithSubject(_ => new Dispatcher(The<MemoryEventSource>().Subscribe));
+
                     LogProvider.SetCurrentLogProvider(UseThe(new FakeLogProvider()));
+
                     UseThe(new ProjectionException("Some message."));
 
-                    // Use async throw to easily get a faulted task.
-#pragma warning disable 1998
-                    Subject.Subscribe(null, async transaction =>
+                    Subject.Subscribe(null, (transaction, info) =>
                     {
+                        // Use async throw to easily get a faulted task.
                         throw The<ProjectionException>();
+                    }, 
+                    new SubscriptionOptions
+                    {
+                        Id = "mySubscription"
                     });
-#pragma warning restore 1998
                 });
 
-                When(() => The<FakeEventStore>().Handler(null));
+                When(() =>
+                {
+                    return The<MemoryEventSource>().Write(new List<Transaction>());
+                });
             }
 
             [Fact]
             public void It_should_unsubscribe()
             {
-                The<FakeEventStore>().IsSubscribed.Should().BeFalse();
+                The<MemoryEventSource>().HasSubscriptionForId("mySubscription").Should().BeFalse();
             }
 
             [Fact]
@@ -57,14 +66,158 @@ public void It_should_include_the_exception()
             }
         }
 
+        public class When_the_requested_checkpoint_is_ahead_of_the_store_and_auto_restart_is_configured : GivenSubject<Dispatcher>
+        {
+            private readonly BlockingCollection<string> trace = new BlockingCollection<string>();
+
+            private readonly BlockingCollection<Transaction> receivedTransactions = new BlockingCollection<Transaction>();
+
+            private readonly TaskCompletionSource<IEnumerable<Transaction>> allTransactionsReceived = 
+                new TaskCompletionSource<IEnumerable<Transaction>>();
+
+            public When_the_requested_checkpoint_is_ahead_of_the_store_and_auto_restart_is_configured()
+            {
+                Given(async () =>
+                {
+                    UseThe(new MemoryEventSource());
+
+                    WithSubject(_ => new Dispatcher(The<MemoryEventSource>().Subscribe));
+
+                    await The<MemoryEventSource>().Write(
+                        new TransactionBuilder().WithCheckpoint(1).Build());
+
+                    await The<MemoryEventSource>().Write(
+                        new TransactionBuilder().WithCheckpoint(999).Build());
+                });
+
+                When(() =>
+                {
+                    var options = new SubscriptionOptions
+                    {
+                        Id = "someId",
+                        RestartWhenAhead = true,
+                        BeforeRestarting = () =>
+                        {
+                            trace.Add("BeforeRestarting");
+                            return Task.FromResult(0);
+                        }
+                    };
+
+                    Subject.Subscribe(1000, (transactions, info) =>
+                    {
+                        trace.Add("TransactionsReceived");
+
+                        foreach (var transaction in transactions)
+                        {
+                            receivedTransactions.Add(transaction);
+                        }
+
+                        if (receivedTransactions.Count == 2)
+                        {
+                            allTransactionsReceived.SetResult(transactions);
+                        }
+
+                        return Task.FromResult(0);
+
+                    }, options);
+                });
+            }
+
+            [Fact]
+            public async Task It_should_allow_the_subscriber_to_cleanup_before_restarting()
+            {
+                await allTransactionsReceived.Task.TimeoutAfter(30.Seconds());
+
+                trace.Should().Equal("BeforeRestarting", "TransactionsReceived", "TransactionsReceived");
+            }
+
+            [Fact]
+            public async Task It_should_restart_sending_transactions_from_the_beginning()
+            {
+                var transactions = await allTransactionsReceived.Task.TimeoutAfter(30.Seconds());
+
+                transactions.First().Checkpoint.Should().Be(999);
+            }
+        }
+        public class When_there_are_no_new_transactions_available_and_auto_restart_is_configured : GivenSubject<Dispatcher>
+        {
+            private readonly BlockingCollection<Transaction> receivedTransactions = new BlockingCollection<Transaction>();
+
+            private readonly TaskCompletionSource<bool> allTransactionsReceived = 
+                new TaskCompletionSource<bool>();
+
+            private bool restarted = false;
+
+            public When_there_are_no_new_transactions_available_and_auto_restart_is_configured()
+            {
+                Given(async () =>
+                {
+                    UseThe(new MemoryEventSource());
+
+                    WithSubject(_ => new Dispatcher(The<MemoryEventSource>().Subscribe));
+
+                    await The<MemoryEventSource>().Write(
+                        new TransactionBuilder().WithCheckpoint(123).Build());
+
+                    await The<MemoryEventSource>().Write(
+                        new TransactionBuilder().WithCheckpoint(456).Build());
+                });
+
+                When(() =>
+                {
+                    var options = new SubscriptionOptions
+                    {
+                        Id = "someId",
+                        RestartWhenAhead = true,
+                        BeforeRestarting = () =>
+                        {
+                            restarted = true;
+                            return Task.FromResult(0);
+                        }
+                    };
+
+                    Subject.Subscribe(123, (transactions, info) =>
+                    {
+                        foreach (var transaction in transactions)
+                        {
+                            receivedTransactions.Add(transaction);
+                        }
+
+                        if (receivedTransactions.Count == 1)
+                        {
+                            allTransactionsReceived.SetResult(true);
+                        }
+
+                        return Task.FromResult(0);
+
+                    }, options);
+                });
+            }
+
+            [Fact]
+            public async Task It_should_only_provide_the_newer_transactions()
+            {
+                await allTransactionsReceived.Task.TimeoutAfter(30.Seconds());
+
+                receivedTransactions.Should().HaveCount(1);
+            }
+
+            public async Task It_should_not_have_restartedAsync()
+            {
+                await allTransactionsReceived.Task.TimeoutAfter(30.Seconds());
+
+                restarted.Should().BeFalse();
+            }
+        }
+
         internal class FakeEventStore
         {
             private readonly object syncRoot = new object();
 
             public bool IsSubscribed { get; private set; }
-            public Func<IReadOnlyList<Transaction>, Task> Handler { get; private set; }
+            public Func<IReadOnlyList<Transaction>, SubscriptionInfo, Task> Handler { get; private set; }
 
-            public IDisposable Subscribe(long? previousCheckpoint, Func<IReadOnlyList<Transaction>, Task> handler, string subscriptionId = null)
+            public IDisposable Subscribe(long? previousCheckpoint, Func<IReadOnlyList<Transaction>, SubscriptionInfo,Task> handler, string subscriptionId = null)
             {
                 lock (syncRoot)
                 {
diff --git a/Tests/LiquidProjections.Specs/ProjectorSpecs.cs b/Tests/LiquidProjections.Specs/ProjectorSpecs.cs
index 26f317a..d283a29 100644
--- a/Tests/LiquidProjections.Specs/ProjectorSpecs.cs
+++ b/Tests/LiquidProjections.Specs/ProjectorSpecs.cs
@@ -3,6 +3,8 @@
 using System.Threading.Tasks;
 using Chill;
 using FluentAssertions;
+using LiquidProjections.Abstractions;
+using LiquidProjections.Testing;
 using Xunit;
 
 namespace LiquidProjections.Specs
@@ -27,7 +29,10 @@ public Given_a_projector_with_an_in_memory_event_source()
 
             protected void StartProjecting()
             {
-                The<MemoryEventSource>().Subscribe(110, Subject.Handle, "");
+                The<MemoryEventSource>().Subscribe(110, new Subscriber
+                {
+                    HandleTransactions = async (transactions, info) => await Subject.Handle(transactions)
+                }, "");
             }
         }
 
@@ -115,7 +120,7 @@ public When_an_event_is_not_mapped_at_all()
                 When(async () =>
                 {
                     await The<MemoryEventSource>().Write(new ProductAddedToCatalogEvent());
-                }, deferedExecution: true);
+                }, deferredExecution: true);
             }
 
             [Fact]
@@ -230,7 +235,7 @@ public When_event_handling_fails()
                     StartProjecting();
                 });
 
-                When(() => The<MemoryEventSource>().Write(The<Transaction>()), deferedExecution: true);
+                When(() => The<MemoryEventSource>().Write(The<Transaction>()), deferredExecution: true);
             }
 
             [Fact]
diff --git a/Tests/LiquidProjections.Specs/TaskExtensions.cs b/Tests/LiquidProjections.Specs/TaskExtensions.cs
new file mode 100644
index 0000000..a759ccf
--- /dev/null
+++ b/Tests/LiquidProjections.Specs/TaskExtensions.cs
@@ -0,0 +1,25 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace LiquidProjections.Specs
+{
+    internal static class TaskExtensions
+    {
+        public static async Task<TResult> TimeoutAfter<TResult>(this Task<TResult> task, TimeSpan timeout)
+        {
+            var timeoutCancellationTokenSource = new CancellationTokenSource();
+
+            var completedTask = await Task.WhenAny(task, Task.Delay(timeout, timeoutCancellationTokenSource.Token));
+            if (completedTask == task)
+            {
+                timeoutCancellationTokenSource.Cancel();
+                return await task;  // Very important in order to propagate exceptions
+            }
+            else
+            {
+                throw new TimeoutException("The operation has timed out.");
+            }
+        }
+    }
+}
diff --git a/Tests/LiquidProjections.Specs/TransactionBuilder.cs b/Tests/LiquidProjections.Specs/TransactionBuilder.cs
new file mode 100644
index 0000000..5bd4228
--- /dev/null
+++ b/Tests/LiquidProjections.Specs/TransactionBuilder.cs
@@ -0,0 +1,26 @@
+using System;
+
+namespace LiquidProjections.Specs
+{
+    internal class TransactionBuilder
+    {
+        private long checkpoint;
+
+        public TransactionBuilder WithCheckpoint(long checkpoint)
+        {
+            this.checkpoint = checkpoint;
+            return this;
+        }
+
+        public Transaction Build()
+        {
+            return new Transaction
+            {
+                Checkpoint = checkpoint,
+                Id = Guid.NewGuid().ToString(),
+                StreamId = Guid.NewGuid().ToString(),
+                TimeStampUtc = DateTime.UtcNow
+            };
+        }
+    }
+}
\ No newline at end of file