From ee7137046cab130fc2ca1e7231a4ce059fef4e0b Mon Sep 17 00:00:00 2001
From: neuecc <ils@neue.cc>
Date: Wed, 27 Dec 2023 23:40:25 +0900
Subject: [PATCH] AsUnitObservable, Cast, OfType

---
 sandbox/ConsoleApp1/Program.cs                |  2 +-
 src/R3/Operators/AsUnitObservable.cs          | 41 +++++++++++++++++++
 src/R3/Operators/Cast.cs                      | 37 +++++++++++++++++
 src/R3/Operators/OfType.cs                    | 39 ++++++++++++++++++
 .../OperatorTests/AsUnitObservableTest.cs     | 20 +++++++++
 tests/R3.Tests/OperatorTests/CastTest.cs      | 21 ++++++++++
 tests/R3.Tests/OperatorTests/OfTypeTest.cs    | 23 +++++++++++
 7 files changed, 182 insertions(+), 1 deletion(-)
 create mode 100644 src/R3/Operators/AsUnitObservable.cs
 create mode 100644 src/R3/Operators/Cast.cs
 create mode 100644 src/R3/Operators/OfType.cs
 create mode 100644 tests/R3.Tests/OperatorTests/AsUnitObservableTest.cs
 create mode 100644 tests/R3.Tests/OperatorTests/CastTest.cs
 create mode 100644 tests/R3.Tests/OperatorTests/OfTypeTest.cs

diff --git a/sandbox/ConsoleApp1/Program.cs b/sandbox/ConsoleApp1/Program.cs
index 5bf2674c..2d784ce2 100644
--- a/sandbox/ConsoleApp1/Program.cs
+++ b/sandbox/ConsoleApp1/Program.cs
@@ -31,7 +31,7 @@
 var range = System.Reactive.Linq.Observable.Range(1, 10);
 
 
-
+Enumerable.Range(1, 10).Cast<int>();
 
 // range.Catch(
 // range.Append(
diff --git a/src/R3/Operators/AsUnitObservable.cs b/src/R3/Operators/AsUnitObservable.cs
new file mode 100644
index 00000000..a3f096ba
--- /dev/null
+++ b/src/R3/Operators/AsUnitObservable.cs
@@ -0,0 +1,41 @@
+namespace R3;
+
+public static partial class ObservableExtensions
+{
+    public static Observable<Unit> AsUnitObservable<T>(this Observable<T> source)
+    {
+        if (source is Observable<Unit> unit)
+        {
+            return unit;
+        }
+
+        return new AsUnitObservable<T>(source);
+    }
+}
+
+internal sealed class AsUnitObservable<T>(Observable<T> source) : Observable<Unit>
+{
+    protected override IDisposable SubscribeCore(Observer<Unit> observer)
+    {
+        return source.Subscribe(new _AsUnitObservable(observer));
+    }
+
+    sealed class _AsUnitObservable(Observer<Unit> observer) : Observer<T>
+    {
+        protected override void OnNextCore(T value)
+        {
+            observer.OnNext(default);
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            observer.OnErrorResume(error);
+        }
+
+        protected override void OnCompletedCore(Result result)
+        {
+            observer.OnCompleted(result);
+        }
+    }
+}
+
diff --git a/src/R3/Operators/Cast.cs b/src/R3/Operators/Cast.cs
new file mode 100644
index 00000000..0df1029f
--- /dev/null
+++ b/src/R3/Operators/Cast.cs
@@ -0,0 +1,37 @@
+namespace R3;
+
+public static partial class ObservableExtensions
+{
+    public static Observable<TResult> Cast<T, TResult>(this Observable<T> source)
+    {
+        return new Cast<T, TResult>(source);
+    }
+}
+
+internal sealed class Cast<T, TResult>(Observable<T> source) : Observable<TResult>
+{
+    protected override IDisposable SubscribeCore(Observer<TResult> observer)
+    {
+        return source.Subscribe(new _Cast(observer));
+    }
+
+    sealed class _Cast(Observer<TResult> observer) : Observer<T>
+    {
+        protected override void OnNextCore(T value)
+        {
+            var v = (TResult?)(object?)value;
+            observer.OnNext(v!);
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            observer.OnErrorResume(error);
+        }
+
+        protected override void OnCompletedCore(Result result)
+        {
+            observer.OnCompleted(result);
+        }
+    }
+}
+
diff --git a/src/R3/Operators/OfType.cs b/src/R3/Operators/OfType.cs
new file mode 100644
index 00000000..f4c97e05
--- /dev/null
+++ b/src/R3/Operators/OfType.cs
@@ -0,0 +1,39 @@
+namespace R3;
+
+public static partial class ObservableExtensions
+{
+    public static Observable<TResult> OfType<T, TResult>(this Observable<T> source)
+    {
+        return new OfType<T, TResult>(source);
+    }
+}
+
+internal sealed class OfType<T, TResult>(Observable<T> source) : Observable<TResult>
+{
+    protected override IDisposable SubscribeCore(Observer<TResult> observer)
+    {
+        return source.Subscribe(new _OfType(observer));
+    }
+
+    sealed class _OfType(Observer<TResult> observer) : Observer<T>
+    {
+        protected override void OnNextCore(T value)
+        {
+            if (value is TResult v)
+            {
+                observer.OnNext(v);
+            }
+        }
+
+        protected override void OnErrorResumeCore(Exception error)
+        {
+            observer.OnErrorResume(error);
+        }
+
+        protected override void OnCompletedCore(Result result)
+        {
+            observer.OnCompleted(result);
+        }
+    }
+}
+
diff --git a/tests/R3.Tests/OperatorTests/AsUnitObservableTest.cs b/tests/R3.Tests/OperatorTests/AsUnitObservableTest.cs
new file mode 100644
index 00000000..cf82acf6
--- /dev/null
+++ b/tests/R3.Tests/OperatorTests/AsUnitObservableTest.cs
@@ -0,0 +1,20 @@
+namespace R3.Tests.OperatorTests;
+
+public class AsUnitObservableTest
+{
+    [Fact]
+    public void Test()
+    {
+        var subject = new Subject<int>();
+        using var list = subject.AsUnitObservable().ToLiveList();
+
+        subject.OnNext(10);
+        subject.OnNext(20);
+
+        list.AssertEqual([Unit.Default, Unit.Default]);
+
+        subject.OnCompleted();
+
+        list.AssertIsCompleted();
+    }
+}
diff --git a/tests/R3.Tests/OperatorTests/CastTest.cs b/tests/R3.Tests/OperatorTests/CastTest.cs
new file mode 100644
index 00000000..486376d5
--- /dev/null
+++ b/tests/R3.Tests/OperatorTests/CastTest.cs
@@ -0,0 +1,21 @@
+namespace R3.Tests.OperatorTests;
+
+public class CastTest
+{
+    [Fact]
+    public void Cast()
+    {
+        var subject = new Subject<object>();
+        using var list = subject.Cast<object, int>().ToLiveList();
+
+        subject.OnNext(10);
+        subject.OnNext(20);
+        subject.OnNext(30);
+
+        list.AssertEqual([10, 20, 30]);
+
+        subject.OnCompleted();
+
+        list.AssertIsCompleted();
+    }
+}
diff --git a/tests/R3.Tests/OperatorTests/OfTypeTest.cs b/tests/R3.Tests/OperatorTests/OfTypeTest.cs
new file mode 100644
index 00000000..a4a6e2c8
--- /dev/null
+++ b/tests/R3.Tests/OperatorTests/OfTypeTest.cs
@@ -0,0 +1,23 @@
+namespace R3.Tests.OperatorTests;
+
+public class OfTypeTest
+{
+    [Fact]
+    public void Test()
+    {
+        var subject = new Subject<object>();
+        using var list = subject.OfType<object, int>().ToLiveList();
+
+        subject.OnNext(10);
+        subject.OnNext("hello");
+        subject.OnNext(20);
+        subject.OnNext(30);
+        subject.OnNext("world");
+        subject.OnNext(40);
+
+        list.AssertEqual([10, 20, 30, 40]);
+
+        subject.OnCompleted();
+        list.AssertIsCompleted();
+    }
+}