improve reset. still observing a writer disposed.

Author: jmd

src/DotJEM.Json.Index2.Management/IJsonIndexManager.cs

@@ -13,6 +13,7 @@
 using DotJEM.Json.Index2.Management.Writer;
 using DotJEM.ObservableExtensions.InfoStreams;
 using DotJEM.Web.Scheduler;
+using Lucene.Net.Search;
 
 namespace DotJEM.Json.Index2.Management;
 
@@ -75,7 +76,7 @@ public async Task RunAsync()
 
         await Task
             .WhenAll(
-                jsonDocumentSource.RunAsync(),
+                jsonDocumentSource.StartAsync(),
                 snapshots.RunAsync(Tracker, restoredFromSnapshot))
             .ConfigureAwait(false);
     }
@@ -101,17 +102,25 @@ public async Task<bool> RestoreSnapshotAsync()
         return true;
     }
 
+    //TODO: Area is a concept that belongs to the specific storage implementation.
+    //      So is the generation, How can we pass these in a decoupled way?
     public Task UpdateGenerationAsync(string area, long generation)
     {
         jsonDocumentSource.UpdateGeneration(area, generation);
         return Task.CompletedTask;
     }
 
+
+    /// <summary>
+    /// Stops the underlying <see cref="IJsonDocumentSource"/>, deletes it's storage,
+    /// requests reset of the underlying <see cref="IJsonDocumentSource"/> and then starts it again.
+    /// </summary>
     public async Task ResetIndexAsync()
     {
-        
+        await jsonDocumentSource.StopAsync().ConfigureAwait(false);
         index.Storage.Delete();
         await jsonDocumentSource.ResetAsync().ConfigureAwait(false);
+        await jsonDocumentSource.StartAsync().ConfigureAwait(false);
     }
 
     private void CaptureChange(IJsonDocumentSourceEvent sourceEvent)
@@ -120,7 +129,7 @@ private void CaptureChange(IJsonDocumentSourceEvent sourceEvent)
         {
             switch (sourceEvent)
             {
-                case JsonDocumentSourceDigestCompleted commitSignal:
+                case JsonDocumentSourceDigestCompleted:
                     writer.Commit();
                     break;
                 case JsonDocumentCreated created:

src/DotJEM.Json.Index2.Management/Source/IJsonDocumentSource.cs

@@ -5,14 +5,49 @@
 
 namespace DotJEM.Json.Index2.Management.Source;
 
+/// <summary>
+/// 
+/// </summary>
 public interface IJsonDocumentSource
 {
+    /// <summary>
+    /// 
+    /// </summary>
     IInfoStream InfoStream { get; }
 
+    /// <summary>
+    /// 
+    /// </summary>
     IObservable<IJsonDocumentSourceEvent> DocumentChanges { get; }
+    
+    /// <summary>
+    /// 
+    /// </summary>
     IObservableValue<bool> Initialized { get; }
 
-    Task RunAsync();
+    /// <summary>
+    /// 
+    /// </summary>
+    /// <returns></returns>
+    Task StartAsync();
+    
+    /// <summary>
+    /// 
+    /// </summary>
+    /// <returns></returns>
+    Task StopAsync();
+
+    /// <summary>
+    /// 
+    /// </summary>
+    /// <param name="area"></param>
+    /// <param name="generation"></param>
     void UpdateGeneration(string area, long generation);
+    
+    /// <summary>
+    /// 
+    /// </summary>
+    /// <returns></returns>
     Task ResetAsync();
-}
+}
+

src/Stress/StressTester/Adapter/IJsonStorageAreaObserver.cs

@@ -15,14 +15,47 @@ public interface IJsonStorageAreaObserver : IJsonDocumentSource
     string AreaName { get; }
 }
 
+internal class Atomic<T>
+{
+    private T value;
+    private readonly object padlock = new ();
+
+    public Atomic(T value)
+    {
+        this.value = value;
+    }
+
+    public T Read() {
+        lock (padlock)
+        {
+            return value;
+        }
+    }
+
+    public T Exchange(T value)
+    {
+        lock (padlock)
+        {
+            T current = this.value;
+            this.value = value;
+            return current;
+        }
+    }
+
+    public static implicit operator Atomic<T>(T value) => new (value);
+    public static implicit operator T(Atomic<T> value) => value.Read();
+}
+
+
 public class JsonStorageAreaObserver : IJsonStorageAreaObserver
 {
     private readonly string pollInterval;
     private readonly IWebTaskScheduler scheduler;
     private readonly IStorageAreaLog log;
     private readonly DocumentChangesStream observable = new();
     private readonly IInfoStream<JsonStorageAreaObserver> infoStream = new InfoStream<JsonStorageAreaObserver>();
-    
+    private readonly Atomic<bool> started = false;
+
     private long generation = 0;
     private long initialGeneration = 0;
     private IScheduledTask task;
@@ -41,19 +74,25 @@ public JsonStorageAreaObserver(IStorageArea storageArea, IWebTaskScheduler sched
         log = storageArea.Log;
     }
 
-    public async Task RunAsync()
+    public async Task StartAsync()
     {
+        if(started.Exchange(true))
+            return;
+
         infoStream.WriteJsonSourceEvent(JsonSourceEventType.Starting, StorageArea.Name, $"Ingest starting for storageArea '{StorageArea.Name}'.");
         task = scheduler.Schedule($"JsonStorageAreaObserver:{StorageArea.Name}", _ => RunUpdateCheck(), pollInterval);
         task.InfoStream.Subscribe(infoStream);
-        await task.WhenCompleted();
+        await task.Signal(true);
     }
 
     public async Task StopAsync()
     {
+        if(!started.Exchange(false))
+            return;
+
         task.Dispose();
-        await task.WhenCompleted();
-        infoStream.WriteJsonSourceEvent(JsonSourceEventType.Stopped, StorageArea.Name, $"Initializing for storageArea '{StorageArea.Name}'.");
+        await task.WhenCompleted().ConfigureAwait(false);
+        infoStream.WriteJsonSourceEvent(JsonSourceEventType.Stopped, StorageArea.Name, $"Stopping for storageArea '{StorageArea.Name}'.");
     }
 
     public void UpdateGeneration(string area, long value)
@@ -70,7 +109,6 @@ public async Task ResetAsync()
     {
         UpdateGeneration(AreaName, initialGeneration);
         observable.Publish(new JsonDocumentSourceReset(AreaName));
-        task.Signal();
     }
 
     public void RunUpdateCheck()
@@ -80,23 +118,19 @@ public void RunUpdateCheck()
         Stopwatch timer = Stopwatch.StartNew();
         if (!Initialized.Value)
         {
-            BeforeInitialize();
             infoStream.WriteJsonSourceEvent(JsonSourceEventType.Initializing, StorageArea.Name, $"Initializing for storageArea '{StorageArea.Name}'.");
             using IStorageAreaLogReader changes = log.OpenLogReader(generation, Initialized.Value );
             PublishChanges(changes, row => new JsonDocumentCreated(row.Area, row.CreateEntity(), row.Size, new GenerationInfo(row.Generation, latestGeneration)));
             Initialized.Value  = true;
             infoStream.WriteJsonSourceEvent(JsonSourceEventType.Initialized, StorageArea.Name, $"Initialization complete for storageArea '{StorageArea.Name}' in {timer.Elapsed}.");
-            AfterInitialize();
 
         }
         else
         {
-            BeforeUpdate();
             infoStream.WriteJsonSourceEvent(JsonSourceEventType.Updating, StorageArea.Name, $"Checking updates for storageArea '{StorageArea.Name}'.");
             using IStorageAreaLogReader changes = log.OpenLogReader(generation, Initialized.Value );
             PublishChanges(changes, MapRow);
             infoStream.WriteJsonSourceEvent(JsonSourceEventType.Updated, StorageArea.Name, $"Done checking updates for storageArea '{StorageArea.Name}' in {timer.Elapsed}.");
-            AfterUpdate();
         }
         PublishCommitSignal();
 
@@ -128,9 +162,4 @@ void PublishChanges(IStorageAreaLogReader changes, Func<IChangeLogRow, IJsonDocu
             }
         }
     }
-    public virtual void BeforeInitialize() { }
-    public virtual void AfterInitialize() { }
-
-    public virtual void BeforeUpdate() { }
-    public virtual void AfterUpdate() { }
 }

src/Stress/StressTester/Adapter/JsonStorageDocumentSource.cs

@@ -55,10 +55,17 @@ private void InitializedChanged()
         this.Initialized.Value = observers.Values.All(observer => observer.Initialized.Value);
     }
 
-    public async Task RunAsync()
+    public async Task StartAsync()
     {
         await Task.WhenAll(
-            observers.Values.Select(async observer => await observer.RunAsync().ConfigureAwait(false))
+            observers.Values.Select(async observer => await observer.StartAsync().ConfigureAwait(false))
+        ).ConfigureAwait(false);
+    }
+
+    public async Task StopAsync()
+    {
+        await Task.WhenAll(
+            observers.Values.Select(async observer => await observer.StopAsync().ConfigureAwait(false))
         ).ConfigureAwait(false);
     }
 

src/Stress/StressTester/Program.cs

@@ -78,14 +78,6 @@
     //,genTask
 );
 
-
-jsonIndexManager.Tracker.WhenState(IngestInitializationState.Initialized).ContinueWith(task =>
-{
-    Console.Clear();
-    Console.WriteLine("COMPLETED");
-
-});
-
 while (true)
 {
     string? input = Console.ReadLine();
@@ -99,21 +91,12 @@
             await jsonIndexManager.TakeSnapshotAsync();
             break;
 
-        case 'C':
-            jsonIndexManager.Tracker.WhenState(IngestInitializationState.Initialized).ContinueWith(task =>
-            {
-                Console.Clear();
-                Console.WriteLine("COMPLETED");
-            });
-
-            break;
-
         case 'I':
             Console.Clear();
             break;
 
-        case 'L':
-            Console.Clear();
+        case 'R':
+            jsonIndexManager.ResetIndexAsync();
             break;
 
         case 'Q':