do not swallow digest complete or reset event

Author: jmd

src/DotJEM.Json.Index2.Management/Tracking/IIngestProgressTracker.cs

@@ -98,14 +98,27 @@ public void OnNext(IJsonDocumentSourceEvent value)
         switch (value)
         {
             case JsonDocumentCreated created:
-                observerTrackers.AddOrUpdate(value.Area, _ => throw new InvalidDataException(), (_, state) => state.UpdateState(created.Generation, created.Size));
+                observerTrackers.AddOrUpdate(value.Area,
+                    _ => throw new InvalidDataException(), 
+                    (_, state) => state.UpdateState(created.Generation, created.Size));
                 break;
             case JsonDocumentUpdated updated:
-                observerTrackers.AddOrUpdate(value.Area, _ => throw new InvalidDataException(), (_, state) => state.UpdateState(updated.Generation, updated.Size));
+                observerTrackers.AddOrUpdate(value.Area, 
+                    _ => throw new InvalidDataException(), 
+                    (_, state) => state.UpdateState(updated.Generation, updated.Size));
                 break;
             case JsonDocumentDeleted deleted:
-                observerTrackers.AddOrUpdate(value.Area, _ => throw new InvalidDataException(), (_, state) => state.UpdateState(deleted.Generation, deleted.Size));
+                observerTrackers.AddOrUpdate(value.Area,
+                    _ => throw new InvalidDataException(),
+                    (_, state) => state.UpdateState(deleted.Generation, deleted.Size));
                 break;
+            case JsonDocumentSourceDigestCompleted:
+                observerTrackers.AddOrUpdate(value.Area,
+                    _ => throw new InvalidDataException(), (_, state) => state);
+                break;
+            case JsonDocumentSourceReset:
+                break;
+
             default:
                 return;
         }
@@ -114,8 +127,10 @@ public void OnNext(IJsonDocumentSourceEvent value)
 
     public void UpdateState(StorageAreaIngestState state)
     {
-        observerTrackers.AddOrUpdate(state.Area, s => new StorageAreaIngestStateTracker(s, JsonSourceEventType.Initialized).UpdateState(state)
+        observerTrackers.AddOrUpdate(state.Area,
+            s => new StorageAreaIngestStateTracker(s, JsonSourceEventType.Initialized).UpdateState(state)
             , (s, tracker) => tracker.UpdateState(state));
+        InternalPublish(IngestState);
     }
 
     public void SetInitialized(bool initialized)

src/Stress/StressTester/Program.cs

@@ -68,7 +68,7 @@
 IWebTaskScheduler scheduler = new WebTaskScheduler();
 IJsonIndexManager jsonIndexManager = new JsonIndexManager(
     new JsonStorageDocumentSource(new JsonStorageAreaObserverFactory(storage, scheduler,areas)),
-    new JsonIndexSnapshotManager(index, new ZipSnapshotStrategy(".\\app_data\\snapshots"), scheduler, "10s"),
+    new JsonIndexSnapshotManager(index, new ZipSnapshotStrategy(".\\app_data\\snapshots"), scheduler, "30m"),
     index
 );
 
@@ -83,11 +83,6 @@
 {
     Console.Clear();
     Console.WriteLine("COMPLETED");
-    Console.WriteLine("COMPLETED");
-    Console.WriteLine("COMPLETED");
-    Console.WriteLine("COMPLETED");
-    Console.WriteLine("COMPLETED");
-    Console.WriteLine("COMPLETED");
 
 });