Merge pull request #118 from github/feature/pr/streaming-cache

Pull request list apis with cache

Author: haacked

src/GitHub.App/Api/ApiClient.cs

@@ -7,7 +7,6 @@
 using System.Reactive.Linq;
 using System.Security.Cryptography;
 using System.Text;
-using GitHub.Authentication;
 using GitHub.Primitives;
 using NLog;
 using NullGuard;
@@ -122,8 +121,6 @@ public IObservable<LicenseMetadata> GetLicenses()
 
         public HostAddress HostAddress { get; }
 
-        public ITwoFactorChallengeHandler TwoFactorChallengeHandler { get; private set; }
-
         static string GetSha256Hash(string input)
         {
             try
@@ -198,5 +195,15 @@ public IObservable<Unit> DeleteApplicationAuthorization(int id, [AllowNull]strin
         {
             return gitHubClient.Authorization.Delete(id, twoFactorAuthorizationCode);
         }
+
+        public IObservable<PullRequest> GetPullRequestsForRepository(string owner, string name)
+        {
+            return gitHubClient.PullRequest.GetAllForRepository(owner, name,
+                new PullRequestRequest {
+                    State = ItemState.All,
+                    SortProperty = PullRequestSort.Updated,
+                    SortDirection = SortDirection.Descending
+                });
+        }
     }
 }

src/GitHub.App/Caches/CacheIndex.cs

@@ -0,0 +1,67 @@
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.Linq;
+using System.Reactive.Linq;
+using Akavache;
+using NullGuard;
+
+namespace GitHub.Caches
+{
+    public class CacheIndex
+    {
+        public static CacheIndex Create(string key)
+        {
+            return new CacheIndex { IndexKey = key };
+        }
+
+        public CacheIndex()
+        {
+            Keys = new List<string>();
+            OldKeys = new List<string>();
+        }
+
+        public IObservable<CacheIndex> AddAndSave(IBlobCache cache, string indexKey, CacheItem item,
+            DateTimeOffset? absoluteExpiration = null)
+        {
+            var k = string.Format(CultureInfo.InvariantCulture, "{0}|{1}", IndexKey, item.Key);
+            if (!Keys.Contains(k))
+                Keys.Add(k);
+            UpdatedAt = DateTimeOffset.UtcNow;
+            return cache.InsertObject(IndexKey, this, absoluteExpiration)
+                        .Select(x => this);
+        }
+
+        public static IObservable<CacheIndex> AddAndSaveToIndex(IBlobCache cache, string indexKey, CacheItem item,
+            DateTimeOffset? absoluteExpiration = null)
+        {
+            return cache.GetOrCreateObject(indexKey, () => Create(indexKey))
+                .Do(index =>
+                {
+                    var k = string.Format(CultureInfo.InvariantCulture, "{0}|{1}", index.IndexKey, item.Key);
+                    if (!index.Keys.Contains(k))
+                        index.Keys.Add(k);
+                    index.UpdatedAt = DateTimeOffset.UtcNow;
+                })
+                .SelectMany(index => cache.InsertObject(index.IndexKey, index, absoluteExpiration)
+                .Select(x => index));
+        }
+
+        public IObservable<CacheIndex> Clear(IBlobCache cache, string indexKey, DateTimeOffset? absoluteExpiration = null)
+        {
+            OldKeys = Keys.ToList();
+            Keys.Clear();
+            UpdatedAt = DateTimeOffset.UtcNow;
+            return cache
+                .InvalidateObject<CacheIndex>(indexKey)
+                .SelectMany(_ => cache.InsertObject(indexKey, this, absoluteExpiration))
+                .Select(_ => this);
+        }
+
+        [AllowNull]
+        public string IndexKey {[return: AllowNull] get; set; }
+        public List<string> Keys { get; set; }
+        public DateTimeOffset UpdatedAt { get; set; }
+        public List<string> OldKeys { get; set; }
+    }
+}

src/GitHub.App/Caches/CacheItem.cs

@@ -0,0 +1,25 @@
+using System;
+using System.Globalization;
+using System.Reactive.Linq;
+using Akavache;
+using NullGuard;
+
+namespace GitHub.Caches
+{
+    public class CacheItem
+    {
+        [AllowNull]
+        public string Key {[return: AllowNull] get; set; }
+        public DateTimeOffset Timestamp { get; set; }
+
+        public IObservable<T> Save<T>(IBlobCache cache, string key, DateTimeOffset? absoluteExpiration = null)
+            where T : CacheItem
+        {
+            var k = string.Format(CultureInfo.InvariantCulture, "{0}|{1}", key, Key);
+            return cache
+                .InvalidateObject<T>(k)
+                .Select(_ => cache.InsertObject(k, this, absoluteExpiration))
+                .Select(_ => this as T);
+        }
+    }
+}

src/GitHub.App/Extensions/AkavacheExtensions.cs

@@ -1,6 +1,9 @@
 using System;
+using System.Linq;
+using System.Reactive;
 using System.Reactive.Linq;
 using Akavache;
+using GitHub.Caches;
 
 namespace GitHub.Extensions
 {
@@ -14,6 +17,7 @@ public static class AkavacheExtensions
         /// the stale value will be produced first, followed by the fresh value
         /// when the fetch func completes.
         /// </summary>
+        /// <param name="blobCache">The cache to retrieve the object from.</param>
         /// <param name="key">The key to look up the cache value with.</param>
         /// <param name="fetchFunc">The fetch function.</param>
         /// <param name="refreshInterval">
@@ -59,6 +63,7 @@ public static IObservable<T> GetAndRefreshObject<T>(
         /// the stale value will be produced first, followed by the fresh value
         /// when the fetch func completes.
         /// </summary>
+        /// <param name="blobCache">The cache to retrieve the object from.</param>
         /// <param name="key">The key to look up the cache value with.</param>
         /// <param name="fetchFunc">The fetch function.</param>
         /// <param name="refreshInterval">
@@ -133,10 +138,105 @@ static IObservable<byte[]> GetAndFetchLatestBytes(
                 .RefCount();
         }
 
+        /// <summary>
+        /// This method attempts to return a series of cached value(s) aggregated by
+        /// a <paramref name="key"/>. In the case of a
+        /// cache miss the fetchFunc will be used to provide a fresh value which
+        /// is then returned. In the case of a cache hit where the cache item is
+        /// considered stale (but not expired) as determined by <paramref name="refreshInterval"/>
+        /// the stale values will be produced first, followed by the fresh values
+        /// when the fetch func completes.
+        /// </summary>
+        /// <param name="blobCache">The cache to retrieve the object from.</param>
+        /// <param name="key">The key to look up the cache value with.</param>
+        /// <param name="fetchFunc">The fetch function.</param>
+        /// <param name="removedItemsCallback">The callback that receives items that
+        /// were a part of the cache but not of the list list of values.</param>
+        /// <param name="refreshInterval">
+        /// Cache objects with an age exceeding this value will be treated as stale
+        /// and the fetch function will be invoked to refresh it.
+        /// </param>
+        /// <param name="maxCacheDuration">
+        /// The maximum age of a cache object before the object is treated as
+        /// expired and unusable. Cache objects older than this will be treated
+        /// as a cache miss.
+        /// </param>
+        public static IObservable<T> GetAndFetchLatestFromIndex<T>(
+            this IBlobCache blobCache,
+            string key,
+            Func<IObservable<T>> fetchFunc,
+            Action<T> removedItemsCallback,
+            TimeSpan refreshInterval,
+            TimeSpan maxCacheDuration)
+                where T : CacheItem
+        {
+            return Observable.Defer(() =>
+            {
+                var absoluteExpiration = blobCache.Scheduler.Now + maxCacheDuration;
+
+                try
+                {
+                    return blobCache.GetAndFetchLatestFromIndex(
+                        key,
+                        fetchFunc,
+                        removedItemsCallback,
+                        createdAt => IsExpired(blobCache, createdAt, refreshInterval),
+                        absoluteExpiration);
+                }
+                catch (Exception exc)
+                {
+                    return Observable.Throw<T>(exc);
+                }
+            });
+        }
+
+        static IObservable<T> GetAndFetchLatestFromIndex<T>(this IBlobCache This,
+            string key,
+            Func<IObservable<T>> fetchFunc,
+            Action<T> removedItemsCallback,
+            Func<DateTimeOffset, bool> fetchPredicate = null,
+            DateTimeOffset? absoluteExpiration = null,
+            bool shouldInvalidateOnError = false)
+                where T : CacheItem
+        {
+            var fetch = Observable.Defer(() => This.GetOrCreateObject(key, () => CacheIndex.Create(key))
+                .Select(x => Tuple.Create(x, fetchPredicate == null || !x.Keys.Any() || fetchPredicate(x.UpdatedAt)))
+                .Where(predicateIsTrue => predicateIsTrue.Item2)
+                .Select(x => x.Item1)
+                .SelectMany(index => index.Clear(This, key, absoluteExpiration))
+                .SelectMany(index =>
+                {
+                    var fetchObs = fetchFunc().Catch<T, Exception>(ex =>
+                    {
+                        var shouldInvalidate = shouldInvalidateOnError ?
+                            This.InvalidateObject<CacheIndex>(key) :
+                            Observable.Return(Unit.Default);
+                        return shouldInvalidate.SelectMany(__ => Observable.Throw<T>(ex));
+                    });
+
+                    return fetchObs
+                        .SelectMany(x => x.Save<T>(This, key, absoluteExpiration))
+                        .Do(x => index.AddAndSave(This, key, x, absoluteExpiration))
+                        .Finally(() =>
+                        {
+                            This.GetObjects<T>(index.OldKeys.Except(index.Keys))
+                                .Do(dict => This.InvalidateObjects<T>(dict.Keys))
+                                .SelectMany(dict => dict.Values)
+                                .Do(removedItemsCallback)
+                                .Subscribe();
+                        });
+                }));
+
+            var cache = Observable.Defer(() => This.GetOrCreateObject(key, () => CacheIndex.Create(key))
+                .SelectMany(index => This.GetObjects<T>(index.Keys))
+                .SelectMany(dict => dict.Values));
+
+            return cache.Merge(fetch).Replay().RefCount();
+        }
+
         static bool IsExpired(IBlobCache blobCache, DateTimeOffset itemCreatedAt, TimeSpan cacheDuration)
         {
-            var now = blobCache.Scheduler.Now;
-            var elapsed = now - itemCreatedAt;
+            var elapsed = blobCache.Scheduler.Now - itemCreatedAt.ToUniversalTime();
 
             return elapsed > cacheDuration;
         }

src/GitHub.App/GitHub.App.csproj

@@ -123,12 +123,15 @@
     <None Include="..\..\script\Key.snk" Condition="$(Buildtype) == 'Internal'">
       <Link>Key.snk</Link>
     </None>
+    <Compile Include="Caches\CacheIndex.cs" />
+    <Compile Include="Caches\CacheItem.cs" />
     <Compile Include="Caches\ImageCache.cs" />
     <Compile Include="Extensions\AkavacheExtensions.cs" />
     <Compile Include="Extensions\EnvironmentExtensions.cs" />
     <Compile Include="Extensions\ValidationExtensions.cs" />
     <Compile Include="GlobalSuppressions.cs" />
     <Compile Include="Infrastructure\LoggingConfiguration.cs" />
+    <Compile Include="Models\PullRequestModel.cs" />
     <Compile Include="Resources.Designer.cs">
       <AutoGen>True</AutoGen>
       <DesignTime>True</DesignTime>

src/GitHub.App/Models/PullRequestModel.cs

@@ -0,0 +1,113 @@
+using System;
+using System.Globalization;
+using GitHub.Primitives;
+using GitHub.VisualStudio.Helpers;
+using NullGuard;
+
+namespace GitHub.Models
+{
+    public sealed class PullRequestModel : NotificationAwareObject, IPullRequestModel
+    {
+        public PullRequestModel(int number, string title, IAccount author, DateTimeOffset createdAt, DateTimeOffset? updatedAt = null)
+        {
+            Number = number;
+            Title = title;
+            Author = author;
+            CreatedAt = createdAt;
+            UpdatedAt = updatedAt ?? CreatedAt;
+        }
+
+        public void CopyFrom(IPullRequestModel other)
+        {
+            if (!Equals(other))
+                throw new ArgumentException("Instance to copy from doesn't match this instance. this:(" + this + ") other:(" + other + ")", nameof(other));
+            Title = other.Title;
+            UpdatedAt = other.UpdatedAt;
+            CommentCount = other.CommentCount;
+            HasNewComments = other.HasNewComments;
+        }
+
+        public override bool Equals([AllowNull]object obj)
+        {
+            if (ReferenceEquals(this, obj))
+                return true;
+            var other = obj as PullRequestModel;
+            return other != null && Number == other.Number;
+        }
+
+        public override int GetHashCode()
+        {
+            return Number;
+        }
+
+        bool IEquatable<IPullRequestModel>.Equals([AllowNull]IPullRequestModel other)
+        {
+            if (ReferenceEquals(this, other))
+                return true;
+            return other != null && Number == other.Number;
+        }
+
+        public int CompareTo([AllowNull]IPullRequestModel other)
+        {
+            return other != null ? UpdatedAt.CompareTo(other.UpdatedAt) : 1;
+        }
+
+        public static bool operator >([AllowNull]PullRequestModel lhs, [AllowNull]PullRequestModel rhs)
+        {
+            if (ReferenceEquals(lhs, rhs))
+                return false;
+            return lhs?.CompareTo(rhs) > 0;
+        }
+
+        public static bool operator <([AllowNull]PullRequestModel lhs, [AllowNull]PullRequestModel rhs)
+        {
+            if (ReferenceEquals(lhs, rhs))
+                return false;
+            return (object)lhs == null || lhs.CompareTo(rhs) < 0;
+        }
+
+        public static bool operator ==([AllowNull]PullRequestModel lhs, [AllowNull]PullRequestModel rhs)
+        {
+            return Equals(lhs, rhs) && ((object)lhs == null || lhs.CompareTo(rhs) == 0);
+        }
+
+        public static bool operator !=([AllowNull]PullRequestModel lhs, [AllowNull]PullRequestModel rhs)
+        {
+            return !(lhs == rhs);
+        }
+
+        public int Number { get; }
+
+        string title;
+        public string Title
+        {
+            get { return title; }
+            set { title = value; this.RaisePropertyChange(); }
+        }
+
+        int commentCount;
+        public int CommentCount
+        {
+            get { return commentCount; }
+            set { commentCount = value; this.RaisePropertyChange(); }
+        }
+
+        bool hasNewComments;
+        public bool HasNewComments
+        {
+            get { return hasNewComments; }
+            set { hasNewComments = value; this.RaisePropertyChange(); }
+        }
+
+        public DateTimeOffset CreatedAt { get; set; }
+        public DateTimeOffset UpdatedAt { get; set; }
+        public IAccount Author { get; set; }
+
+
+        [return: AllowNull] // nullguard thinks a string.Format can return null. sigh.
+        public override string ToString()
+        {
+            return string.Format(CultureInfo.InvariantCulture, "id:{0} title:{1} created:{2:u} updated:{3:u}", Number, Title, CreatedAt, UpdatedAt);
+        }
+    }
+}