Merge pull request #398 from github/fixes/tracking-collection-allow-adding-before-subscribe

Allow manually adding items to TrackingCollection before subscription

Author: paladique

src/GitHub.Exports.Reactive/Collections/TrackingCollection.cs

@@ -131,9 +131,9 @@ enum TheAction
         IConnectableObservable<Unit> cachePump;
         ConcurrentQueue<ActionData> cache;
 
-        Subject<Unit> signalHaveData;
-        Subject<Unit> signalNeedData;
-        Subject<ActionData> dataListener;
+        ReplaySubject<Unit> signalHaveData;
+        ReplaySubject<Unit> signalNeedData;
+        ReplaySubject<ActionData> dataListener;
 
         bool resetting = false;
 
@@ -1110,11 +1110,11 @@ void Reset()
             originalSourceIsCompleted = false;
             signalOriginalSourceCompletion = false;
             cache = new ConcurrentQueue<ActionData>();
-            dataListener = new Subject<ActionData>();
+            dataListener = new ReplaySubject<ActionData>();
             disposables.Add(dataListener);
-            signalHaveData = new Subject<Unit>();
+            signalHaveData = new ReplaySubject<Unit>();
             disposables.Add(signalHaveData);
-            signalNeedData = new Subject<Unit>();
+            signalNeedData = new ReplaySubject<Unit>();
             disposables.Add(signalNeedData);
             originalSourceCompleted = new ReplaySubject<Unit>();
 

src/TrackingCollectionTests/TrackingCollectionTests.cs

@@ -1957,4 +1957,24 @@ public void RemovingWithNoObservableSetThrows()
         ITrackingCollection<Thing> col = new TrackingCollection<Thing>();
         Assert.Throws<InvalidOperationException>(() => col.RemoveItem(new Thing()));
     }
+
+    [Test]
+    public async void AddingBeforeSubscribingWorks()
+    {
+        ITrackingCollection<Thing> col = new TrackingCollection<Thing>(Observable.Empty<Thing>());
+        ReplaySubject<Thing> done = new ReplaySubject<Thing>();
+        col.AddItem(GetThing(1));
+        col.AddItem(GetThing(2));
+        var count = 0;
+        done.OnNext(null);
+        col.Subscribe(t =>
+        {
+            done.OnNext(t);
+            if (++count == 2)
+                done.OnCompleted();
+        }, () => {});
+
+        await Observable.Timeout(done, TimeSpan.FromMilliseconds(500));
+        Assert.AreEqual(2, col.Count);
+    }
 }