Ensured that the memory event store doesn't shallow projection except…

…ions

.gitignore

@@ -181,4 +181,5 @@ Build/**
 Tools/**
 
 # Visual Studio
-.vs/
+**/.vs/**
+.vs/**

Src/LiquidProjections.Testing/MemoryEventSource.cs

@@ -10,7 +10,7 @@ namespace LiquidProjections.Testing
     /// <summary>
     /// An event source which stores all the transactions in memory and has methods which are convenient for testing.
     /// </summary>
-    public partial class MemoryEventSource
+    public class MemoryEventSource
     {
         private readonly int batchSize;
         private readonly List<MemorySubscription> subscriptions = new List<MemorySubscription>();

Src/LiquidProjections.Testing/MemorySubscription.cs

@@ -18,6 +18,9 @@ public class MemorySubscription : IDisposable
         private CancellationTokenSource cancellationTokenSource;
         private readonly object syncRoot = new object();
         private Task task;
+
+        // Returns the last exception that caused the subscription to abort or <c>null</c>.
+        private Exception exception;
         private TaskCompletionSource<long> progressCompletionSource = new TaskCompletionSource<long>();
 
         private readonly TaskCompletionSource<bool> waitForCheckingWhetherItIsAheadCompletionSource =
@@ -63,8 +66,13 @@ public void Start()
                             {
                                 await RunAsync(info).ConfigureAwait(false);
                             }
-                            catch (Exception)
+                            catch (OperationCanceledException)
+                            {
+                                Dispose();
+                            }
+                            catch (Exception exc)
                             {
+                                exception = exc;
                                 Dispose();
                             }
                         },
@@ -208,7 +216,19 @@ public async Task WaitUntilCheckpoint(long checkpoint)
                     }
                 }
 
-                await progressTask.ConfigureAwait(false);
+                try
+                {
+                    await progressTask.ConfigureAwait(false);
+                }
+                catch (OperationCanceledException)
+                {
+                    break;
+                }
+            }
+
+            if (exception != null)
+            {
+                throw new AggregateException(exception);
             }
         }
 

Tests/LiquidProjections.Specs/DispatcherSpecs.cs

@@ -25,11 +25,11 @@ public When_a_projector_throws_an_exception()
             {
                 Given(() =>
                 {
+                    LogProvider.SetCurrentLogProvider(UseThe(new FakeLogProvider()));
+
                     UseThe(new MemoryEventSource());
                     WithSubject(_ => new Dispatcher(The<MemoryEventSource>().Subscribe));
 
-                    LogProvider.SetCurrentLogProvider(UseThe(new FakeLogProvider()));
-
                     UseThe(new ProjectionException("Some message."));
 
                     Subject.Subscribe(null, (transaction, info) =>

Tests/LiquidProjections.Specs/LiquidProjections.Specs.csproj

@@ -21,7 +21,7 @@
     <DefineConstants>DEBUG;TRACE</DefineConstants>
     <ErrorReport>prompt</ErrorReport>
     <WarningLevel>4</WarningLevel>
-    <LangVersion>6</LangVersion>
+    <LangVersion>7</LangVersion>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
   </PropertyGroup>
   <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
@@ -31,7 +31,7 @@
     <DefineConstants>TRACE</DefineConstants>
     <ErrorReport>prompt</ErrorReport>
     <WarningLevel>4</WarningLevel>
-    <LangVersion>6</LangVersion>
+    <LangVersion>7</LangVersion>
     <TreatWarningsAsErrors>true</TreatWarningsAsErrors>
   </PropertyGroup>
   <ItemGroup>
@@ -83,6 +83,7 @@
   <ItemGroup>
     <Compile Include="DispatcherSpecs.cs" />
     <Compile Include="EventMapSpecs.cs" />
+    <Compile Include="MemoryEventSourceSpecs.cs" />
     <Compile Include="ProjectionExceptionSpecs.cs" />
     <Compile Include="ProjectionStatsSpecs.cs" />
     <Compile Include="ProjectorSpecs.cs" />

Tests/LiquidProjections.Specs/MemoryEventSourceSpecs.cs

@@ -0,0 +1,47 @@
+using System;
+using System.Collections.Generic;
+using Chill;
+using FluentAssertions;
+using LiquidProjections.Testing;
+using Xunit;
+
+namespace LiquidProjections.Specs
+{
+    namespace MemoryEventSourceSpecs
+    {
+        public class When_a_subscriber_throws_an_exception_that_the_dispatcher_rethrows : GivenSubject<Dispatcher>
+        {
+            public When_a_subscriber_throws_an_exception_that_the_dispatcher_rethrows()
+            {
+                Given(() =>
+                {
+                    UseThe(new MemoryEventSource());
+                    WithSubject(_ => new Dispatcher(The<MemoryEventSource>().Subscribe)
+                    {
+                        ExceptionHandler = (exception, attempts, info) => throw exception
+                    });
+
+                    Subject.Subscribe(null, (transaction, info) =>
+                    {
+                        throw new ArgumentException();
+                    });
+                });
+
+                WhenLater(() =>
+                {
+                    return The<MemoryEventSource>().Write(new List<Transaction>
+                    {
+                        new Transaction()
+                    });
+                });
+            }
+
+            [Fact]
+            public void Then_the_exception_should_bubble_up_through_the_memory_event_source()
+            {
+                WhenAction.Should().Throw<ArgumentException>();
+            }
+        }
+
+    }
+}