working ObserveOn

src/R3/Internal/PooledThreadPoolWorkItem.cs

@@ -0,0 +1,41 @@
+using System.Collections.Concurrent;
+
+namespace R3.Internal;
+
+internal sealed class PooledThreadPoolWorkItem<T> : IThreadPoolWorkItem
+{
+    static ConcurrentQueue<PooledThreadPoolWorkItem<T>> pool = new();
+
+    T state = default!;
+    Action<T> action = default!;
+
+    PooledThreadPoolWorkItem()
+    {
+    }
+
+    public static IThreadPoolWorkItem Create(Action<T> action, T state)
+    {
+        if (!pool.TryDequeue(out var item))
+        {
+            item = new PooledThreadPoolWorkItem<T>();
+        }
+
+        item.state = state;
+        item.action = action;
+        return item;
+    }
+
+    public void Execute()
+    {
+        try
+        {
+            action(state);
+        }
+        finally
+        {
+            state = default!;
+            action = default!;
+            pool.Enqueue(this);
+        }
+    }
+}

src/R3/Notification.cs

@@ -0,0 +1,60 @@
+namespace R3;
+
+public enum NotificationKind
+{
+    OnNext,
+    OnErrorResume,
+    OnCompleted
+}
+
+public readonly struct Notification<T>
+{
+    readonly NotificationKind kind;
+    readonly T? value;
+    readonly Exception? error;
+    readonly Result? result;
+
+    public NotificationKind Kind => kind;
+    public T? Value => value;
+    public Exception? Error => error;
+    public Result? Result => result;
+
+    public Notification(T value)
+    {
+        this.kind = NotificationKind.OnNext;
+        this.value = value;
+        this.error = null;
+        this.result = default;
+    }
+
+    public Notification(Exception error)
+    {
+        this.kind = NotificationKind.OnErrorResume;
+        this.value = default;
+        this.error = error;
+        this.result = default;
+    }
+
+    public Notification(Result result)
+    {
+        this.kind = NotificationKind.OnCompleted;
+        this.value = default;
+        this.error = default;
+        this.result = result;
+    }
+
+    public override string? ToString()
+    {
+        switch (kind)
+        {
+            case NotificationKind.OnNext:
+                return value!.ToString();
+            case NotificationKind.OnErrorResume:
+                return error!.ToString();
+            case NotificationKind.OnCompleted:
+                return result!.Value.ToString();
+            default:
+                return "";
+        }
+    }
+}

src/R3/Operators/Do.cs

@@ -43,7 +43,7 @@ protected override void OnErrorResumeCore(Exception error)
         protected override void OnCompletedCore(Result result)
         {
             onCompleted?.Invoke(result);
-            observer.OnCompleted();
+            observer.OnCompleted(result);
         }
 
         protected override void DisposeCore()
@@ -78,7 +78,7 @@ protected override void OnErrorResumeCore(Exception error)
         protected override void OnCompletedCore(Result result)
         {
             onCompleted?.Invoke(result, state);
-            observer.OnCompleted();
+            observer.OnCompleted(result);
         }
 
         protected override void DisposeCore()