Skip to content

Add support for System.Threading.Lock #2254

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

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Add support for System.Threading.Lock #2254

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
98adf33
Config and filename adjustments, to allow for building on Linux.
JakenVeina Oct 11, 2025
2709981
Added a new overload for `.Synchronize()` utilizing System.Threading.…
JakenVeina Nov 1, 2025
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
0 AsyncRx.NET/Directory.build.props → AsyncRx.NET/Directory.Build.props
View file
Open in desktop
File renamed without changes.
8 changes: 4 additions & 4 deletions Ix.NET/Documentation/adr/0001-Ix-Ref-Assembly-Mismatches.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ At the time of writing this, the current version of `System.Interactive` is 6.0.
* `net6.0`
* `netstandard2.1`


The use of `net4.8` in `ref` seems to have been a bug: that should have been `net48`. (The main reason I am confident it's a bug, and not a clever but obscure trick that we've not understood, is that the [commit of 2021/12/06 that added this](https://github.com/dotnet/reactive/commit/a2410b2267abe193191f3894d243771ae4b126fd) used [`net48` in reference assemblies for one of the other packages](https://github.com/dotnet/reactive/commit/a2410b2267abe193191f3894d243771ae4b126fd#diff-3b568c93a468dab1b1a619a450bf1c4d88d3ec9539737d09fa6fb7659bc0ae5fR7), so this just seems to have been a slip.)

The other discrepancy is that we have `netstandard2.0` in the `lib` folder but `netstandard2.1` in the ref folder. At first glance, this too looks quite a lot like a mistake, particularly when you examine the history. Here is the point in the release history at which the `ref` folder first started having a `netstandard2.1` folder:
Expand All @@ -45,12 +45,12 @@ And yet, on closer inspection, this appears to be deliberate. Looking at this co

https://github.com/dotnet/reactive/commit/0252fb537c9d335b9bc863b65291f152c07ba385

we see a [comment in Ix.NET/Source/refs/Directory.build.props](https://github.com/dotnet/reactive/commit/0252fb537c9d335b9bc863b65291f152c07ba385#diff-909504334cbab5c432709c95ae78c24fb2910d850958af2ef6de444b18e5c8ecR6) saying:
we see a [comment in Ix.NET/Source/refs/Directory.Build.props](https://github.com/dotnet/reactive/commit/0252fb537c9d335b9bc863b65291f152c07ba385#diff-909504334cbab5c432709c95ae78c24fb2910d850958af2ef6de444b18e5c8ecR6) saying:

> This is here so we can create a fake .NET Standard 2.1 facade

I can only guess that they knew .NET Standard 2.1 was coming, and wanted to ensure that `System.Interactive` was ready for it when it shipped.

So it was deliberate. But offering reference assemblies for a platform without any corresponding implementation for that platform is an odd choice. (And although at the time this was a placholder for a forthcoming .NET Standard version, it continued to look like this after .NET Standard 2.1 shipped. All subsequent Ix.NET releases have continued to provide `netstandard2.1` in the `ref` folder with no matching folder in `lib`. So it wasn't just a temporary measure.) What purpose does this serve?

Some of the features that Ix offers eventually became available in .NET Core, such as `EnumerableEx.SkipLast`. This method exists in the implementation assemblies for every TFM of Ix.NET, but the `netstandard2.1` and `net6.0` reference assemblies omit it. This has the effect that if you're targetting any version of .NET recent enough to have these methods built into the .NET runtime libraries, the Ix.NET equivalents will:
Expand Down
4 changes: 2 additions & 2 deletions Ix.NET/Source/Ix.Async.NET.sln
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,8 +11,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
AsyncQueryableGenerator.t4 = AsyncQueryableGenerator.t4
..\..\azure-pipelines.ix.yml = ..\..\azure-pipelines.ix.yml
CodeCoverage.runsettings = CodeCoverage.runsettings
Directory.build.props = Directory.build.props
Directory.build.targets = Directory.build.targets
Directory.Build.props = Directory.Build.props
Directory.Build.targets = Directory.Build.targets
global.json = global.json
NuGet.Config = NuGet.Config
version.json = version.json
Expand Down
6 changes: 3 additions & 3 deletions Ix.NET/Source/Ix.NET.sln
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
..\..\.editorconfig = ..\..\.editorconfig
..\..\azure-pipelines.ix.yml = ..\..\azure-pipelines.ix.yml
CodeCoverage.runsettings = CodeCoverage.runsettings
Directory.build.props = Directory.build.props
Directory.build.targets = Directory.build.targets
Directory.Build.props = Directory.Build.props
Directory.Build.targets = Directory.Build.targets
version.json = version.json
EndProjectSection
EndProject
Expand Down Expand Up @@ -61,7 +61,7 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "FasterLinq", "FasterLinq\Fa
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Refs", "Refs", "{A3D72E6E-4ADA-42E0-8B2A-055B1F244281}"
ProjectSection(SolutionItems) = preProject
refs\Directory.build.props = refs\Directory.build.props
refs\Directory.Build.props = refs\Directory.Build.props
EndProjectSection
EndProject
Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "System.Interactive", "refs\System.Interactive\System.Interactive.csproj", "{2EC0C302-B029-4DDB-AC91-000BF11006AD}"
Expand Down
4 changes: 2 additions & 2 deletions Rx.NET/Documentation/adr/0001-net7.0-era-tooling-update.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,9 +74,9 @@ The following sections describe how we will deal with each of the issues raised

In all projects that used to target `net5.0`, change this to `net6.0`. Likewise, in any projects that target `net5.0-windows10.0.19041`, change that to `net6.0-windows10.0.19041`.

Modify the entries in `Directory.build.targets`
Modify the entries in `Directory.Build.targets`

Remove the entries in `Directory.build.targets` that refer to the out-of-support TFMs. This means that several preprocessor constants are no longer used. We should scour the codebase and remove all conditionally compiled sections of code that will no longer be used because the target frameworks that used to bring them in no longer exist. The constants no longer in use are:
Remove the entries in `Directory.Build.targets` that refer to the out-of-support TFMs. This means that several preprocessor constants are no longer used. We should scour the codebase and remove all conditionally compiled sections of code that will no longer be used because the target frameworks that used to bring them in no longer exist. The constants no longer in use are:

* `NETSTANDARD1_0`
* `NETSTANDARD1_3`
Expand Down
12 changes: 6 additions & 6 deletions Rx.NET/Documentation/adr/0003-uap-targets.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,14 +99,14 @@ Note that the specific version we actually want to target, 18362, isn't in there

The following sections explain how we enable `uap10.0.18362` to be specified as a target framework, even though the tools do not support this.

The project has `Directory.build.props` and `Directory.build.targets` files. The build tools search for these and automatically load them for all projects in the solution. The `Directory.build.props` file has a `<PropertyGroup>` with a `Condition` that means it runs only when the `uap10.0.18362` target is being built, and it sets numerous properties, as described in the following sections.
The project has `Directory.Build.props` and `Directory.Build.targets` files. The build tools search for these and automatically load them for all projects in the solution. The `Directory.Build.props` file has a `<PropertyGroup>` with a `Condition` that means it runs only when the `uap10.0.18362` target is being built, and it sets numerous properties, as described in the following sections.


#### Target Platform Version

We make our minmum platform version match the one in the TFM:

```xml
```xml
<TargetPlatformMinVersion>10.0.18362</TargetPlatformMinVersion>
<TargetPlatformVersion>10.0.18362.0</TargetPlatformVersion>
```
Expand All @@ -130,7 +130,7 @@ However, only _some_ properties should use the old name. We need to set _all_ of

#### Compiler Constants

When using the supported UWP build tools (with the old-form project system, which we can't use because we also need to build modern targets), the `WINDOWS_UWP` define constant is set, enabling source code compiled into multiple targets to detect that it is being built for UWP with a `#if WINDOWS_UWP`. So we need this in `Directory.build.props`:
When using the supported UWP build tools (with the old-form project system, which we can't use because we also need to build modern targets), the `WINDOWS_UWP` define constant is set, enabling source code compiled into multiple targets to detect that it is being built for UWP with a `#if WINDOWS_UWP`. So we need this in `Directory.Build.props`:

```xml
<DefineConstants>$(DefineConstants);WINDOWS_UWP</DefineConstants>
Expand All @@ -141,15 +141,15 @@ When using the supported UWP build tools (with the old-form project system, whic

Normally, when you specify a TFM, the .NET SDK works out what framework library references are required and adds them for you. So if you write `<TargetFramework>net8.0<TargetFramework>` in a project file, you will automatically have access to all the .NET 8.0 runtime libraries. But because the .NET SDK does not support UWP, this doesn't work at all. So we need to do three things.

First, we need to set this property in `Directory.build.props`:
First, we need to set this property in `Directory.Build.props`:

```xml
<NoStdLib>True</NoStdLib>
```

Without this, the build tools attempt to add a reference to `mscorlib.dll`, but they don't seem to realise that a) this is the wrong thing and b) they don't actually have a correct location for that, so the reference ends up being `\mscorlib.dll` (i.e., it looks on the root of the hard drive).

Second we need an `ItemGroup` in `Directory.build.targets` containing this:
Second we need an `ItemGroup` in `Directory.bBild.targets` containing this:

```xml
<PackageReference Include="Microsoft.NETCore.UniversalWindowsPlatform"
Expand All @@ -171,7 +171,7 @@ You might be wondering about that 26100 in there. Why is that not 18362, consist

#### Prevent Over-Zealous WinRT Interop Code Generation

The .NET SDK has a feature by which it can generate WinRT versions of .NET types to enable interop between .NET and WinRT code. Unfortunately, the way we've rigged things up to be able to build for `uap10.0.18362.0` seems to cause this to generate these interop types for any .NET class that implements `IDisposable`! This is not helpful. So we disable the feature in `Directory.build.targets`:
The .NET SDK has a feature by which it can generate WinRT versions of .NET types to enable interop between .NET and WinRT code. Unfortunately, the way we've rigged things up to be able to build for `uap10.0.18362.0` seems to cause this to generate these interop types for any .NET class that implements `IDisposable`! This is not helpful. So we disable the feature in `Directory.Build.targets`:

```xml
<CsWinRTAotOptimizerEnabled>false</CsWinRTAotOptimizerEnabled>
Expand Down
5 changes: 3 additions & 2 deletions Rx.NET/Source/Directory.build.props → Rx.NET/Source/Directory.Build.props
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,11 +18,12 @@
<EmbedUntrackedSources>true</EmbedUntrackedSources>
<PublishRepositoryUrl>true</PublishRepositoryUrl>
<LangVersion>latest</LangVersion>
<EnableWindowsTargeting>true</EnableWindowsTargeting>
</PropertyGroup>

<PropertyGroup Condition="'$(Configuration)' != 'Debug'">
<IncludeSymbols>true</IncludeSymbols>
<SymbolPackageFormat>snupkg</SymbolPackageFormat>
<SymbolPackageFormat>snupkg</SymbolPackageFormat>
</PropertyGroup>

<PropertyGroup Condition="'$(TF_BUILD)' == 'true'">
Expand All @@ -31,7 +32,7 @@

<ItemGroup>
<PackageReference Include="Microsoft.SourceLink.GitHub" Version="1.1.1" PrivateAssets="All"/>

<!--
Nerdbank.GitVersioning 3.6.128 injects a reference to a .proj file that doesn't work inside the
UWP test runner project. We don't ship that as a NuGet package, so it doesn't matter what its
Expand Down
11 changes: 7 additions & 4 deletions Rx.NET/Source/Directory.build.targets → Rx.NET/Source/Directory.Build.targets
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,11 @@
<Project>

<!-- This props all need to be set in targets as they depend on the values set earlier -->
<PropertyGroup>

<PropertyGroup>
<Product>$(AssemblyName) ($(TargetFramework))</Product>
</PropertyGroup>

<PropertyGroup Condition="'$(TargetFramework)' == 'net472'">
<DefineConstants>$(DefineConstants);HAS_WINFORMS;HAS_WPF;HAS_WINRT;HAS_DISPATCHER;HAS_REMOTING;DESKTOPCLR;NO_NULLABLE_ATTRIBUTES</DefineConstants>
</PropertyGroup>
Expand All @@ -18,9 +18,12 @@
<PropertyGroup Condition="$(TargetFramework.StartsWith('net6.0')) or $(TargetFramework.StartsWith('net7.0')) or $(TargetFramework.StartsWith('net8.0'))">
<DefineConstants>$(DefineConstants);HAS_TRIMMABILITY_ATTRIBUTES</DefineConstants>
</PropertyGroup>
<PropertyGroup Condition="$(TargetFramework.StartsWith('net6.0-windows')) or $(TargetFramework.StartsWith('net8.0-windows')) or $(TargetFramework.StartsWith('net9.0-windows'))">
<PropertyGroup Condition="$(TargetFramework.StartsWith('net6.0-windows'))">
<DefineConstants>$(DefineConstants);HAS_WINRT;HAS_WINFORMS;HAS_WPF;HAS_DISPATCHER;DESKTOPCLR;WINDOWS;CSWINRT</DefineConstants>
</PropertyGroup>
<PropertyGroup Condition="$(TargetFramework.StartsWith('net9.0'))">
<DefineConstants>$(DefineConstants);HAS_SYSTEM_THREADING_LOCK</DefineConstants>
</PropertyGroup>

<ItemGroup Condition="('$(TargetFramework)' == 'net472' or '$(TargetFramework)' == 'uap10.0.18362' or '$(TargetFramework)' == 'netstandard2.0') and $(IsPackable)">
<PackageReference Include="System.Threading.Tasks.Extensions" Version="4.5.4" />
Expand Down
10 changes: 5 additions & 5 deletions Rx.NET/Source/System.Reactive.sln
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,8 +19,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution
.editorconfig = .editorconfig
analyzers.globalconfig = analyzers.globalconfig
..\..\azure-pipelines.rx.yml = ..\..\azure-pipelines.rx.yml
Directory.build.props = Directory.build.props
Directory.build.targets = Directory.build.targets
Directory.Build.props = Directory.Build.props
Directory.Build.targets = Directory.Build.targets
global.json = global.json
Rx.ruleset = Rx.ruleset
Test.ruleset = Test.ruleset
Expand All @@ -30,13 +30,13 @@ EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Tests", "Tests", "{D324579D-CBE6-4867-8980-D7842C7C45A2}"
ProjectSection(SolutionItems) = preProject
tests\.editorconfig = tests\.editorconfig
tests\Directory.build.props = tests\Directory.build.props
tests\Directory.build.targets = tests\Directory.build.targets
tests\Directory.Build.props = tests\Directory.Build.props
tests\Directory.Build.targets = tests\Directory.Build.targets
EndProjectSection
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Facades", "Facades", "{A0F39260-B8F8-4FCB-9679-0ED917A22BDF}"
ProjectSection(SolutionItems) = preProject
facades\Directory.build.props = facades\Directory.build.props
facades\Directory.Build.props = facades\Directory.Build.props
facades\System.Reactive.Compatibility.nuspec = facades\System.Reactive.Compatibility.nuspec
EndProjectSection
EndProject
Expand Down
2 changes: 1 addition & 1 deletion Rx.NET/Source/facades/Directory.build.props → Rx.NET/Source/facades/Directory.Build.props
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
<Project>
<Import Project="..\Directory.build.props" />
<Import Project="..\Directory.Build.props" />
<PropertyGroup>
<GenerateAssemblyVersionAttribute>false</GenerateAssemblyVersionAttribute>
<GenerateAssemblyVersionInfo>false</GenerateAssemblyVersionInfo>
Expand Down
3 changes: 3 additions & 0 deletions Rx.NET/Source/facades/Directory.Build.targets
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
<Project>
<Import Project="..\Directory.Build.targets" />
</Project>
3 changes: 0 additions & 3 deletions Rx.NET/Source/facades/Directory.build.targets
View file
Open in desktop

This file was deleted.

3 changes: 3 additions & 0 deletions Rx.NET/Source/src/Directory.Build.props
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
<Project>
<Import Project="..\Directory.Build.props" />
</Project>
3 changes: 3 additions & 0 deletions Rx.NET/Source/src/Directory.Build.targets
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
<Project>
<Import Project="..\Directory.Build.targets" />
</Project>
3 changes: 0 additions & 3 deletions Rx.NET/Source/src/Directory.build.props
View file
Open in desktop

This file was deleted.

3 changes: 0 additions & 3 deletions Rx.NET/Source/src/Directory.build.targets
View file
Open in desktop

This file was deleted.

64 changes: 59 additions & 5 deletions Rx.NET/Source/src/System.Reactive/Concurrency/Synchronization.Synchronize.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,21 +1,22 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT License.
// See the LICENSE file in the project root for more information.
// See the LICENSE file in the project root for more information.
using System.Threading;

namespace System.Reactive.Concurrency
{
internal sealed class Synchronize<TSource> : Producer<TSource, Synchronize<TSource>._>
internal sealed class SynchronizeWithObject<TSource> : Producer<TSource, SynchronizeWithObject<TSource>._>
{
private readonly IObservable<TSource> _source;
private readonly object? _gate;

public Synchronize(IObservable<TSource> source, object gate)
public SynchronizeWithObject(IObservable<TSource> source, object gate)
{
_source = source;
_gate = gate;
}

public Synchronize(IObservable<TSource> source)
public SynchronizeWithObject(IObservable<TSource> source)
{
_source = source;
}
Expand All @@ -28,7 +29,7 @@ internal sealed class _ : IdentitySink<TSource>
{
private readonly object _gate;

public _(Synchronize<TSource> parent, IObserver<TSource> observer)
public _(SynchronizeWithObject<TSource> parent, IObserver<TSource> observer)
: base(observer)
{
_gate = parent._gate ?? new object();
Expand Down Expand Up @@ -59,4 +60,57 @@ public override void OnCompleted()
}
}
}

#if HAS_SYSTEM_THREADING_LOCK
internal sealed class SynchronizeWithLock<TSource> : Producer<TSource, SynchronizeWithLock<TSource>._>
{
private readonly IObservable<TSource> _source;
private readonly Lock _gate;

public SynchronizeWithLock(IObservable<TSource> source, Lock gate)
{
_source = source;
_gate = gate;
}

protected override _ CreateSink(IObserver<TSource> observer) => new(this, observer);

protected override void Run(_ sink) => sink.Run(_source);

internal sealed class _ : IdentitySink<TSource>
{
private readonly Lock _gate;

public _(SynchronizeWithLock<TSource> parent, IObserver<TSource> observer)
: base(observer)
{
_gate = parent._gate;
}

public override void OnNext(TSource value)
{
lock (_gate)
{
ForwardOnNext(value);
}
}

public override void OnError(Exception error)
{
lock (_gate)
{
ForwardOnError(error);
}
}

public override void OnCompleted()
{
lock (_gate)
{
ForwardOnCompleted();
}
}
}
}
#endif
}
31 changes: 28 additions & 3 deletions Rx.NET/Source/src/System.Reactive/Concurrency/Synchronization.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT License.
// See the LICENSE file in the project root for more information.
// See the LICENSE file in the project root for more information.

using System.ComponentModel;
using System.Reactive.Disposables;
Expand Down Expand Up @@ -229,7 +229,7 @@ public static IObservable<TSource> Synchronize<TSource>(IObservable<TSource> sou
throw new ArgumentNullException(nameof(source));
}

return new Synchronize<TSource>(source);
return new SynchronizeWithObject<TSource>(source);
}

/// <summary>
Expand All @@ -252,9 +252,34 @@ public static IObservable<TSource> Synchronize<TSource>(IObservable<TSource> sou
throw new ArgumentNullException(nameof(gate));
}

return new Synchronize<TSource>(source, gate);
return new SynchronizeWithObject<TSource>(source, gate);
}

#if HAS_SYSTEM_THREADING_LOCK
/// <summary>
/// Wraps the source sequence in order to ensure observer callbacks are synchronized using the specified gate object.
/// </summary>
/// <typeparam name="TSource">The type of the elements in the source sequence.</typeparam>
/// <param name="source">Source sequence.</param>
/// <param name="gate">Gate object to synchronize each observer call on.</param>
/// <returns>The source sequence whose outgoing calls to observers are synchronized on the given gate object.</returns>
/// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="gate"/> is <c>null</c>.</exception>
public static IObservable<TSource> Synchronize<TSource>(IObservable<TSource> source, Lock gate)
Copy link
Collaborator

@idg10 idg10 Oct 15, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is not the only additional public method required. Unless I've missed it, you've not added an equivalent extra overload in the Observable.Concurrency.cs type. (And you might also need to add an equivalent method to IQueryLanguage.)

Oh you'll also need to rerun the homoiconicity project to generate the IQbservable<T> forms of this, which I don't think you've done?

Copy link
Author

@JakenVeina JakenVeina Oct 16, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I forgot about the whole IQueryLanguage layer, yeah. I've debugged through it before, so I'm familiar with how it works, but I've never tried to implement anything.

I have basically no knowledge of anything related to IQbservable<T>, but I'll look into it.

Copy link
Collaborator

@idg10 idg10 Oct 16, 2025 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The basic idea behind IQbservable<T> is that it supports exactly the same API as IObservable<T>, but the output is effectively a description of whatever query you've written. So if you do this:

IObservable<int> src = GetSomeObservable();

IObservable<int> xs = src
    .Where(x => x > 0)
    .Select(x => x * 2);

then xs is a thing you can actually subscribe to that removes negative numbers, and then doubles everything else. But if we write the very similar:

IQbservable<int> xs = src
    .AsQbservable()
    .Where(x => x > 0)
    .Select(x => x * 2);

Console.WriteLine(xs.Expression);

then we've basically got the same query but this time as an IQbservable<int>, and that means that this is a description of the observable source. Instead of just being a thing we can subscribe to, we can inspect this. That Console.WriteLine(xs.Expression); displays this:

System.Reactive.Linq.ObservableImpl.RangeRecursive.Where(x => (x > 0)).Select(x => (x * 2))

(The RangeRecursive there comes from the fact that in the example I'm testing this in, my GetSomeObservable() is returning Observable.Range(0, 10);. So that's really just the type of src here.)

So you can see that this thing knows that this is a Where clause followed by a Select clause. And if you were to inspect xs in the debugger, you'll see a DebugView property that looks like this:

.Call System.Reactive.Linq.Qbservable.Select(
    .Call System.Reactive.Linq.Qbservable.Where(
        .Constant<System.Reactive.ObservableQuery`1[System.Int32]>(System.Reactive.Linq.ObservableImpl.RangeRecursive),
        '(.Lambda #Lambda1<System.Func`2[System.Int32,System.Boolean]>)),
    '(.Lambda #Lambda2<System.Func`2[System.Int32,System.Int32]>))

.Lambda #Lambda1<System.Func`2[System.Int32,System.Boolean]>(System.Int32 $x) {
    $x > 0
}

.Lambda #Lambda2<System.Func`2[System.Int32,System.Int32]>(System.Int32 $x) {
    $x * 2
}

So the basic idea here is that an IQbservable<T> is a complete description of the subscription, one that can be inspected at runtime.

IQbservable<T> is to IObservable<T> as IQueryable<T> is to IEnumerable<T>. Systems like Entity Framework exploit the fact that an IQueryable<T> remembers exactly how the query was constructed—just like with the IQbservable<T> I've shown here, an IQueryable<T> would remember that it was built up as (say) a Where and a Select, and EF would then use that information to work out what SQL query to generate to be able to execute the logic that the query represents on a database.

The idea behind IQueryable<T> is that it could be used to support similar mechanisms. You could imagine a remote monitoring device presenting an IQueryable<T> API, and if that remote device were able to support local filtering, you could imagine this implementation of IQueryable<T> detecting when your query starts with a Where clause, and translating that into whatever format the remote device uses to express the filtering (in exactly the same way that Entity Framework translates a LINQ Where into a SQL WHERE clause.)

Admittedly, I'm not aware of any public libraries that actually do that. But it's a model we support, and it's also the basis of distributed query execution in Reaqtor (https://reaqtive.net/). So there are automated tests that check that the public API we define for IObservable<T> is fully matched by the one available for IQueryable<T>. And we have a tool in the repo that auto-generates the necessary code (the Homoiconicity tool). Although the tool is a bit cranky, and currently I have to manually fix the code it generates...you end up running it and then discarding most of what it changed, and keeping just the new bits. At some point I need to fix the tool so that we don't have to do that every time we add a new API feature.

Copy link
Author

@JakenVeina JakenVeina Oct 17, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IQbservable<T> is to IObservable<T> as IQueryable<T> is to IEnumerable<T>.

I suspected it might be something like that. And you're saying there's no actual implementations here, just the modeling. I can work with that.

{
if (source == null)
{
throw new ArgumentNullException(nameof(source));
}

if (gate == null)
{
throw new ArgumentNullException(nameof(gate));
}

return new SynchronizeWithLock<TSource>(source, gate);
}
#endif

#endregion
}
}
Loading
Loading