From d360778263b3df599b6db281673a82bfbcf2c99e Mon Sep 17 00:00:00 2001 From: Federico Alterio Date: Sun, 19 Jul 2026 19:35:15 +0200 Subject: [PATCH 1/2] Dotnet reactive interop --- src/R3Async.R3Interop/ToAsyncObservable.cs | 119 ---------- src/R3Async.R3Interop/ToObservable.cs | 113 ---------- .../Factories/FromSystemObservableTest.cs | 208 ++++++++++++++++++ .../Operators/ToSystemObservableTest.cs | 126 +++++++++++ src/R3Async/AsyncToSyncStrategy.cs | 119 ++++++++++ src/R3Async/BackpressureStrategy.cs | 123 +++++++++++ src/R3Async/Factories/FromSystemObservable.cs | 148 +++++++++++++ src/R3Async/Operators/ToSystemObservable.cs | 99 +++++++++ 8 files changed, 823 insertions(+), 232 deletions(-) create mode 100644 src/R3Async.Tests/Factories/FromSystemObservableTest.cs create mode 100644 src/R3Async.Tests/Operators/ToSystemObservableTest.cs create mode 100644 src/R3Async/AsyncToSyncStrategy.cs create mode 100644 src/R3Async/BackpressureStrategy.cs create mode 100644 src/R3Async/Factories/FromSystemObservable.cs create mode 100644 src/R3Async/Operators/ToSystemObservable.cs diff --git a/src/R3Async.R3Interop/ToAsyncObservable.cs b/src/R3Async.R3Interop/ToAsyncObservable.cs index 7867d81..2142352 100644 --- a/src/R3Async.R3Interop/ToAsyncObservable.cs +++ b/src/R3Async.R3Interop/ToAsyncObservable.cs @@ -6,125 +6,6 @@ namespace R3Async.R3Interop; -public static class BackpressureStrategy -{ - public static BlockingBackpressureStrategy Blocking { get; } = new(); - - public static UnboundedChannelBackpressureStrategy FromUnboundedChannel(UnboundedChannelOptions? options = null) - { - return new(options); - } - - public static ChannelBackpressureStrategy FromUnboundedChannel(Action> onErrorResume, - UnboundedChannelOptions? options = null) - { - if (onErrorResume is null) - throw new ArgumentNullException(nameof(onErrorResume)); - - return UnboundedChannelBackpressureStrategy.ToChannelStrategy(options, onErrorResume); - } - - public static BoundedChannelBackpressureStrategy FromBoundedChannel(int capacity) - { - return new(new BoundedChannelOptions(capacity)); - } - - public static BoundedChannelBackpressureStrategy FromBoundedChannel(BoundedChannelOptions options) - { - if (options is null) - throw new ArgumentNullException(nameof(options)); - - return new(options); - } - - public static ChannelBackpressureStrategy FromBoundedChannel(Action> onErrorResume, int capacity) - { - return FromBoundedChannel(onErrorResume, new BoundedChannelOptions(capacity)); - } - - public static ChannelBackpressureStrategy FromBoundedChannel(Action> onErrorResume, BoundedChannelOptions options) - { - if (onErrorResume is null) - throw new ArgumentNullException(nameof(onErrorResume)); - if (options is null) - throw new ArgumentNullException(nameof(options)); - - return BoundedChannelBackpressureStrategy.ToChannelStrategy(options, onErrorResume); - } - - public static ChannelBackpressureStrategy FromChannel(Func> channelFactory, - Action>? onNext = null, - Action>? onErrorResume = null) - { - if (channelFactory is null) - throw new ArgumentNullException(nameof(channelFactory)); - - return new(channelFactory, onNext ?? (static (x, c) => c.TryWrite(x)), onErrorResume); - } -} - -public sealed class BlockingBackpressureStrategy -{ - internal BlockingBackpressureStrategy() - { - } -} - -public sealed class UnboundedChannelBackpressureStrategy -{ - readonly UnboundedChannelOptions? options; - - internal UnboundedChannelBackpressureStrategy(UnboundedChannelOptions? options) - { - this.options = options; - } - - internal ChannelBackpressureStrategy ToChannelStrategy() => ToChannelStrategy(options, null); - - internal static ChannelBackpressureStrategy ToChannelStrategy(UnboundedChannelOptions? options, - Action>? onErrorResume) - { - return BackpressureStrategy.FromChannel(() => options is null ? Channel.CreateUnbounded() : Channel.CreateUnbounded(options), - static (x, c) => c.TryWrite(x), - onErrorResume); - } -} - -public sealed class BoundedChannelBackpressureStrategy -{ - readonly BoundedChannelOptions options; - - internal BoundedChannelBackpressureStrategy(BoundedChannelOptions options) - { - this.options = options; - } - - internal ChannelBackpressureStrategy ToChannelStrategy() => ToChannelStrategy(options, null); - - internal static ChannelBackpressureStrategy ToChannelStrategy(BoundedChannelOptions options, - Action>? onErrorResume) - { - return BackpressureStrategy.FromChannel(() => Channel.CreateBounded(options), - onErrorResume: onErrorResume); - } -} - -public sealed class ChannelBackpressureStrategy -{ - internal ChannelBackpressureStrategy(Func> channelFactory, - Action> onNext, - Action>? onErrorResume) - { - ChannelFactory = channelFactory; - OnNext = onNext; - OnErrorResume = onErrorResume; - } - - internal Func> ChannelFactory { get; } - internal Action> OnNext { get; } - internal Action>? OnErrorResume { get; } -} - public static class R3ToAsyncObservableExtensions { /// diff --git a/src/R3Async.R3Interop/ToObservable.cs b/src/R3Async.R3Interop/ToObservable.cs index f5a6b31..f732b36 100644 --- a/src/R3Async.R3Interop/ToObservable.cs +++ b/src/R3Async.R3Interop/ToObservable.cs @@ -5,67 +5,6 @@ namespace R3Async.R3Interop; -public sealed class AsyncToSyncStrategy -{ - readonly Action? _onException; - - private AsyncToSyncStrategy(Action? onException) => _onException = onException; - - static readonly AsyncToSyncStrategy DefaultFireAndForget = new(null); - public static AsyncToSyncStrategy Blocking { get; } = new(null); - - public static AsyncToSyncStrategy FireAndForget(Action? onException = null) => - onException is null ? DefaultFireAndForget : new(onException); - - internal bool IsBlocking => ReferenceEquals(this, Blocking); - - internal void Execute(ValueTask operation) - { - if (IsBlocking) - { - if (operation.IsCompleted) - operation.GetAwaiter().GetResult(); - else - operation.AsTask().GetAwaiter().GetResult(); - - return; - } - - ExecuteFireAndForget(operation); - } - - async void ExecuteFireAndForget(ValueTask operation) - { - try - { - await operation; - } - catch (Exception e) - { - if (_onException is null) - { - UnhandledExceptionHandler.OnUnhandledException(e); - return; - } - - try - { - _onException(e); - } - catch (Exception handlerException) - { - UnhandledExceptionHandler.OnUnhandledException(handlerException); - } - } - } -} - -public sealed class ToObservableConfiguration -{ - public required AsyncToSyncStrategy SubscribeStrategy { get; init; } - public required AsyncToSyncStrategy DisposeStrategy { get; init; } -} - public static class AsyncToR3ObservableExtensions { /// @@ -145,56 +84,4 @@ protected override ValueTask OnCompletedAsyncCore(Result result) } } - sealed class SubscribedDisposable(IAsyncDisposable subscription, AsyncToSyncStrategy disposeStrategy) : IDisposable - { - int _disposed; - - public void Dispose() - { - if (Interlocked.Exchange(ref _disposed, 1) != 0) - return; - - disposeStrategy.Execute(DisposeCoreAsync()); - } - - ValueTask DisposeCoreAsync() - { - try - { - var a = subscription.DisposeAsync(); - return a; - } - catch (Exception e) - { - return new(Task.FromException(e)); - } - } - } - - sealed class PendingSubscriptionDisposable(ValueTask subscriptionTask, CancellationTokenSource cts, AsyncToSyncStrategy disposeStrategy) : IDisposable - { - int _disposed; - - public void Dispose() - { - if (Interlocked.Exchange(ref _disposed, 1) != 0) - return; - - cts.Cancel(); - disposeStrategy.Execute(DisposeCoreAsync()); - } - - async ValueTask DisposeCoreAsync() - { - try - { - var subscription = await subscriptionTask; - await subscription.DisposeAsync(); - } - finally - { - cts.Dispose(); - } - } - } } diff --git a/src/R3Async.Tests/Factories/FromSystemObservableTest.cs b/src/R3Async.Tests/Factories/FromSystemObservableTest.cs new file mode 100644 index 0000000..7491502 --- /dev/null +++ b/src/R3Async.Tests/Factories/FromSystemObservableTest.cs @@ -0,0 +1,208 @@ +using Shouldly; +#pragma warning disable CS1998 + +namespace R3Async.Tests.Factories; + +public class FromSystemObservableTest +{ + [Fact] + public async Task ToAsyncObservable_Blocking_ForwardsValuesAndCompletion() + { + var source = new SystemSubject(); + var observable = source.ToAsyncObservable(BackpressureStrategy.Blocking); + + var results = new List(); + var completedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var subscription = await observable.SubscribeAsync( + async (x, token) => results.Add(x), + async (ex, token) => { }, + async result => completedTcs.TrySetResult(result), + CancellationToken.None); + + source.OnNext(1); + source.OnNext(2); + source.OnCompleted(); + + var result = await completedTcs.Task; + result.IsSuccess.ShouldBeTrue(); + results.ShouldBe(new[] { 1, 2 }); + } + + [Fact] + public async Task ToAsyncObservable_Blocking_OnErrorBecomesFailureCompletion() + { + var source = new SystemSubject(); + var observable = source.ToAsyncObservable(BackpressureStrategy.Blocking); + + var completedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var subscription = await observable.SubscribeAsync( + async (x, token) => { }, + async (ex, token) => { }, + async result => completedTcs.TrySetResult(result), + CancellationToken.None); + + var expected = new InvalidOperationException("boom"); + source.OnError(expected); + + var result = await completedTcs.Task; + result.IsFailure.ShouldBeTrue(); + result.Exception.ShouldBe(expected); + } + + [Fact] + public async Task ToAsyncObservable_Blocking_DisposeUnsubscribesFromSource() + { + var source = new SystemSubject(); + var observable = source.ToAsyncObservable(BackpressureStrategy.Blocking); + + var subscription = await observable.SubscribeAsync(async (x, token) => { }, CancellationToken.None); + source.ObserverCount.ShouldBe(1); + + await subscription.DisposeAsync(); + source.ObserverCount.ShouldBe(0); + } + + [Fact] + public async Task ToAsyncObservable_UnboundedChannel_ForwardsValuesAndCompletion() + { + var source = new SystemSubject(); + var observable = source.ToAsyncObservable(BackpressureStrategy.FromUnboundedChannel()); + + var results = new List(); + var completedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var subscription = await observable.SubscribeAsync( + async (x, token) => results.Add(x), + async (ex, token) => { }, + async result => completedTcs.TrySetResult(result), + CancellationToken.None); + + source.OnNext(1); + source.OnNext(2); + source.OnCompleted(); + + var result = await completedTcs.Task; + result.IsSuccess.ShouldBeTrue(); + results.ShouldBe(new[] { 1, 2 }); + } + + [Fact] + public async Task ToAsyncObservable_UnboundedChannel_OnErrorBecomesFailureCompletion() + { + var source = new SystemSubject(); + var observable = source.ToAsyncObservable(BackpressureStrategy.FromUnboundedChannel()); + + var completedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + await using var subscription = await observable.SubscribeAsync( + async (x, token) => { }, + async (ex, token) => { }, + async result => completedTcs.TrySetResult(result), + CancellationToken.None); + + var expected = new InvalidOperationException("boom"); + source.OnError(expected); + + var result = await completedTcs.Task; + result.IsFailure.ShouldBeTrue(); + result.Exception.ShouldBe(expected); + } + + [Fact] + public async Task ToAsyncObservable_UnboundedChannel_DisposeUnsubscribesFromSource() + { + var source = new SystemSubject(); + var observable = source.ToAsyncObservable(BackpressureStrategy.FromUnboundedChannel()); + + var subscription = await observable.SubscribeAsync(async (x, token) => { }, CancellationToken.None); + source.ObserverCount.ShouldBe(1); + + await subscription.DisposeAsync(); + source.ObserverCount.ShouldBe(0); + } + + sealed class SystemSubject : IObservable + { + readonly object _gate = new(); + readonly List> _observers = new(); + bool _terminated; + + public int ObserverCount + { + get + { + lock (_gate) + { + return _observers.Count; + } + } + } + + public IDisposable Subscribe(IObserver observer) + { + lock (_gate) + { + _observers.Add(observer); + } + + return new Unsubscriber(this, observer); + } + + public void OnNext(T value) + { + foreach (var observer in Snapshot()) + { + observer.OnNext(value); + } + } + + public void OnError(Exception error) + { + if (Terminate()) return; + foreach (var observer in Snapshot()) + { + observer.OnError(error); + } + } + + public void OnCompleted() + { + if (Terminate()) return; + foreach (var observer in Snapshot()) + { + observer.OnCompleted(); + } + } + + bool Terminate() + { + lock (_gate) + { + if (_terminated) return true; + _terminated = true; + return false; + } + } + + IObserver[] Snapshot() + { + lock (_gate) + { + return _observers.ToArray(); + } + } + + sealed class Unsubscriber(SystemSubject parent, IObserver observer) : IDisposable + { + public void Dispose() + { + lock (parent._gate) + { + parent._observers.Remove(observer); + } + } + } + } +} diff --git a/src/R3Async.Tests/Operators/ToSystemObservableTest.cs b/src/R3Async.Tests/Operators/ToSystemObservableTest.cs new file mode 100644 index 0000000..aab4a19 --- /dev/null +++ b/src/R3Async.Tests/Operators/ToSystemObservableTest.cs @@ -0,0 +1,126 @@ +using R3Async.Subjects; +using Shouldly; + +namespace R3Async.Tests.Operators; + +public class ToSystemObservableTest +{ + static ToObservableConfiguration BlockingConfiguration { get; } = new() + { + SubscribeStrategy = AsyncToSyncStrategy.Blocking, + DisposeStrategy = AsyncToSyncStrategy.Blocking, + }; + + [Fact] + public async Task ToSystemObservable_ForwardsValuesAndCompletion() + { + var subject = Subject.Create(); + var observable = subject.Values.ToSystemObservable(BlockingConfiguration); + + var observer = new RecordingObserver(); + using var subscription = observable.Subscribe(observer); + + await subject.OnNextAsync(1, CancellationToken.None); + await subject.OnNextAsync(2, CancellationToken.None); + await subject.OnCompletedAsync(Result.Success); + + observer.Values.ShouldBe(new[] { 1, 2 }); + observer.Completed.ShouldBeTrue(); + observer.Error.ShouldBeNull(); + } + + [Fact] + public async Task ToSystemObservable_FailureCompletionBecomesOnError() + { + var subject = Subject.Create(); + var observable = subject.Values.ToSystemObservable(BlockingConfiguration); + + var observer = new RecordingObserver(); + using var subscription = observable.Subscribe(observer); + + var expected = new InvalidOperationException("boom"); + await subject.OnCompletedAsync(Result.Failure(expected)); + + observer.Completed.ShouldBeFalse(); + observer.Error.ShouldBe(expected); + } + + [Fact] + public async Task ToSystemObservable_OnErrorResumeTerminatesWithOnError() + { + var subject = Subject.Create(); + var observable = subject.Values.ToSystemObservable(BlockingConfiguration); + + var observer = new RecordingObserver(); + using var subscription = observable.Subscribe(observer); + + var expected = new InvalidOperationException("boom"); + await subject.OnNextAsync(1, CancellationToken.None); + await subject.OnErrorResumeAsync(expected, CancellationToken.None); + await subject.OnNextAsync(2, CancellationToken.None); + + observer.Values.ShouldBe(new[] { 1 }); + observer.Error.ShouldBe(expected); + observer.Completed.ShouldBeFalse(); + } + + [Fact] + public async Task ToSystemObservable_DisposeStopsDelivery() + { + var subject = Subject.Create(); + var observable = subject.Values.ToSystemObservable(BlockingConfiguration); + + var observer = new RecordingObserver(); + var subscription = observable.Subscribe(observer); + + await subject.OnNextAsync(1, CancellationToken.None); + subscription.Dispose(); + await subject.OnNextAsync(2, CancellationToken.None); + + observer.Values.ShouldBe(new[] { 1 }); + observer.Completed.ShouldBeFalse(); + observer.Error.ShouldBeNull(); + } + + [Fact] + public async Task ToSystemObservable_FireAndForgetStrategiesDeliverValues() + { + var configuration = new ToObservableConfiguration + { + SubscribeStrategy = AsyncToSyncStrategy.FireAndForget(), + DisposeStrategy = AsyncToSyncStrategy.FireAndForget(), + }; + + var subject = Subject.Create(); + var observable = subject.Values.ToSystemObservable(configuration); + + var observer = new RecordingObserver(); + var completedTcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + observer.OnCompletedCallback = () => completedTcs.TrySetResult(true); + + using var subscription = observable.Subscribe(observer); + + await subject.OnNextAsync(42, CancellationToken.None); + await subject.OnCompletedAsync(Result.Success); + + (await completedTcs.Task).ShouldBeTrue(); + observer.Values.ShouldBe(new[] { 42 }); + } + + sealed class RecordingObserver : IObserver + { + public List Values { get; } = new(); + public Exception? Error { get; private set; } + public bool Completed { get; private set; } + public Action? OnCompletedCallback { get; set; } + + public void OnNext(T value) => Values.Add(value); + public void OnError(Exception error) => Error = error; + + public void OnCompleted() + { + Completed = true; + OnCompletedCallback?.Invoke(); + } + } +} diff --git a/src/R3Async/AsyncToSyncStrategy.cs b/src/R3Async/AsyncToSyncStrategy.cs new file mode 100644 index 0000000..1c5ef74 --- /dev/null +++ b/src/R3Async/AsyncToSyncStrategy.cs @@ -0,0 +1,119 @@ +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace R3Async; + +public sealed class AsyncToSyncStrategy +{ + readonly Action? _onException; + + private AsyncToSyncStrategy(Action? onException) => _onException = onException; + + static readonly AsyncToSyncStrategy DefaultFireAndForget = new(null); + public static AsyncToSyncStrategy Blocking { get; } = new(null); + + public static AsyncToSyncStrategy FireAndForget(Action? onException = null) => + onException is null ? DefaultFireAndForget : new(onException); + + internal bool IsBlocking => ReferenceEquals(this, Blocking); + + internal void Execute(ValueTask operation) + { + if (IsBlocking) + { + if (operation.IsCompleted) + operation.GetAwaiter().GetResult(); + else + operation.AsTask().GetAwaiter().GetResult(); + + return; + } + + ExecuteFireAndForget(operation); + } + + async void ExecuteFireAndForget(ValueTask operation) + { + try + { + await operation; + } + catch (Exception e) + { + if (_onException is null) + { + UnhandledExceptionHandler.OnUnhandledException(e); + return; + } + + try + { + _onException(e); + } + catch (Exception handlerException) + { + UnhandledExceptionHandler.OnUnhandledException(handlerException); + } + } + } +} + +public sealed class ToObservableConfiguration +{ + public required AsyncToSyncStrategy SubscribeStrategy { get; init; } + public required AsyncToSyncStrategy DisposeStrategy { get; init; } +} + +internal sealed class SubscribedDisposable(IAsyncDisposable subscription, AsyncToSyncStrategy disposeStrategy) : IDisposable +{ + int _disposed; + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + disposeStrategy.Execute(DisposeCoreAsync()); + } + + ValueTask DisposeCoreAsync() + { + try + { + var a = subscription.DisposeAsync(); + return a; + } + catch (Exception e) + { + return new(Task.FromException(e)); + } + } +} + +internal sealed class PendingSubscriptionDisposable(ValueTask subscriptionTask, CancellationTokenSource cts, AsyncToSyncStrategy disposeStrategy) : IDisposable +{ + int _disposed; + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) != 0) + return; + + cts.Cancel(); + disposeStrategy.Execute(DisposeCoreAsync()); + } + + async ValueTask DisposeCoreAsync() + { + try + { + var subscription = await subscriptionTask; + await subscription.DisposeAsync(); + } + finally + { + cts.Dispose(); + } + } +} diff --git a/src/R3Async/BackpressureStrategy.cs b/src/R3Async/BackpressureStrategy.cs new file mode 100644 index 0000000..15b50e3 --- /dev/null +++ b/src/R3Async/BackpressureStrategy.cs @@ -0,0 +1,123 @@ +using System; +using System.Threading.Channels; + +namespace R3Async; + +public static class BackpressureStrategy +{ + public static BlockingBackpressureStrategy Blocking { get; } = new(); + + public static UnboundedChannelBackpressureStrategy FromUnboundedChannel(UnboundedChannelOptions? options = null) + { + return new(options); + } + + public static ChannelBackpressureStrategy FromUnboundedChannel(Action> onErrorResume, + UnboundedChannelOptions? options = null) + { + if (onErrorResume is null) + throw new ArgumentNullException(nameof(onErrorResume)); + + return UnboundedChannelBackpressureStrategy.ToChannelStrategy(options, onErrorResume); + } + + public static BoundedChannelBackpressureStrategy FromBoundedChannel(int capacity) + { + return new(new BoundedChannelOptions(capacity)); + } + + public static BoundedChannelBackpressureStrategy FromBoundedChannel(BoundedChannelOptions options) + { + if (options is null) + throw new ArgumentNullException(nameof(options)); + + return new(options); + } + + public static ChannelBackpressureStrategy FromBoundedChannel(Action> onErrorResume, int capacity) + { + return FromBoundedChannel(onErrorResume, new BoundedChannelOptions(capacity)); + } + + public static ChannelBackpressureStrategy FromBoundedChannel(Action> onErrorResume, BoundedChannelOptions options) + { + if (onErrorResume is null) + throw new ArgumentNullException(nameof(onErrorResume)); + if (options is null) + throw new ArgumentNullException(nameof(options)); + + return BoundedChannelBackpressureStrategy.ToChannelStrategy(options, onErrorResume); + } + + public static ChannelBackpressureStrategy FromChannel(Func> channelFactory, + Action>? onNext = null, + Action>? onErrorResume = null) + { + if (channelFactory is null) + throw new ArgumentNullException(nameof(channelFactory)); + + return new(channelFactory, onNext ?? (static (x, c) => c.TryWrite(x)), onErrorResume); + } +} + +public sealed class BlockingBackpressureStrategy +{ + internal BlockingBackpressureStrategy() + { + } +} + +public sealed class UnboundedChannelBackpressureStrategy +{ + readonly UnboundedChannelOptions? options; + + internal UnboundedChannelBackpressureStrategy(UnboundedChannelOptions? options) + { + this.options = options; + } + + internal ChannelBackpressureStrategy ToChannelStrategy() => ToChannelStrategy(options, null); + + internal static ChannelBackpressureStrategy ToChannelStrategy(UnboundedChannelOptions? options, + Action>? onErrorResume) + { + return BackpressureStrategy.FromChannel(() => options is null ? Channel.CreateUnbounded() : Channel.CreateUnbounded(options), + static (x, c) => c.TryWrite(x), + onErrorResume); + } +} + +public sealed class BoundedChannelBackpressureStrategy +{ + readonly BoundedChannelOptions options; + + internal BoundedChannelBackpressureStrategy(BoundedChannelOptions options) + { + this.options = options; + } + + internal ChannelBackpressureStrategy ToChannelStrategy() => ToChannelStrategy(options, null); + + internal static ChannelBackpressureStrategy ToChannelStrategy(BoundedChannelOptions options, + Action>? onErrorResume) + { + return BackpressureStrategy.FromChannel(() => Channel.CreateBounded(options), + onErrorResume: onErrorResume); + } +} + +public sealed class ChannelBackpressureStrategy +{ + internal ChannelBackpressureStrategy(Func> channelFactory, + Action> onNext, + Action>? onErrorResume) + { + ChannelFactory = channelFactory; + OnNext = onNext; + OnErrorResume = onErrorResume; + } + + internal Func> ChannelFactory { get; } + internal Action> OnNext { get; } + internal Action>? OnErrorResume { get; } +} diff --git a/src/R3Async/Factories/FromSystemObservable.cs b/src/R3Async/Factories/FromSystemObservable.cs new file mode 100644 index 0000000..80fcc65 --- /dev/null +++ b/src/R3Async/Factories/FromSystemObservable.cs @@ -0,0 +1,148 @@ +using System; +using System.Threading; +using System.Threading.Channels; +using System.Threading.Tasks; + +namespace R3Async; + +public static class SystemToAsyncObservableExtensions +{ + /// + /// Converts a into an delivering each notification + /// synchronously on the emitting thread, which blocks until the async observer has finished processing it. + /// A slow consumer therefore slows down the source itself. OnError is mapped to a failure completion. + /// + public static AsyncObservable ToAsyncObservable(this IObservable @this, BlockingBackpressureStrategy backpressureStrategy) + { + if (@this is null) + throw new ArgumentNullException(nameof(@this)); + if (backpressureStrategy is null) + throw new ArgumentNullException(nameof(backpressureStrategy)); + + return CreateBlocking(@this); + } + + /// + /// Converts a into an buffering notifications through + /// an unbounded channel: the source is never blocked, and a background loop drains the buffer into the async observer. + /// OnError is mapped to a failure completion. + /// + public static AsyncObservable ToAsyncObservable(this IObservable @this, UnboundedChannelBackpressureStrategy backpressureStrategy) + { + if (@this is null) + throw new ArgumentNullException(nameof(@this)); + if (backpressureStrategy is null) + throw new ArgumentNullException(nameof(backpressureStrategy)); + + return CreateNonBlocking(@this, backpressureStrategy.ToChannelStrategy()); + } + + /// + /// Converts a into an buffering notifications through + /// a bounded channel drained by a background loop. Values are written with , + /// so what happens when the buffer is full is governed by : with drop modes + /// values are discarded accordingly, while with (the default) the write simply + /// fails and the value is lost. Use with a custom onNext for waiting + /// semantics. OnError is mapped to a failure completion. + /// + public static AsyncObservable ToAsyncObservable(this IObservable @this, BoundedChannelBackpressureStrategy backpressureStrategy) + { + if (@this is null) + throw new ArgumentNullException(nameof(@this)); + if (backpressureStrategy is null) + throw new ArgumentNullException(nameof(backpressureStrategy)); + + return CreateNonBlocking(@this, backpressureStrategy.ToChannelStrategy()); + } + + /// + /// Converts a into an buffering notifications through + /// a user-provided channel: the strategy supplies the channel and how values are written to it. A background loop + /// drains the channel into the async observer. The IObservable grammar has no resumable-error channel, so the + /// strategy's onErrorResume hook is never invoked; OnError is mapped to a failure completion. + /// + public static AsyncObservable ToAsyncObservable(this IObservable @this, ChannelBackpressureStrategy backpressureStrategy) + { + if (@this is null) + throw new ArgumentNullException(nameof(@this)); + if (backpressureStrategy is null) + throw new ArgumentNullException(nameof(backpressureStrategy)); + + return CreateNonBlocking(@this, backpressureStrategy); + } + + static AsyncObservable CreateBlocking(IObservable source) + { + return AsyncObservable.Create((observer, cancellationToken) => + { + var subscription = source.Subscribe(new BlockingObserver(observer)); + return new ValueTask(new SubscriptionAsyncDisposable(subscription)); + }); + } + + static AsyncObservable CreateNonBlocking(IObservable source, ChannelBackpressureStrategy backpressureStrategy) + { + return AsyncObservable.CreateAsBackgroundJob(async (observer, cancellationToken) => + { + var channel = backpressureStrategy.ChannelFactory(); + + using var subscription = source.Subscribe(new ChannelObserver(channel.Writer, backpressureStrategy.OnNext)); + + try + { + while (await channel.Reader.WaitToReadAsync(cancellationToken)) + { + while (channel.Reader.TryRead(out var value)) + { + await observer.OnNextAsync(value, cancellationToken); + } + } + + await observer.OnCompletedAsync(Result.Success); + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception e) + { + await observer.OnCompletedAsync(Result.Failure(e)); + } + }, startSynchronously: true); + } + + sealed class BlockingObserver(AsyncObserver observer) : IObserver + { + public void OnNext(T value) => WaitSynchronously(observer.OnNextAsync(value, CancellationToken.None)); + + public void OnError(Exception error) => WaitSynchronously(observer.OnCompletedAsync(Result.Failure(error))); + + public void OnCompleted() => WaitSynchronously(observer.OnCompletedAsync(Result.Success)); + + static void WaitSynchronously(ValueTask task) + { + if (task.IsCompletedSuccessfully) + return; + + task.AsTask().GetAwaiter().GetResult(); + } + } + + sealed class SubscriptionAsyncDisposable(IDisposable subscription) : IAsyncDisposable + { + public ValueTask DisposeAsync() + { + subscription.Dispose(); + return default; + } + } + + sealed class ChannelObserver(ChannelWriter writer, Action> onNext) : IObserver + { + public void OnNext(T value) => onNext(value, writer); + + public void OnError(Exception error) => writer.TryComplete(error); + + public void OnCompleted() => writer.TryComplete(); + } +} diff --git a/src/R3Async/Operators/ToSystemObservable.cs b/src/R3Async/Operators/ToSystemObservable.cs new file mode 100644 index 0000000..984bc56 --- /dev/null +++ b/src/R3Async/Operators/ToSystemObservable.cs @@ -0,0 +1,99 @@ +using System; +using System.Threading; +using System.Threading.Tasks; + +namespace R3Async; + +public static class AsyncToSystemObservableExtensions +{ + /// + /// Converts an into a . Since IObservable's Subscribe + /// and Dispose are synchronous while R3Async's are not, the configuration decides per operation how the async work + /// is consumed: blocks the caller until it completes (exceptions + /// propagate), while starts it without waiting and routes failures + /// to its optional onException callback (or the ). + /// The IObservable grammar has no resumable-error channel, so OnErrorResume terminates the sequence via OnError and + /// tears down the subscription. + /// + public static IObservable ToSystemObservable(this AsyncObservable @this, ToObservableConfiguration configuration) + { + if (@this is null) + throw new ArgumentNullException(nameof(@this)); + + return new AsyncToSystemObservableAdapter(@this, configuration); + } + + sealed class AsyncToSystemObservableAdapter(AsyncObservable source, ToObservableConfiguration configuration) : IObservable + { + public IDisposable Subscribe(IObserver observer) + { + if (observer is null) + throw new ArgumentNullException(nameof(observer)); + + return configuration.SubscribeStrategy.IsBlocking + ? SubscribeBlocking(observer) + : SubscribeFireAndForget(observer); + } + + IDisposable SubscribeBlocking(IObserver observer) + { + var subscriptionTask = source.SubscribeAsync(new ObserverAdapter(observer), CancellationToken.None); + var subscription = subscriptionTask.IsCompletedSuccessfully + ? subscriptionTask.GetAwaiter().GetResult() + : subscriptionTask.AsTask().GetAwaiter().GetResult(); + + return new SubscribedDisposable(subscription, configuration.DisposeStrategy); + } + + IDisposable SubscribeFireAndForget(IObserver observer) + { + var cts = new CancellationTokenSource(); + + ValueTask subscriptionTask; + try + { + subscriptionTask = source.SubscribeAsync(new ObserverAdapter(observer), cts.Token).Preserve(); + } + catch (Exception e) + { + subscriptionTask = new(Task.FromException(e)); + } + + configuration.SubscribeStrategy.Execute(AwaitSubscription(subscriptionTask)); + + return new PendingSubscriptionDisposable(subscriptionTask, cts, configuration.DisposeStrategy); + } + + static async ValueTask AwaitSubscription(ValueTask subscriptionTask) => await subscriptionTask; + } + + sealed class ObserverAdapter(IObserver observer) : AsyncObserver + { + protected override ValueTask OnNextAsyncCore(T value, CancellationToken cancellationToken) + { + observer.OnNext(value); + return default; + } + + protected override async ValueTask OnErrorResumeAsyncCore(Exception error, CancellationToken cancellationToken) + { + // IObservable has no resumable-error channel: OnError is terminal, so the subscription must die with it. + observer.OnError(error); + await DisposeAsync(); + } + + protected override ValueTask OnCompletedAsyncCore(Result result) + { + if (result.IsSuccess) + { + observer.OnCompleted(); + } + else + { + observer.OnError(result.Exception); + } + + return default; + } + } +} From 0fb8842b8b0faac9aaddcc469cfca1726bd82bd9 Mon Sep 17 00:00:00 2001 From: Federico Alterio Date: Sun, 19 Jul 2026 19:36:42 +0200 Subject: [PATCH 2/2] updated readme --- README.md | 51 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 51 insertions(+) diff --git a/README.md b/README.md index e60fdcc..c8d5b0c 100644 --- a/README.md +++ b/README.md @@ -889,6 +889,55 @@ UnhandledExceptionHandler.Register(exception => Note: `OperationCanceledException` is automatically ignored by the unhandled exception handler. +## System.IObservable Interop + +The core package can convert between `System.IObservable` and `AsyncObservable` in both directions, with no extra package required. Since `IObservable` is synchronous while `AsyncObservable` is not, both directions require you to state explicitly how the sync/async boundary is handled. + +### IObservable → AsyncObservable + +`ToAsyncObservable` converts a `System.IObservable` into an `AsyncObservable`. Since the source pushes values synchronously while the async observer may be slow, you must choose a `BackpressureStrategy`: + +```csharp +// Blocking: each notification is delivered synchronously on the emitting thread, +// which blocks until the async observer finishes processing it +var asyncObservable = observable.ToAsyncObservable(BackpressureStrategy.Blocking); + +// Unbounded channel: values are buffered without limit and consumed by a background loop +var asyncObservable = observable.ToAsyncObservable(BackpressureStrategy.FromUnboundedChannel()); + +// Bounded channel: values are buffered up to a capacity +var asyncObservable = observable.ToAsyncObservable(BackpressureStrategy.FromBoundedChannel( + new BoundedChannelOptions(16) { FullMode = BoundedChannelFullMode.DropOldest })); + +// Full control: provide the channel and how values are written to it +var asyncObservable = observable.ToAsyncObservable(BackpressureStrategy.FromChannel( + () => Channel.CreateBounded(2), + onNext: (value, writer) => writer.WriteAsync(value).AsTask().GetAwaiter().GetResult())); +``` + +Notes: + +- Channel-based strategies write with `TryWrite` by default, so with a bounded channel using `FullMode = Wait` (the default) values are **silently dropped** when the buffer is full. Use a drop mode, or a custom `onNext` via `FromChannel`, to get different semantics. +- The `IObservable` grammar has no resumable-error channel: `OnError` is mapped to a failure completion (`Result.Failure`), terminating the stream. + +### AsyncObservable → IObservable + +`ToSystemObservable` converts an `AsyncObservable` into a `System.IObservable`. `IObservable`'s `Subscribe`/`Dispose` are synchronous while R3Async's are not, so you must decide how each async operation is consumed from the sync world via an `AsyncToSyncStrategy`: + +```csharp +var observable = asyncObservable.ToSystemObservable(new ToObservableConfiguration +{ + // Blocking: block the caller until the async operation completes (exceptions propagate) + SubscribeStrategy = AsyncToSyncStrategy.Blocking, + + // FireAndForget: start the operation without waiting; failures go to the + // optional onException callback (or the UnhandledExceptionHandler) + DisposeStrategy = AsyncToSyncStrategy.FireAndForget(onException: ex => Console.WriteLine(ex)) +}); +``` + +Since `IObserver` has no resumable-error channel, an `OnErrorResume` notification from the source is delivered as a terminal `OnError` and the subscription is torn down. A failure completion is likewise delivered as `OnError`; a success completion as `OnCompleted`. + ## R3 Interop The [R3Async.R3Interop](https://www.nuget.org/packages/R3Async.R3Interop) package bridges R3Async with [R3](https://github.com/Cysharp/R3), converting between the synchronous `Observable` and the asynchronous `AsyncObservable` in both directions. @@ -897,6 +946,8 @@ The [R3Async.R3Interop](https://www.nuget.org/packages/R3Async.R3Interop) packag using R3Async.R3Interop; ``` +The configuration types (`BackpressureStrategy`, `AsyncToSyncStrategy`, `ToObservableConfiguration`) live in the core `R3Async` package and are shared with the `System.IObservable` interop described above; the conversions work the same way in both. + ### Observable → AsyncObservable `ToAsyncObservable` converts an R3 observable into an R3Async one. Since an R3 source pushes values synchronously while the async observer may be slow, you must choose a `BackpressureStrategy`: