Skip to content

Dotnet reactive interop #17

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
fedeAlterio merged 2 commits into from
Jul 19, 2026
Merged

Dotnet reactive interop #17

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d360778
Dotnet reactive interop
fedeAlterio Jul 19, 2026
0fb8842
updated readme
fedeAlterio Jul 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 51 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<T>` and `AsyncObservable<T>` in both directions, with no extra package required. Since `IObservable<T>` is synchronous while `AsyncObservable<T>` is not, both directions require you to state explicitly how the sync/async boundary is handled.

### IObservable<T> → AsyncObservable<T>

`ToAsyncObservable` converts a `System.IObservable<T>` into an `AsyncObservable<T>`. 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<int>(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<T>` grammar has no resumable-error channel: `OnError` is mapped to a failure completion (`Result.Failure`), terminating the stream.

### AsyncObservable<T> → IObservable<T>

`ToSystemObservable` converts an `AsyncObservable<T>` into a `System.IObservable<T>`. `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<T>` 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<T>` and the asynchronous `AsyncObservable<T>` in both directions.
Expand All @@ -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<T> → AsyncObservable<T>

`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`:
Expand Down
119 changes: 0 additions & 119 deletions src/R3Async.R3Interop/ToAsyncObservable.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<T> FromUnboundedChannel<T>(Action<Exception, ChannelWriter<T>> 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<T> FromBoundedChannel<T>(Action<Exception, ChannelWriter<T>> onErrorResume, int capacity)
{
return FromBoundedChannel(onErrorResume, new BoundedChannelOptions(capacity));
}

public static ChannelBackpressureStrategy<T> FromBoundedChannel<T>(Action<Exception, ChannelWriter<T>> 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<T> FromChannel<T>(Func<Channel<T>> channelFactory,
Action<T, ChannelWriter<T>>? onNext = null,
Action<Exception, ChannelWriter<T>>? 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<T> ToChannelStrategy<T>() => ToChannelStrategy<T>(options, null);

internal static ChannelBackpressureStrategy<T> ToChannelStrategy<T>(UnboundedChannelOptions? options,
Action<Exception, ChannelWriter<T>>? onErrorResume)
{
return BackpressureStrategy.FromChannel(() => options is null ? Channel.CreateUnbounded<T>() : Channel.CreateUnbounded<T>(options),
static (x, c) => c.TryWrite(x),
onErrorResume);
}
}

public sealed class BoundedChannelBackpressureStrategy
{
readonly BoundedChannelOptions options;

internal BoundedChannelBackpressureStrategy(BoundedChannelOptions options)
{
this.options = options;
}

internal ChannelBackpressureStrategy<T> ToChannelStrategy<T>() => ToChannelStrategy<T>(options, null);

internal static ChannelBackpressureStrategy<T> ToChannelStrategy<T>(BoundedChannelOptions options,
Action<Exception, ChannelWriter<T>>? onErrorResume)
{
return BackpressureStrategy.FromChannel(() => Channel.CreateBounded<T>(options),
onErrorResume: onErrorResume);
}
}

public sealed class ChannelBackpressureStrategy<T>
{
internal ChannelBackpressureStrategy(Func<Channel<T>> channelFactory,
Action<T, ChannelWriter<T>> onNext,
Action<Exception, ChannelWriter<T>>? onErrorResume)
{
ChannelFactory = channelFactory;
OnNext = onNext;
OnErrorResume = onErrorResume;
}

internal Func<Channel<T>> ChannelFactory { get; }
internal Action<T, ChannelWriter<T>> OnNext { get; }
internal Action<Exception, ChannelWriter<T>>? OnErrorResume { get; }
}

public static class R3ToAsyncObservableExtensions
{
/// <summary>
Expand Down
113 changes: 0 additions & 113 deletions src/R3Async.R3Interop/ToObservable.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,67 +5,6 @@

namespace R3Async.R3Interop;

public sealed class AsyncToSyncStrategy
{
readonly Action<Exception>? _onException;

private AsyncToSyncStrategy(Action<Exception>? onException) => _onException = onException;

static readonly AsyncToSyncStrategy DefaultFireAndForget = new(null);
public static AsyncToSyncStrategy Blocking { get; } = new(null);

public static AsyncToSyncStrategy FireAndForget(Action<Exception>? 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
{
/// <summary>
Expand Down Expand Up @@ -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<IAsyncDisposable> 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();
}
}
}
}
Loading
Loading