Upgraded to LiquidProjections.PollingEventStore.Sources 1.1.0 (#8)

liquidprojections · Sep 14, 2017 · acb4730 · acb4730
1 parent bf33fbd
commit acb4730
Show file tree

Hide file tree

Showing 7 changed files with 74 additions and 46 deletions.
diff --git a/Src/LiquidProjections.NEventStore/.nuspec b/Src/LiquidProjections.NEventStore/.nuspec
@@ -13,9 +13,9 @@
         <releaseNotes/>
       <dependencies>
         <group>
-			<dependency id="NEventStore" version="[5.0, 6.0)"/>
-			<dependency id="LiquidProjections.Abstractions" version="[2.0, 3.0)"/>
-		</group>
+          <dependency id="NEventStore" version="[5.0, 6.0)"/>
+          <dependency id="LiquidProjections.Abstractions" version="[2.3, 3.0)"/>
+        </group>
       </dependencies>
     </metadata>
     <files>

diff --git a/....NEventStore/App_Packages/LiquidProjections.PollingEventStore/PollingEventStoreAdapter.cs b/....NEventStore/App_Packages/LiquidProjections.PollingEventStore/PollingEventStoreAdapter.cs
@@ -32,7 +32,6 @@ public class PollingEventStoreAdapter : IDisposable
         /// </summary>
         private readonly LruCache<long, Transaction> transactionCacheByPreviousCheckpoint;
 
-        private readonly CancellationTokenSource cancellationTokenSource = new CancellationTokenSource();
         private CheckpointRequestTimestamp lastSuccessfulPollingRequestWithoutResults;
 
         /// <summary>
@@ -173,11 +172,14 @@ private async Task<Page> LoadNextPageSequentially(long previousCheckpoint, strin
         {
             while (true)
             {
-                cancellationTokenSource.Token.ThrowIfCancellationRequested();
-
                 if (isDisposed)
                 {
-                    return new Page(previousCheckpoint, new Transaction[0]);
+#if DEBUG
+                    LogProvider.GetLogger(typeof(PollingEventStoreAdapter)).Debug(() =>
+                        $"Page loading for subscription {subscriptionId} cancelled because the adapter is disposed.");
+#endif
+
+                    throw new OperationCanceledException();
                 }
 
                 CheckpointRequestTimestamp effectiveLastExistingCheckpointRequest =
@@ -246,8 +248,22 @@ private Task<Page> TryLoadNextPageSequentiallyOrWaitForCurrentLoadingToFinish(lo
                                      $"for a page after checkpoint {previousCheckpoint}.");
 #endif
 
-                    // Ignore result.
-                    Task _ = TryLoadNextPageAndMakeLoaderComplete(previousCheckpoint, taskCompletionSource, subscriptionId);
+                    if (isDisposed)
+                    {
+#if DEBUG
+                        LogProvider.GetLogger(typeof(PollingEventStoreAdapter))
+                            .Debug(() => $"The loader {loader.Id} is cancelled because the adapter is disposed.");
+#endif
+
+                        // If the adapter is disposed before the current task is set, we cancel the task
+                        // so we do not touch the event store. 
+                        taskCompletionSource.SetCanceled();
+                    }
+                    else
+                    {
+                        // Ignore result.
+                        Task _ = TryLoadNextPageAndMakeLoaderComplete(previousCheckpoint, taskCompletionSource, subscriptionId);
+                    }
                 }
                 else
                 {
@@ -394,17 +410,23 @@ public void Dispose()
                 {
                     isDisposed = true;
 
-                    cancellationTokenSource.Cancel();
-
                     foreach (Subscription subscription in subscriptions.ToArray())
                     {
                         subscription.Complete();
                     }
 
+                    // New loading tasks are no longer started at this point.
+                    // After the current loading task is finished, the event store is no longer used and can be disposed.
                     Task loaderToWaitFor = Volatile.Read(ref currentLoader);
-                    loaderToWaitFor?.Wait();
 
-                    cancellationTokenSource.Dispose();
+                    try
+                    {
+                        loaderToWaitFor?.Wait();
+                    }
+                    catch (AggregateException)
+                    {
+                        // Ignore.
+                    }
 
                     (eventStore as IDisposable)?.Dispose();
                 }

diff --git a/...dProjections.NEventStore/App_Packages/LiquidProjections.PollingEventStore/Subscription.cs b/...dProjections.NEventStore/App_Packages/LiquidProjections.PollingEventStore/Subscription.cs
@@ -48,10 +48,11 @@ public void Start()
                 LogProvider.GetLogger(typeof(Subscription)).Debug(() => $"Subscription {Id} has been started.");
 #endif
 
-                SubscriptionInfo info = new SubscriptionInfo
+                var info = new SubscriptionInfo
                 {
                     Id = Id,
-                    Subscription = this
+                    Subscription = this,
+                    CancellationToken = cancellationTokenSource.Token
                 };
 
                 Task = Task.Factory.StartNew(async () =>
@@ -162,43 +163,48 @@ public void Complete()
 
         public void Dispose()
         {
-            bool isDisposing;
-
             lock (syncRoot)
             {
-                isDisposing = !isDisposed;
-
-                if (isDisposing)
+                if (!isDisposed)
                 {
                     isDisposed = true;
-                }
-            }
 
-            if (isDisposing)
-            {
-                if (cancellationTokenSource != null)
-                {
+                    // Wait for the task asynchronously.
+                    Task.Run(() =>
+                    {
+                        if (cancellationTokenSource != null)
+                        {
 #if DEBUG
-                    LogProvider.GetLogger(typeof(Subscription)).Debug(() => $"Subscription {Id} is being stopped.");
+                            LogProvider.GetLogger(typeof(Subscription)).Debug(() => $"Subscription {Id} is being stopped.");
 #endif
 
-                    if (!cancellationTokenSource.IsCancellationRequested)
-                    {
-                        cancellationTokenSource.Cancel();
-                    }
+                            if (!cancellationTokenSource.IsCancellationRequested)
+                            {
+                                cancellationTokenSource.Cancel();
+                            }
 
-                    Task?.Wait();
-                    cancellationTokenSource.Dispose();
-                }
+                            try
+                            {
+                                Task?.Wait();
+                            }
+                            catch (AggregateException)
+                            {
+                                // Ignore.
+                            }
 
-                lock (eventStoreAdapter.subscriptionLock)
-                {
-                    eventStoreAdapter.subscriptions.Remove(this);
-                }
+                            cancellationTokenSource.Dispose();
+                        }
+
+                        lock (eventStoreAdapter.subscriptionLock)
+                        {
+                            eventStoreAdapter.subscriptions.Remove(this);
+                        }
 
 #if DEBUG
-                LogProvider.GetLogger(typeof(Subscription)).Debug(() => $"Subscription {Id} has been stopped.");
+                        LogProvider.GetLogger(typeof(Subscription)).Debug(() => $"Subscription {Id} has been stopped.");
 #endif
+                    });
+                }
             }
         }
     }

diff --git a/Src/LiquidProjections.NEventStore/LiquidProjections.NEventStore.csproj b/Src/LiquidProjections.NEventStore/LiquidProjections.NEventStore.csproj
@@ -39,8 +39,8 @@
     <LangVersion>6</LangVersion>
   </PropertyGroup>
   <ItemGroup>
-    <Reference Include="LiquidProjections.Abstractions, Version=2.2.0.0, Culture=neutral, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\LiquidProjections.Abstractions.2.2.0\lib\netstandard1.1\LiquidProjections.Abstractions.dll</HintPath>
+    <Reference Include="LiquidProjections.Abstractions, Version=2.3.0.0, Culture=neutral, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\LiquidProjections.Abstractions.2.3.0\lib\netstandard1.1\LiquidProjections.Abstractions.dll</HintPath>
     </Reference>
     <Reference Include="NEventStore, Version=5.2.0.0, Culture=neutral, processorArchitecture=MSIL">
       <HintPath>..\..\packages\NEventStore.5.2.0\lib\net40\NEventStore.dll</HintPath>

diff --git a/Src/LiquidProjections.NEventStore/packages.config b/Src/LiquidProjections.NEventStore/packages.config
@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="utf-8"?>
 <packages>
-  <package id="LiquidProjections.Abstractions" version="2.2.0" targetFramework="net45" />
-  <package id="LiquidProjections.PollingEventStore.Sources" version="1.0.1" targetFramework="net45" />
+  <package id="LiquidProjections.Abstractions" version="2.3.0" targetFramework="net45" />
+  <package id="LiquidProjections.PollingEventStore.Sources" version="1.1.0" targetFramework="net45" />
   <package id="Microsoft.CSharp" version="4.3.0" targetFramework="net45" />
   <package id="NEventStore" version="5.2.0" targetFramework="net45" />
   <package id="System.Dynamic.Runtime" version="4.3.0" targetFramework="net45" />

diff --git a/Tests/LiquidProjections.NEventStore.Specs/LiquidProjections.NEventStore.Specs.csproj b/Tests/LiquidProjections.NEventStore.Specs/LiquidProjections.NEventStore.Specs.csproj
@@ -46,8 +46,8 @@
     <Reference Include="FluentAssertions.Core, Version=4.19.3.0, Culture=neutral, PublicKeyToken=33f2691a05b67b6a, processorArchitecture=MSIL">
       <HintPath>..\..\packages\FluentAssertions.4.19.3\lib\net45\FluentAssertions.Core.dll</HintPath>
     </Reference>
-    <Reference Include="LiquidProjections.Abstractions, Version=2.2.0.0, Culture=neutral, processorArchitecture=MSIL">
-      <HintPath>..\..\packages\LiquidProjections.Abstractions.2.2.0\lib\netstandard1.1\LiquidProjections.Abstractions.dll</HintPath>
+    <Reference Include="LiquidProjections.Abstractions, Version=2.3.0.0, Culture=neutral, processorArchitecture=MSIL">
+      <HintPath>..\..\packages\LiquidProjections.Abstractions.2.3.0\lib\netstandard1.1\LiquidProjections.Abstractions.dll</HintPath>
     </Reference>
     <Reference Include="NEventStore, Version=5.2.0.0, Culture=neutral, processorArchitecture=MSIL">
       <HintPath>..\..\packages\NEventStore.5.2.0\lib\net40\NEventStore.dll</HintPath>

diff --git a/Tests/LiquidProjections.NEventStore.Specs/packages.config b/Tests/LiquidProjections.NEventStore.Specs/packages.config
@@ -4,7 +4,7 @@
   <package id="Chill" version="3.0.1" targetFramework="net452" />
   <package id="FakeItEasy" version="3.4.2" targetFramework="net452" />
   <package id="FluentAssertions" version="4.19.3" targetFramework="net452" />
-  <package id="LiquidProjections.Abstractions" version="2.2.0" targetFramework="net452" />
+  <package id="LiquidProjections.Abstractions" version="2.3.0" targetFramework="net452" />
   <package id="NEventStore" version="5.2.0" targetFramework="net452" />
   <package id="xunit" version="2.2.0" targetFramework="net452" />
   <package id="xunit.abstractions" version="2.0.1" targetFramework="net452" />