Reactive Extensions
======================
This repository contains four libraries which are conceptually related in that they are all concerned with LINQ over sequences of things:
* [Reactive Extensions for .NET](Rx.NET/) aka Rx.NET or Rx ([System.Reactive](https://www.nuget.org/packages/System.Reactive/)): a library for event-driven programming with a composable, declarative model
* [AsyncRx.NET](AsyncRx.NET/) (experimental preview) ([System.Reactive.Async](https://www.nuget.org/packages/System.Reactive.Async)): experimental implementation of Rx for `IAsyncObservable` offering deeper `async`/`await` support
* [Interactive Extensions for .NET](Ix.NET/), aka Ix ([System.Interactive](https://www.nuget.org/packages/System.Interactive/)): extended LINQ operators for `IAsyncEnumerable` and `IEnumerable`
* [LINQ for `IAsyncEnumerable`](./Ix.NET/Source/System.Linq.Async/) ([System.Linq.Async](https://www.nuget.org/packages/System.Linq.Async/)): implements standard LINQ operators for `IAsyncEnumerable`
Each will be described later in this README.
## FREE Introduction to Rx.NET 2nd Edition eBook
Reactive programming provides clarity when our code needs to respond to events. The Rx.NET libraries were designed to enable cloud-native applications to process live data in reliable, predictable ways.
We've written a FREE book which explains the vital abstractions that underpin Rx, and shows how to exploit the powerful and extensive functionality built into the Rx.NET libraries.
Based on Lee Campbell's 2010 book (kindly donated to the project), it has been re-written to bring it up to date with Rx.NET v6.0, .NET 8.0, and modern cloud native use cases such as IoT and real-time stream data processing.
Introduction to Rx.NET is available [Online](https://introtorx.com/), [on GitHub](Rx.NET/Documentation/IntroToRx/), as [PDF](https://endjincdn.blob.core.windows.net/assets/ebooks/introduction-to-rx-dotnet/introduction-to-rx-dotnet-2nd-edition.pdf), and [EPUB](https://endjincdn.blob.core.windows.net/assets/ebooks/introduction-to-rx-dotnet/introduction-to-rx-dotnet-2nd-edition.epub).
## Getting the bits
Channel | Rx | AsyncRx | Ix | System.Linq.Async
--- | --- | --- | --- |--- |
NuGet.org | [](https://www.nuget.org/packages/System.Reactive/)| [](https://www.nuget.org/packages/System.Reactive.Async/) | [](https://www.nuget.org/packages/System.Interactive/) | [](https://www.nuget.org/packages/System.Linq.Async/)
NuGet.org preview (if newer than release) | [](https://www.nuget.org/packages/System.Reactive/) | [](https://www.nuget.org/packages/System.Reactive.Async/) | [](https://www.nuget.org/packages/System.Interactive/) | [](https://www.nuget.org/packages/System.Linq.Async/)
Build | [](https://dev.azure.com/dotnet/Rx.NET/_build/latest?definitionId=9) | [](https://dev.azure.com/dotnet/Rx.NET/_build/latest?definitionId=191) | [](https://dev.azure.com/dotnet/Rx.NET/_build/latest?definitionId=28) | Built as part of Ix
[Azure
Artifacts](https://dev.azure.com/dotnet/Rx.NET/_packaging?_a=feed&feed=RxNet) | [](https://dev.azure.com/dotnet/Rx.NET/_packaging?_a=package&feed=5afc77bd-23b4-46f8-b725-40ebedab630c&package=3c02dce4-f7e9-43ec-a014-28ea9fc46f82&preferRelease=true) | [](https://dev.azure.com/dotnet/Rx.NET/_artifacts/feed/RxNet/NuGet/System.Reactive.Async/&preferRelease=true) | [](https://dev.azure.com/dotnet/Rx.NET/_packaging?_a=package&feed=5afc77bd-23b4-46f8-b725-40ebedab630c&package=a3311bc0-c6ea-4460-bea8-b65d633e2583&preferRelease=true) | [](https://dev.azure.com/dotnet/Rx.NET/_artifacts/feed/RxNet/NuGet/System.Linq.Async/&preferRelease=true)
Release history | [ReleaseHistory](Rx.NET/Documentation/ReleaseHistory/) | [ReleaseHistory](Ix.NET/Documentation/ReleaseHistory/)| [ReleaseHistory](Ix.NET/Documentation/ReleaseHistory/)
For nightly builds, configure NuGet to use this feed: `https://pkgs.dev.azure.com/dotnet/Rx.NET/_packaging/RxNet/nuget/v3/index.json`
### Join the conversation
Catch us in the #rxnet channel over at https://reactivex.slack.com/
## A Brief Introduction to Rx
In this digital age, live data streams are ubiquitous. Financial applications depend on a swift response to timely information. Computer networks have always been able to provide extensive information about their health and operation. Utility companies such as water providers have vast numbers of devices monitoring their operations. User interface and game building frameworks report user interactions in great detail. Delivery vans continuously report their progress. Aircraft provide performance telemetry to detect potential maintenance issues before they become serious problems, and cars are now starting to do the same. Many of us wear or carry devices that track our physical activity and even [vital signs](https://www.youtube.com/watch?v=6yjl_h7-WYA&t=2443s). And the improvements in machine learning have enriched the insights that can be derived from the ever-increasing volume and variety of live data.
But despite being so widespread, live information streams have always been something of a second class citizen. Almost all programming languages have some innate way to work with lists of data (e.g., arrays), but these mechanisms tend to presume that the relevant data is already sitting in memory, ready for us to work with it. What's missing is the liveness—the fact that an information source might produce new data at any moment, on its own schedule.
Rx elevates the support for live streams of information to the same level as we expect for things like arrays. Here's an example:
```cs
var bigTrades =
from trade in trades
where trade.Volume > 1_000_000
select trade;
```
This uses C#'s LINQ feature to filter `trades` down to those entities with a volume greater than one million. This query expression syntax is just a shorthand for method calls, so we could also write it this way:
```cs
var bigTrades = trades.Where(trade => trade.Volume > 1_000_000);
```
The exact behaviour of these two (equivalent) code snippets depends on what type `trades` has. If it were a `IEnumerable`, then this query would just iterate through the list, and `bigTrades` would be an enumerable sequence containing just the matching objects. If `trades` were an object representing a database table (e.g., an [Entity Framework](https://learn.microsoft.com/en-us/ef/core/) [DbSet](https://learn.microsoft.com/en-us/dotnet/api/microsoft.entityframeworkcore.dbset-1), this would be translated into a database query. But if we're using Rx, `trades` would be an `IObservable`, an object reporting live events as they happen. And `bigTrades` would also be an `IObservable`, reporting only those trades with a volume over a million. We can provide Rx with a callback to be invoked each time an observable source has something for us:
```cs
bigTrades.Subscribe(t => Console.WriteLine($"{t.Symbol}: trade with volume {t.Volume}"));
```
The two key features of Rx are:
* a clearly defined way to represent and handle live sequences of data ([`IObservable`](https://learn.microsoft.com/en-us/dotnet/api/system.iobservable-1))
* a set of operators (such as the `Where` operator just shown) enabling event processing logic to be expressed declaratively
Rx has been particularly successfully applied in user interfaces. (This is also true outside of .NET—[RxJS](https://rxjs.dev/) is a JavaScript spin-off of Rx, and it is very popular in user interface code.) The https://github.com/reactiveui/reactiveui makes deep use of Rx to support .NET UI development.
Ian Griffiths presented a concise 60 minute overview of [Reactive Extensions for .NET](https://endjin.com/what-we-think/talks/reactive-extensions-for-dotnet) at the dotnetsheff meetup in 2020. More videos are available on the [Rx playlist](https://www.youtube.com/playlist?list=PLJt9xcgQpM60Fz20FIXBvj6ku4a7WOLGb).
## AsyncRx.Net
Although Rx is a natural way to model asynchronous processes, its original design presumed that code acting on notifications would run synchronously. This is because Rx's design predates C#'s `async`/`await` language features. So although Rx offer adapters that can convert between [`IObservable`](https://learn.microsoft.com/en-us/dotnet/api/system.iobservable-1) and [`Task`](https://learn.microsoft.com/en-us/dotnet/api/system.threading.tasks.task-1), there were certain cases where `async` was not an option.
AsyncRx.Net lifts this restriction by defining `IAsyncObservable`. This enables observers to use asynchronous code. For example, if `bigTrades` were an `IAsyncObservable` we could write this:
```cs
bigTrades.Subscribe(async t => await bigTradeStore.LogTradeAsync(t));
```
AsyncRx.Net is currently in preview.
## Interactive Extensions
Rx defines all the standard LINQ operators available for other providers, but it also adds numerous additional operators. For example, it defines `Scan`, which performs the same basic processing as the standard [`Aggregate`](https://learn.microsoft.com/en-us/dotnet/api/system.linq.enumerable.aggregate) operator, but instead of producing a single result after processing every element, it produces a sequence containing the aggregated value after every single step. (For example, if the operation being aggregated is addition, `Aggregate` would return the sum total as a single output, whereas `Scan` would produce a running total for each input. Given a sequence `[1,2,3]`, `Aggregate((a, x) => a + x)` produces just `6`, whereas `Scan` would produce `[1,3,6]`.)
Some of the additional operators Rx defines are useful only when you're working with events. But some are applicable to sequences of any kind. So the Interactive Extensions (Ix for short) define implementations for `IEnumerable`. Ix is effectively an extension of LINQ to Objects, adding numerous additional operators. (Its usefulness is borne out by the fact that the .NET runtime libraries have, over time, added some of the operators that used to be available only in Ix. For example, .NET 6 added [`MinBy`](https://learn.microsoft.com/en-us/dotnet/api/system.linq.enumerable.minby) and [`MaxBy`](https://learn.microsoft.com/en-us/dotnet/api/system.linq.enumerable.maxby), operators previously only defined by Ix.)
This library is called the "Interactive Extensions" because "Interactive" is in a sense the opposite of "Reactive". (The name does not refer to user interactions.)
## LINQ for `IAsyncEnumerable` (`System.Linq.Async`)
One of the features pioneered by Ix was an asynchronous version of `IEnumerable`. This is another example of a feature so useful that it was eventually added to the .NET runtime libraries: .NET Core 3.0 introduced [`IAsyncEnumerable`](https://learn.microsoft.com/en-us/dotnet/api/system.collections.generic.iasyncenumerable-1), and the associated version C# (8.0) added intrinsic support for this interface with its `await foreach` construct.
Although .NET Core 3.0 defined `IAsyncEnumerable`, it did not add any corresponding LINQ implementation. Whereas [`IEnumerable` supports all the standard operators](https://learn.microsoft.com/en-us/dotnet/api/system.linq.enumerable) such as `Where`, `GroupBy`, and `SelectMany`, .NET does not have built-in implementations of any of these for `IAsyncEnumerable`. However, Ix had provided LINQ operators for its prototype version of `IAsyncEnumerable` from the start, so when .NET Core 3.0 shipped, it was a relatively straightforward task to update all those existing LINQ operators to work with the new, official [`IAsyncEnumerable`](https://learn.microsoft.com/en-us/dotnet/api/system.collections.generic.iasyncenumerable-1).
Thus, the [System.Linq.Async](https://www.nuget.org/packages/System.Linq.Async/) NuGet package was created, providing a LINQ to Objects implementation for `IAsyncEnumerable` to match the one already built into .NET for `IEnumerable`.
Since all of the relevant code was already par