Querying Event Streams with Reactive Extensions in C

Reactive Extensions (Rx.NET) enables developers to treat asynchronous event streams as push-based data collections that can be queried and transformed using LINQ-style syntax. This article covers how to install Rx.NET, convert standard .NET events and asynchronous operations into observable sequences, apply filtering and time-based operators, and handle stream consumption safely through subscriptions.

Getting Started with Rx.NET

To begin using Reactive Extensions in C#, install the official NuGet package:

dotnet add package System.Reactive

Rx centers around two primary interfaces defined in the System namespace:

• IObservable<T>: Represents a provider for push-based notification (the event producer).
• IObserver<T>: Receives notifications from an observable (the event consumer).

Creating Observable Sequences

You can create observables from static values, time-based intervals, or existing standard .NET events.

From Standard .NET Events

To transform traditional UI clicks, file watchers, or custom events into an observable stream, use Observable.FromEventPattern:

using System;
using System.Reactive.Linq;

public class Button
{
    public event EventHandler Clicked;
    public void Click() => Clicked?.Invoke(this, EventArgs.Empty);
}

// Convert the event to an IObservable sequence
Button button = new Button();

IObservable<System.Reactive.EventPattern<EventArgs>> clicks = 
    Observable.FromEventPattern<EventArgs>(
        handler => button.Clicked += handler,
        handler => button.Clicked -= handler);

From Timers or Periodic Operations

You can generate continuous streams using built-in generators like Observable.Interval:

// Emits an incrementing long value every 500 milliseconds
IObservable<long> ticks = Observable.Interval(TimeSpan.FromMilliseconds(500));

Querying Event Streams with LINQ

Because IObservable<T> supports LINQ, you can manipulate asynchronous streams using standard operators such as Where, Select, and Take, as well as time-aware operators unique to Rx.

Filtering and Projection

var numbers = Observable.Range(1, 20);

IObservable<string> evenSquares = numbers
    .Where(x => x % 2 == 0)
    .Select(x => $"Square: {x * x}");

Time-Based Operators

Reactive Extensions excels at handling continuous real-time inputs, such as user keystrokes or sensor data, using time-based operators:

• Throttle (Debounce): Emits an item only after a specified timespan has passed without any new items. Useful for search input fields.
• Buffer: Groups items into batches based on time intervals or count limits.
• Sample: Takes the most recent item emitted over periodic intervals.

// Debouncing rapid input (e.g., search text)
IObservable<string> searchInputs = GetSearchInputStream();

IObservable<string> debouncedQueries = searchInputs
    .Throttle(TimeSpan.FromMilliseconds(300))
    .DistinctUntilChanged();

Consuming the Stream with Subscriptions

To listen to the results of your query, call the Subscribe method on the observable. Subscribe provides callbacks for values emitted (OnNext), errors encountered (OnError), and stream completion (OnCompleted).

IDisposable subscription = debouncedQueries.Subscribe(
    query => Console.WriteLine($"Execute search for: {query}"),
    error => Console.WriteLine($"Error occurred: {error.Message}"),
    () => Console.WriteLine("Query stream completed.")
);

// To stop listening and clean up event handlers:
subscription.Dispose();

Managing Concurrency with Schedulers

Rx uses IScheduler to control threading contexts for observing and subscribing.

• SubscribeOn: Dictates the thread on which the observable begins its work.
• ObserveOn: Dictates the thread on which observer callbacks (OnNext) are executed.

using System.Reactive.Concurrency;

searchInputs
    .ObserveOn(TaskPoolScheduler.Default)      // Process background computations on TaskPool
    .Select(query => HeavySearchOperation(query))
    .ObserveOn(Scheduler.CurrentThread)         // Return results to the main or UI thread
    .Subscribe(results => DisplayResults(results));

By combining observable creation, LINQ transformation pipelines, and thread scheduling, Reactive Extensions provides a declarative and robust framework for handling complex, asynchronous event architectures in C#.