From 89ed7639e49b0b19ca296e105718f7deeaebc201 Mon Sep 17 00:00:00 2001
From: Paul Betts <paul@paulbetts.org>
Date: Fri, 30 Dec 2011 14:53:43 -0500
Subject: [PATCH] Fix race condition where PersistentBlobCache wasn't
 guaranteeing objects were fully serialized / deserialized on ctor and
 Dispose()

---
 Akavache/IBlobCache.cs               | 5 +++--
 Akavache/PersistentBlobCache.cs      | 4 ++--
 Akavache/SimpleFilesystemProvider.cs | 5 +++--
 Akavache/Utility.cs                  | 7 ++++---
 4 files changed, 12 insertions(+), 9 deletions(-)

diff --git a/Akavache/IBlobCache.cs b/Akavache/IBlobCache.cs
index 80daaf22..60e5743f 100644
--- a/Akavache/IBlobCache.cs
+++ b/Akavache/IBlobCache.cs
@@ -21,8 +21,9 @@ public interface IFilesystemProvider
         /// <param name="mode">The file mode</param>
         /// <param name="access">The required access privileges</param>
         /// <param name="share">The allowed file sharing modes.</param>
+        /// <param name="scheduler">The scheduler to schedule the open under.</param>
         /// <returns>A Future result representing the Open file.</returns>
-        IObservable<Stream> SafeOpenFileAsync(string path, FileMode mode, FileAccess access, FileShare share);
+        IObservable<Stream> SafeOpenFileAsync(string path, FileMode mode, FileAccess access, FileShare share, IScheduler scheduler);
 
         /// <summary>
         /// Create a directory and its parents. If the directory already
@@ -82,7 +83,7 @@ public interface IBlobCache : IDisposable
         /// <summary>
         /// Invalidate all entries in the cache (i.e. clear it). Note that
         /// this method is blocking and incurs a significant performance
-        /// penalty if used while the cache is being used on other threads.
+        /// penalty if used while the cache is being used on other threads. 
         /// </summary>
         void InvalidateAll();
 
diff --git a/Akavache/PersistentBlobCache.cs b/Akavache/PersistentBlobCache.cs
index c690a1ac..6c758070 100644
--- a/Akavache/PersistentBlobCache.cs
+++ b/Akavache/PersistentBlobCache.cs
@@ -268,7 +268,7 @@ AsyncSubject<byte[]> FetchOrWriteBlobFromDisk(string key, object byteData, bool
             var ms = new MemoryStream();
 
             var scheduler = synchronous ? System.Reactive.Concurrency.Scheduler.Immediate : Scheduler;
-            filesystem.SafeOpenFileAsync(GetPathForKey(key), FileMode.Open, FileAccess.Read, FileShare.Read)
+            filesystem.SafeOpenFileAsync(GetPathForKey(key), FileMode.Open, FileAccess.Read, FileShare.Read, scheduler)
                 .SelectMany(x => x.CopyToAsync(ms, scheduler))
                 .SelectMany(x => AfterReadFromDiskFilter(ms.ToArray(), scheduler))
                 .Catch<byte[], FileNotFoundException>(ex => Observable.Throw<byte[]>(new KeyNotFoundException()))
@@ -285,7 +285,7 @@ AsyncSubject<byte[]> WriteBlobToDisk(string key, byte[] byteData, bool synchrono
 
             var files = Observable.Zip(
                 BeforeWriteToDiskFilter(byteData, scheduler).Select(x => new MemoryStream(x)),
-                filesystem.SafeOpenFileAsync(GetPathForKey(key), FileMode.Create, FileAccess.Write, FileShare.None),
+                filesystem.SafeOpenFileAsync(GetPathForKey(key), FileMode.Create, FileAccess.Write, FileShare.None, scheduler),
                 (from, to) => new { from, to }
             );
 
diff --git a/Akavache/SimpleFilesystemProvider.cs b/Akavache/SimpleFilesystemProvider.cs
index 928c298c..e02d6c89 100644
--- a/Akavache/SimpleFilesystemProvider.cs
+++ b/Akavache/SimpleFilesystemProvider.cs
@@ -1,13 +1,14 @@
 using System;
 using System.IO;
+using System.Reactive.Concurrency;
 
 namespace Akavache
 {
     public class SimpleFilesystemProvider : IFilesystemProvider
     {
-        public IObservable<Stream> SafeOpenFileAsync(string path, FileMode mode, FileAccess access, FileShare share)
+        public IObservable<Stream> SafeOpenFileAsync(string path, FileMode mode, FileAccess access, FileShare share, IScheduler scheduler)
         {
-            return Utility.SafeOpenFileAsync(path, mode, access, share);
+            return Utility.SafeOpenFileAsync(path, mode, access, share, scheduler);
         }
 
         public void CreateRecursive(string path)
diff --git a/Akavache/Utility.cs b/Akavache/Utility.cs
index e2055729..e2fe6abe 100644
--- a/Akavache/Utility.cs
+++ b/Akavache/Utility.cs
@@ -37,8 +37,9 @@ public static string GetMd5Hash(string input)
             }
         }
 
-        public static IObservable<FileStream> SafeOpenFileAsync(string path, FileMode mode, FileAccess access, FileShare share)
+        public static IObservable<FileStream> SafeOpenFileAsync(string path, FileMode mode, FileAccess access, FileShare share, IScheduler scheduler = null)
         {
+            scheduler = scheduler ?? RxApp.TaskpoolScheduler;
             return Observable.Create<FileStream>(subj =>
             {
                 try
@@ -56,10 +57,10 @@ public static IObservable<FileStream> SafeOpenFileAsync(string path, FileMode mo
                     } else
                     {
 #if SILVERLIGHT
-                        return Observable.Start(() => new FileStream(path, mode, access, share, 4096), RxApp.TaskpoolScheduler)
+                        return Observable.Start(() => new FileStream(path, mode, access, share, 4096), scheduler)
                             .Subscribe(subj);
 #else
-                        return Observable.Start(() => new FileStream(path, mode, access, share, 4096, true), RxApp.TaskpoolScheduler)
+                        return Observable.Start(() => new FileStream(path, mode, access, share, 4096, true), scheduler)
                             .Subscribe(subj);
 #endif
                     }