ElementAtAsync

Cysharp · Dec 14, 2023 · 8a6ccf0 · 8a6ccf0
1 parent 6096017
commit 8a6ccf0
Show file tree

Hide file tree

Showing 6 changed files with 238 additions and 129 deletions.
diff --git a/sandbox/ConsoleApp1/Program.cs b/sandbox/ConsoleApp1/Program.cs
@@ -26,52 +26,21 @@
 EventSystem.DefaultFrameProvider = new ThreadSleepFrameProvider(60);
 
 
-//var t = new Thread(() =>
-//{
-//    while (true)
-//    {
-//        Console.WriteLine("loop"); Thread.Sleep(60);
-//    }
-//});
-//t.IsBackground = true;
-//t.Start();
+// Enumerable.Empty<int>().ElementAtOrDefault(
 
-//var s = new NewThreadScheduler(_ => new Thread(() => { while (true) { Console.WriteLine("loop"); Thread.Sleep(60); } }));
+var i = Enumerable.Range(4, 10).ElementAtOrDefault(^0);
+Console.WriteLine(i);
 
-//s.Schedule(() => Console.WriteLine("do once"));
-//using var f = new ThreadSleepFrameProvider(60);
 
-var source = Event.EveryUpdate(ct.Token);
 
-
-
-source.DoOnDisposed(() => { Console.WriteLine("DISPOSED"); }).WriteLine();
-
-SubscriptionTracker.ForEachActiveTask(x =>
-{
-    Console.WriteLine(x);
-});
-
-
-
-Console.WriteLine("BeforeId:" + Thread.CurrentThread.ManagedThreadId);
-
-await source.WaitAsync();
-Console.WriteLine("Press Key to done.");
-
-
-await Task.Yield();
-
-Console.ReadLine();
-
-
-SubscriptionTracker.ForEachActiveTask(x =>
+IEnumerable<int> Range(int count)
 {
-    Console.WriteLine(x);
-});
-
-Console.WriteLine("----------------");
-Console.WriteLine("AfterId:" + Thread.CurrentThread.ManagedThreadId);
+    for (int i = 0; i < count; i++)
+    {
+        Console.WriteLine(i);
+        yield return i;
+    }
+}
 
 
 public static class Extensions
@@ -92,3 +61,4 @@ public void Dispose()
         CalledCount += 1;
     }
 }
+
diff --git a/src/R3/Factories/Range.cs b/src/R3/Factories/Range.cs
@@ -58,11 +58,12 @@ protected override IDisposable SubscribeCore(Subscriber<int> subscriber)
         {
             if (cancellationToken.IsCancellationRequested)
             {
+                subscriber.OnCompleted();
                 return Disposable.Empty;
             }
             subscriber.OnNext(start + i);
         }
-        subscriber.OnCompleted(default);
+        subscriber.OnCompleted();
         return Disposable.Empty;
     }
 }
diff --git a/src/R3/Factories/Repeat.cs b/src/R3/Factories/Repeat.cs
@@ -57,6 +57,7 @@ protected override IDisposable SubscribeCore(Subscriber<T> subscriber)
         {
             if (cancellationToken.IsCancellationRequested)
             {
+                subscriber.OnCompleted();
                 return Disposable.Empty;
             }
             subscriber.OnNext(value);

diff --git a/src/R3/Operators/ElementAtAsync.cs b/src/R3/Operators/ElementAtAsync.cs
@@ -0,0 +1,134 @@
+namespace R3;
+
+public static partial class EventExtensions
+{
+    public static Task<T> ElementAtAsync<T>(this Event<T> source, int index, CancellationToken cancellationToken = default)
+    {
+        if (index < 0) throw new ArgumentOutOfRangeException("index");
+
+        var subscriber = new ElementAtAsync<T>(index, false, default, cancellationToken);
+        source.Subscribe(subscriber);
+        return subscriber.Task;
+    }
+
+    public static Task<T> ElementAtAsync<T>(this Event<T> source, Index index, CancellationToken cancellationToken = default)
+    {
+        if (index.IsFromEnd)
+        {
+            if (index.Value <= 0) throw new ArgumentOutOfRangeException("index");
+            var subscriber = new ElementAtFromEndAsync<T>(index.Value, false, default, cancellationToken);
+            source.Subscribe(subscriber);
+            return subscriber.Task;
+        }
+        else
+        {
+            return ElementAtAsync(source, index.Value, cancellationToken);
+        }
+    }
+
+    public static Task<T> ElementAtOrDefaultAsync<T>(this Event<T> source, int index, T? defaultValue = default, CancellationToken cancellationToken = default)
+    {
+        if (index < 0) throw new ArgumentOutOfRangeException("index");
+        var subscriber = new ElementAtAsync<T>(index, true, defaultValue, cancellationToken);
+        source.Subscribe(subscriber);
+        return subscriber.Task;
+    }
+
+    public static Task<T> ElementAtOrDefaultAsync<T>(this Event<T> source, Index index, T? defaultValue = default, CancellationToken cancellationToken = default)
+    {
+        if (index.IsFromEnd)
+        {
+            var subscriber = new ElementAtFromEndAsync<T>(index.Value, true, defaultValue, cancellationToken);
+            source.Subscribe(subscriber);
+            return subscriber.Task;
+        }
+        else
+        {
+            return ElementAtOrDefaultAsync(source, index.Value, defaultValue, cancellationToken);
+        }
+    }
+}
+
+internal sealed class ElementAtAsync<T>(int index, bool useDefaultValue, T? defaultValue, CancellationToken cancellationToken)
+    : TaskSubscriberBase<T, T>(cancellationToken)
+{
+    int count = 0;
+
+    protected override void OnNextCore(T value)
+    {
+        if (count++ == index)
+        {
+            TrySetResult(value);
+        }
+    }
+
+    protected override void OnErrorResumeCore(Exception error)
+    {
+        TrySetException(error);
+    }
+
+    protected override void OnCompletedCore(Result result)
+    {
+        if (result.IsFailure)
+        {
+            TrySetException(result.Exception);
+        }
+        else
+        {
+            if (useDefaultValue)
+            {
+                TrySetResult(defaultValue!);
+            }
+            else
+            {
+                TrySetException(new ArgumentOutOfRangeException("index"));
+            }
+        }
+    }
+}
+
+// Index.IsFromEnd
+internal sealed class ElementAtFromEndAsync<T>(int fromEndIndex, bool useDefaultValue, T? defaultValue, CancellationToken cancellationToken)
+    : TaskSubscriberBase<T, T>(cancellationToken)
+{
+    Queue<T> queue = new Queue<T>(fromEndIndex);
+
+    protected override void OnNextCore(T value)
+    {
+        if (queue.Count == fromEndIndex && queue.Count != 0)
+        {
+            queue.Dequeue();
+        }
+
+        queue.Enqueue(value);
+    }
+
+    protected override void OnErrorResumeCore(Exception error)
+    {
+        TrySetException(error);
+    }
+
+    protected override void OnCompletedCore(Result result)
+    {
+        if (result.IsFailure)
+        {
+            TrySetException(result.Exception);
+            return;
+        }
+
+        if (queue.Count == fromEndIndex)
+        {
+            var value = queue.Dequeue();
+            TrySetResult(value);
+            return;
+        }
+
+        if (useDefaultValue)
+        {
+            TrySetResult(defaultValue!);
+            return;
+        }
+
+        TrySetException(new ArgumentOutOfRangeException("index"));
+    }
+}
diff --git a/src/R3/Operators/_Operators.cs b/src/R3/Operators/_Operators.cs
@@ -22,90 +22,3 @@ public static partial class EventExtensions
     // return tasks:
     // All, Any, Contains, SequenceEqual, ElementAt, ElementAtOrDefault, IsEmpty, MaxBy, MinBy, ToDictionary, ToLookup, ForEachAsync
 }
-
-// TODO: now working
-
-
-
-internal sealed class ElementAtAsync<T>(int index, bool useDefaultValue, T? defaultValue, CancellationToken cancellationToken)
-    : TaskSubscriberBase<T, T>(cancellationToken)
-{
-    int count = 0;
-    bool hasValue;
-
-    protected override void OnNextCore(T value)
-    {
-        hasValue = true;
-        if (count++ == index)
-        {
-            TrySetResult(value);
-        }
-    }
-
-    protected override void OnErrorResumeCore(Exception error)
-    {
-        TrySetException(error);
-    }
-
-    protected override void OnCompletedCore(Result result)
-    {
-        if (result.IsFailure)
-        {
-            // TODO:...
-        }
-        else
-        {
-            // TODO:...
-        }
-    }
-}
-
-// Index.IsFromEnd
-internal sealed class ElementAtFromEndAsync<T>(int fromEndIndex, bool useDefaultValue, T? defaultValue, CancellationToken cancellationToken)
-    : TaskSubscriberBase<T, T>(cancellationToken)
-{
-    bool hasValue;
-
-    Queue<T> queue = new Queue<T>(fromEndIndex);
-
-    protected override void OnNextCore(T value)
-    {
-        hasValue = true;
-        if (queue.Count == fromEndIndex)
-        {
-            queue.Dequeue();
-        }
-
-        queue.Enqueue(value);
-    }
-
-    protected override void OnErrorResumeCore(Exception error)
-    {
-        TrySetException(error);
-    }
-
-    protected override void OnCompletedCore(Result result)
-    {
-        if (queue.Count == fromEndIndex)
-        {
-            var value = queue.Dequeue();
-            TrySetResult(value);
-            return;
-        }
-
-        if (useDefaultValue)
-        {
-            TrySetResult(defaultValue!);
-            return;
-        }
-
-        if (!hasValue)
-        {
-            TrySetException(new InvalidOperationException("Sequence contains no elements"));
-        }
-        else
-        {
-            TrySetException(new ArgumentOutOfRangeException("index"));
-        }
-    }
-}