dotnetfoundation/System.Reactivev6.0.0
✓ Verified.NET FW4.7.2.NET 8.0.NET 8.0-windows10.0.19041.NET Std2.0
Reactive Extensions (Rx) for .NET
Get Started
$ dotnet add package System.ReactiveReadme
Rx (Reactive Extensions for .NET)
Rx enables event-driven programming with a composable, declarative model.
Getting started
Run the following at a command line:
mkdir TryRx
cd TryRx
dotnet new console
dotnet add package System.Reactive
Alternatively, if you have Visual Studio installed, create a new .NET Console project, and then use the NuGet package manager to add a reference to System.Reactive.
You can then add this code to your Program.cs. This creates an observable source (ticks) that produces an event once every second. It also adds a handler to that source that writes a message to the console for each event:
using System.Reactive.Linq;
IObservable<long> ticks = Observable.Timer(
dueTime: TimeSpan.Zero,
period: TimeSpan.FromSeconds(1));
ticks.Subscribe(
tick => Console.WriteLine($"Tick {tick}"));
Console.ReadLine();
Examples
Wrapping an existing event source as an Rx IObservable<T>
If you have an existing source of events that does not support Rx directly, but which does offer .NET events, you can bring this into the world of Rx using the Observable.FromEventPattern method:
using System.Reactive.Linq;
FileSystemWatcher fsw = new FileSystemWatcher(@"C:\temp");
IObservable<FileSystemEventArgs> changeEvents = Observable
.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
h => fsw.Changed += h,
h => fsw.Changed -= h)
.Select(e => e.EventArgs);
fsw.EnableRaisingEvents = true;
Waiting for inactivity
It can sometimes be useful to wait for a period of inactivity before taking action. For example, if you have code that monitors a directory in a filesystem, processing modified or added files, it's common for there to be flurries of activity. For example, if a user is copying multiple files into a folder that you're observing, there will be multiple changes, and it could be more efficient to wait until those stop and then process all the changes in one batch, than to attempt to process everything immediately.
This example defines a custom Rx operator that can be attached to any source. It will wait for that source to start producing events, and then, it will wait for it to stop again for the specified period. Each time that happens, it reports all of the activity that occurred between the last two periods of inactivity: