Search WIP

src/Core/FileTime.Core.Abstraction/Extensions/DynamicDataExtensions.cs

@@ -8,8 +8,10 @@ public static class DynamicDataExtensions
 {
     private class DisposableContext<TParam, TTaskResult>
     {
+        private readonly SemaphoreSlim _semaphore = new(1, 1);
         private readonly Func<TParam, TTaskResult> _transformResult;
         private readonly TaskCompletionSource<TTaskResult?> _taskCompletionSource;
+        private bool _isFinished;
         public IDisposable? Disposable { get; set; }
 
         public DisposableContext(Func<TParam, TTaskResult> transformResult,
@@ -22,6 +24,7 @@ public static class DynamicDataExtensions
 
         public void OnNext(TParam param)
         {
+            if (IsFinished()) return;
             Disposable?.Dispose();
             var result = _transformResult(param);
             _taskCompletionSource.SetResult(result);
@@ -29,15 +32,27 @@ public static class DynamicDataExtensions
 
         public void OnError(Exception ex)
         {
+            if (IsFinished()) return;
             Disposable?.Dispose();
             _taskCompletionSource.SetException(ex);
         }
 
         public void OnCompleted()
         {
+            if (IsFinished()) return;
             Disposable?.Dispose();
             _taskCompletionSource.SetResult(default);
         }
+
+        private bool IsFinished()
+        {
+            _semaphore.Wait();
+            var finished = _isFinished;
+            _isFinished = true;
+            _semaphore.Release();
+
+            return finished;
+        }
     }
 
     public static async Task<IEnumerable<AbsolutePath>?> GetItemsAsync(
@@ -46,12 +61,12 @@ public static class DynamicDataExtensions
             .Select(s =>
                 s is null
                     ? new SourceList<AbsolutePath>().Connect().StartWithEmpty().ToCollection()
-                    : s.ToCollection())
+                    : s.StartWithEmpty().ToCollection())
             .Switch());
 
     public static async Task<IEnumerable<AbsolutePath>?> GetItemsAsync(
         this IObservable<IChangeSet<AbsolutePath, string>> stream)
-        => await GetItemsAsync(stream.ToCollection());
+        => await GetItemsAsync(stream.StartWithEmpty().ToCollection());
 
     public static Task<IEnumerable<AbsolutePath>?> GetItemsAsync(
         this IObservable<IReadOnlyCollection<AbsolutePath>> stream)

src/Core/FileTime.Core.Abstraction/Models/IContainer.cs

@@ -4,7 +4,7 @@ namespace FileTime.Core.Models;
 
 public interface IContainer : IItem
 {
-    IObservable<IObservable<IChangeSet<AbsolutePath, string>>?> Items { get; }
+    IObservable<IChangeSet<AbsolutePath, string>> Items { get; }
     IObservable<bool> IsLoading { get; }
     bool AllowRecursiveDeletion { get; }
 }