Add LiveList

sandbox/ConsoleApp1/LiveList.cs

@@ -0,0 +1,99 @@
+using R3;
+using System.Collections;
+using System.Runtime.InteropServices;
+
+public sealed class LiveList<T> : IReadOnlyList<T>, IDisposable
+{
+    readonly List<T> list = new List<T>();
+    readonly IDisposable sourceSubscription;
+
+    public LiveList(Event<T> source)
+    {
+        sourceSubscription = source.Subscribe(new ListSubscriber(list));
+    }
+
+    public T this[int index]
+    {
+        get
+        {
+            lock (list)
+            {
+                return list[index];
+            }
+        }
+    }
+
+    public int Count
+    {
+        get
+        {
+            lock (list)
+            {
+                return list.Count;
+            }
+        }
+    }
+
+    public void Dispose()
+    {
+        sourceSubscription.Dispose();
+    }
+
+    public void ForEach(Action<T> action)
+    {
+        lock (list)
+        {
+            var span = CollectionsMarshal.AsSpan(list);
+            foreach (ref var item in span)
+            {
+                action(item);
+            }
+        }
+    }
+
+    public void ForEach<TState>(Action<T, TState> action, TState state)
+    {
+        lock (list)
+        {
+            var span = CollectionsMarshal.AsSpan(list);
+            foreach (ref var item in span)
+            {
+                action(item, state);
+            }
+        }
+    }
+
+    public IEnumerator<T> GetEnumerator()
+    {
+        lock (list)
+        {
+            // snapshot
+            return CollectionsMarshal.AsSpan(list).ToArray().AsEnumerable().GetEnumerator();
+        }
+    }
+
+    IEnumerator IEnumerable.GetEnumerator()
+    {
+        lock (list)
+        {
+            // snapshot
+            return CollectionsMarshal.AsSpan(list).ToArray().AsEnumerable().GetEnumerator();
+        }
+    }
+
+    sealed class ListSubscriber(List<T> list) : Subscriber<T>
+    {
+        protected override void OnNextCore(T message)
+        {
+            lock (list)
+            {
+                list.Add(message);
+            }
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            EventSystem.GetUnhandledExceptionHandler().Invoke(error);
+        }
+    }
+}

sandbox/ConsoleApp1/Program.cs

@@ -4,7 +4,7 @@
 using System.Reactive;
 using System.Reactive.Concurrency;
 using System.Reactive.Linq;
-using System.Runtime.InteropServices;
+using System.Reactive.Subjects;
 using System.Threading.Channels;
 using ZLogger;
 //using System.Reactive.Disposables;
@@ -81,7 +81,7 @@
 
 Console.WriteLine($"Average: {Enumerable.Empty<int>().Average()}");
 
-
+// s.ToListObservable();
 
 // Observable.Throw(
 // s.Where(
@@ -173,5 +173,3 @@ public static IDisposable WriteLine<T, U>(this CompletableEvent<T, U> source)
         return source.Subscribe(x => Console.WriteLine(x), _ => Console.WriteLine("COMPLETED"));
     }
 }
-
-