Update IntroToRx for Rx 6.1 (#2242)

* Document ResetExceptionDispatchState
* Add official documentation for the hitherto undocumented rules around reuse of exception objects, and also updates the docs for the various operators whose use might create a scenario where you need this new operator.
* Add TakeUntil(Cancellation) docs
* Add DisposeWith documentation
* Update Foreword to refer to 6.1
* Add edition history to foreword

Rx.NET/Documentation/IntroToRx/00_Foreword.md

@@ -47,8 +47,27 @@ Thanks also to those who continued to work on Rx.NET after it ceased to be direc
 
 If you are interested in more information about the origins of Rx, you might find the [A Little History of Reaqtor](https://reaqtive.net/) ebook illuminating.
 
-The version that this book has been written against is `System.Reactive` version 6.0. The source for this book can be found at [https://github.com/dotnet/reactive/tree/main/Rx.NET/Documentation/IntroToRx](https://github.com/dotnet/reactive/tree/main/Rx.NET/Documentation/IntroToRx). If you find any bugs or other issues in this book, please [create an issue](https://github.com/dotnet/reactive/issues) at https://github.com/dotnet/reactive/. You might find the [Reactive X slack](reactivex.slack.com) to be a useful resource if you start using Rx.NET in earnest.
+The version that this book has been written against is `System.Reactive` version 6.1. The source for this book can be found at [https://github.com/dotnet/reactive/tree/main/Rx.NET/Documentation/IntroToRx](https://github.com/dotnet/reactive/tree/main/Rx.NET/Documentation/IntroToRx). If you find any bugs or other issues in this book, please [create an issue](https://github.com/dotnet/reactive/issues) at https://github.com/dotnet/reactive/. You might find the [Reactive X slack](reactivex.slack.com) to be a useful resource if you start using Rx.NET in earnest.
 
 So, fire up Visual Studio and let's get started.
 
+# Edition History
+
+## 1st edition
+
+The original book written by Lee Campbell.
+
+## 2nd edition
+
+Updated and revised by Ian Griffiths to align with the Rx v6.0 release.
+
+## 3rd edition
+
+Updates for Rx v6.1:
+
+* Documented new `TakeUntil(CancellationToken)` overload
+* Documented new `DiposeWith` extension method for `IDisposable`
+* Documented new `ResetExceptionDispatchState` operator
+* Added guidance to clarify rules around reusing exception objects in scenarios where Rx will rethrow them
+
 ---

Rx.NET/Documentation/IntroToRx/03_CreatingObservableSequences.md

@@ -381,6 +381,9 @@ As an example of where you might want to use `Never` for timing purposes, suppos
 IObservable<string> throws = Observable.Throw<string>(new Exception()); 
 ```
 
+Be aware that this if you use this operator in conjunction with any of the mechanisms described in the [Leaving Rx's World](13_LeavingIObservable.md) chapter, you might fall foul of the rules described in [Exception state](13_LeavingIObservable.md#exception-state). If you need to use `await` (or similar mechanisms that will turn a call to `OnError` into a rethrow) you may need to use the [`ResetExceptionDispatchState`](13_LeavingIObservable.md#resetexceptiondispatchstate) operator to ensure that each rethrowing of the exception gets suitably reset exception state. (This problem only arises if you cause the same exception to be rethrown multiple times.)
+
+
 ### Observable.Create
 
 The `Create` factory method is more powerful than the other creation methods because it can be used to create any kind of sequence. You could implement any of the preceding four methods with `Observable.Create`.
@@ -516,7 +519,7 @@ public static IObservable<T> Never<T>()
     });
 }
 
-public static IObservable<T> Throws<T>(Exception exception)
+public static IObservable<T> Throw<T>(Exception exception)
 {
     return Observable.Create<T>(o =>
     {
@@ -1201,6 +1204,9 @@ Sub2: 4
 
 Alternatively, you can specify a time-based limit by passing a `TimeSpan` to the `ReplaySubject<T>` constructor.
 
+Note that if the source reports an error (by calling `OnError`), `ReplaySubject<T>` will retain the `Exception`, and provide it to all current subscribers, and also any subsequent subscribers. This should not be a surprise—this subject's job is to replay what the source did—but be aware that this can cause a problem if you use any of the mechanisms described in the [Leaving Rx's World](13_LeavingIObservable.md) chapter. For example if you `await` an observable that uses a `ReplaySubject<T>` and if the underlying source reported an error, you will no longer be conforming to the rules described in [Exception state](13_LeavingIObservable.md#exception-state). If you need to use `await` (or similar mechanisms that will turn a call to `OnError` into a rethrow) you may need to use the [`ResetExceptionDispatchState`](13_LeavingIObservable.md#resetexceptiondispatchstate) operator to ensure that each rethrowing of the exception gets suitably reset exception state.
+
+
 ## `BehaviorSubject<T>`
 
 Like `ReplaySubject<T>`, `BehaviorSubject<T>` also has a memory, but it remembers exactly one value. However, it's not quite the same as a `ReplaySubject<T>` with a buffer size of 1. Whereas a `ReplaySubject<T>` starts off in a state where it has nothing in its memory, `BehaviorSubject<T>` always remembers _exactly_ one item. How can that work before we've made our first call to `OnNext`? `BehaviorSubject<T>` enforces this by requiring us to supply the initial value when we construct it.

Rx.NET/Documentation/IntroToRx/05_Filtering.md

@@ -383,6 +383,8 @@ We don't have to use a time, `TakeUntil` offers an overload that accept a second
 
 **Note**: these overloads require the second observable to produce a value in order to trigger the start or end. If that second observable completes without producing a single notification, then it has no effect—`TakeUntil` will continue to take items indefinitely; `SkipUntil` will never produce anything. In other words, these operators would treat `Observable.Empty<T>()` as being effectively equivalent to `Observable.Never<T>()`.
 
+There is also an overload of `TakeUntil` that accepts a `CancellationToken`. This forwards notifications from the source until either the source itself completes, or the token signals cancellation, at which point `TakeUntil` will complete.
+
 ### Distinct and DistinctUntilChanged
 
 `Distinct` is yet another standard LINQ operator. It removes duplicates from a sequence. To do this, it needs to remember all the values that its source has ever produced, so that it can filter out any items that it has seen before. Rx includes an implementation of `Distinct`, and this example uses it to display the unique identifier of vessels generating AIS messages, but ensuring that we only display each such identifier the first time we see it:

Rx.NET/Documentation/IntroToRx/13_LeavingIObservable.md

@@ -6,7 +6,181 @@ Rx's compositional nature is the key to its power, but it can look like a proble
 
 You've already seen some answer to these questions. The [Creating Observable Sequences chapter](03_CreatingObservableSequences.md) showed various ways to create observable sources. But when it comes to handling the items that emerge from an `IObservable<T>`, all we've really seen is how to implement [`IObserver<T>`](02_KeyTypes.md#iobserver), and [how to use the callback based `Subscribe` extension methods  to subscribe to an `IObservable<T>`](02_KeyTypes.md#iobservable).
 
-In this chapter, we will look at the methods in Rx which allow you to leave the `IObservable<T>` world, so you can take action based on the notifications that emerge from an Rx source.
+In this chapter, we will look at the methods in Rx which allow you to leave the `IObservable<T>` world, so you can take action based on the notifications that emerge from an Rx source. But first, we need to look at the challenge this creates for error handling.
+
+## Exception state
+
+There's a rule you must conform to when exceptions leave Rx's world: if a developer using Rx chooses to use a mechanism that takes exceptions delivered by an `IObservable<T>` and throws them (e.g. if you `await` an `IObservable<T>`) then it is the developer's responsibility to ensure that either:
+
+* each exception object is used only once
+
+or
+
+* the exception's dispatch state is reset prior to being supplied to the observer that will be rethrowing it (e.g., by executing a `throw`)
+
+For example, if you use the [`Throw`](03_CreatingObservableSequences.md#observablethrow) operator, this creates a problem if you plan to `await` the resulting sequence: by design, `Throw` does nothing at all to the state of the exception you supply. This is deliberate, because you might be giving it an exception that already has important state there and it does not want to destroy that information. However, in cases where the exception has never actually been thrown, the state never gets set (and therefore never reset). There is no good way to detect whether exception state is present, so `Throw` must conservatively assume that it might be. This means that if you were to `await` such a sequence twice, you'd be breaking these rules. Rx 6.1 introduced a new operator, [`ResetExceptionDispatchState`](#resetexceptiondispatchstate), to deal with this scenario. If you write something like this:
+
+```cs
+IObservable<int> t = Observable
+    .Throw<int>(new Exception("Boom!"))
+    .ResetExceptionDispatchState();
+```
+
+the presence of that `ResetExceptionDispatchState` ensures that when the `Throw` reports the error by calling `OnError`, the exception state is reset at that instant.
+
+But why do these rules exist?
+
+Code that lives entirely within Rx's world reports errors by having an `IObservable<T>` invoke an observer's `OnError` method, passing an `Exception` representing the error. Exceptions are reported, discovered, and handled without ever using `throw` or `catch`. This makes Rx's use of exceptions somewhat unusual, and there's a consequence: if you construct but do not throw an exception, its _exception state_ is never set.
+
+Exception state is the information describing the context in which the exception was thrown. It includes a textual description of the call stack, which can help developers diagnose the problem. It also includes more machine-friendly information, describing the crash site with technical identifiers that can help automated systems analyze errors. [Windows Error Reporting](https://learn.microsoft.com/en-us/windows/win32/wer/windows-error-reporting) can use this to distinguish between different kinds of application failures. For desktop applications, this information can (with user consent) be collected centrally, enabling developers to discover which crashes users encounter most often in practical use. Failures are identified by the particular combination of technical identifiers (which effectively identify the particular location in the code that failed, and also the type of failure that occurred). Each unique combination is known informally as a _fault bucket_. You can see this information in the Windows Event Viewer. If a .NET application throws an unhandled exception, you typically get three events in the event log:
+
+* a **.NET Runtime** entry (typically with Event ID 1026) reporting the application executable name, .NET version, and .NET stack trace text
+* an **Application Crashing Events** entry (typically with Event ID 1000) with generic (non-.NET-specific) details in a form that would appear for any application crash, such as the executable path, Win32 SEH exception type, the DLL in which the exception originated and the offset within that DLL
+* a **Windows Error Reporting** entry (typically with Event ID 1001) containing the _fault bucket_ details
+
+Here's an example of that third kind from a crashing .NET application:
+
+```
+Fault bucket 1162564721043373785, type 4
+Event Name: APPCRASH
+Response: Not available
+Cab Id: 0
+
+Problem signature:
+P1: TestResetExceptionDispatchState.exe
+P2: 1.0.0.0
+P3: 68a40000
+P4: KERNELBASE.dll
+P5: 10.0.26100.6584
+P6: 0a9b38fe
+P7: e0434352
+P8: 00000000000c66ca
+P9: 
+P10: 
+```
+
+This is harder to understand than a stack trace, but these numeric identifiers make it easier to determine automatically whether two crashes are in some sense 'the same'. The idea here is to make it possible to discover when the same kind of error is happening a lot, enabling developers to focus their efforts on fixing the types of crashes causing the most trouble.
+
+With a .NET application, we want this information to reflect the origin of the failure. But it's quite common for exceptions to be caught and rethrown. What we _don't_ want is for every single error to be categorised as the same type just because we wrote some common error handling code. .NET itself has to contend with this: when we use `await` to wait for an asynchronous operation that has failed, that failure exception will have been put into a `Task` or `ValueTask`, and .NET rethrows that into the code that calls `await`. It would be deeply unhelpful if every single failing asynchronous operation was reported to Windows Error Handling as having occurred inside the .NET runtime library code that rethrows exceptions captured by a `Task`!
+
+The purpose of _exception state_ is to hold onto the exception's origin information even if the exception gets rethrown. If you rethrow an exception from within a `catch` block by using `throw;` this exceptions state mostly gets preserved. ('Mostly', because the stack information gets augmented: the stack trace will report both the original throw location and the rethrow location.) This does _not_ work correctly if you try to rethrow the exception with `throw ex;` inside the `catch` block, which is why it's important to use the no-arguments `throw;` form when rethrowing. As for exceptions that end up being caught in one place and rethrown in a completely different place (e.g. on a different thread because `await` is being used) the .NET runtime libraries provide a helper called [`ExceptionDispatchInfo`](https://learn.microsoft.com/dotnet/api/system.runtime.exceptionservices.exceptiondispatchinfo) that can help manage this exception state, and ensure that it is used when rethrowing an exception in a completely different context from which it was thrown.
+
+Rx uses this to ensure that examples such as the following behave as expected:
+
+```cs
+IObservable<string> fileLines = Observable.Create<string>(async obs =>
+{
+    using var reader = new StreamReader(@"c:\temp\test.txt");
+
+    while ((await reader.ReadLineAsync()) is string line)
+    {
+        obs.OnNext(line);
+    }
+});
+
+string firstNonEmptyLine = await fileLines
+    .FirstAsync(line => line.Length > 0);
+Console.WriteLine(firstNonEmptyLine);
+```
+
+If the attempt to open the file throws an exception, what do we expect to see? A developer familiar with how exceptions generally work with `async`/`await` in .NET might reasonably expect the exception to report two stack traces: one for the point at which the exception was originally thrown, and another for where it was rethrown from the `await`. And that's exactly what we see:
+
+```
+Unhandled exception. System.IO.FileNotFoundException: Could not find file 'c:\temp\test.txt'.
+File name: 'c:\temp\test.txt'
+   at Microsoft.Win32.SafeHandles.SafeFileHandle.CreateFile(String fullPath, FileMode mode, FileAccess access, FileShare share, FileOptions options)
+   at Microsoft.Win32.SafeHandles.SafeFileHandle.Open(String fullPath, FileMode mode, FileAccess access, FileShare share, FileOptions options, Int64 preallocationSize, Nullable`1 unixCreateMode)
+   at System.IO.Strategies.OSFileStreamStrategy..ctor(String path, FileMode mode, FileAccess access, FileShare share, FileOptions options, Int64 preallocationSize, Nullable`1 unixCreateMode)
+   at System.IO.Strategies.FileStreamHelpers.ChooseStrategyCore(String path, FileMode mode, FileAccess access, FileShare share, FileOptions options, Int64 preallocationSize, Nullable`1 unixCreateMode)
+   at System.IO.StreamReader.ValidateArgsAndOpenPath(String path, Encoding encoding, Int32 bufferSize)
+   at System.IO.StreamReader..ctor(String path)
+   at Program.<>c.<<<Main>$>b__0_0>d.MoveNext() in D:\source\RxThrowExamples\RxThrowExamples\Program.cs:line 5
+--- End of stack trace from previous location ---
+   at System.Reactive.PlatformServices.ExceptionServicesImpl.Rethrow(Exception exception)
+   at System.Reactive.ExceptionHelpers.Throw(Exception exception)
+   at System.Reactive.Subjects.AsyncSubject`1.GetResult()
+   at Program.<Main>$(String[] args) in D:\source\RxThrowExamples\RxThrowExamples\Program.cs:line 13
+   at Program.<Main>(String[] args)
+```
+
+This conforms to the rules described above because the exception is thrown in the conventional .NET manner here. Rx then catches it—this overload of `Observable.Create` wraps the `Task` returned by the callback in an adapter that detects when the `Task` enters a faulted state, in which case it extracts the exception and passes it to the subscribing `IObserver<T>`. And then the awaiter that Rx provides when you `await` an observable rethrows this same exception. Rx uses the `ExceptionDispatchInfo` mechanisms to ensure that the exception state captured at the point where the exception was originally thrown remains present when the exception is eventually rethrown to the code that used `await` on the observable.
+
+This is fine when the exception originates by being thrown in the conventional way, which it does in that last example. Where it can go wrong is if the exception never gets thrown at all: the rethrowing mechanism provided by `ExceptionDispatchInfo` gets a little confused in this case. This problem is not specific to Rx by the way, as the following code illustrates:
+
+```cs
+Exception ox = new("Kaboom!");
+
+for (int i = 0; i < 3; ++i)
+{
+    Console.WriteLine();
+    Console.WriteLine();
+
+    try
+    {
+        await Task.FromException(ox); // Exception thrown here
+
+    }
+    catch (Exception x)
+    {
+        Console.WriteLine(x);
+    }
+}
+```
+
+This creates a single `Exception` object and then wraps it in a `Task` (using `Task.FromException`) which it immediately `await`s. There is no Rx code here. Running this we get the following slightly surprising behaviour:
+
+```
+System.Exception: Kaboom!
+   at Program.<Main>$(String[] args) in D:\source\RxThrowExamples\ThrowStackDupWithoutRx\Program.cs:line 10
+
+
+System.Exception: Kaboom!
+   at Program.<Main>$(String[] args) in D:\source\RxThrowExamples\ThrowStackDupWithoutRx\Program.cs:line 10
+   at Program.<Main>$(String[] args) in D:\source\RxThrowExamples\ThrowStackDupWithoutRx\Program.cs:line 10
+
+
+System.Exception: Kaboom!
+   at Program.<Main>$(String[] args) in D:\source\RxThrowExamples\ThrowStackDupWithoutRx\Program.cs:line 10
+   at Program.<Main>$(String[] args) in D:\source\RxThrowExamples\ThrowStackDupWithoutRx\Program.cs:line 10
+   at Program.<Main>$(String[] args) in D:\source\RxThrowExamples\ThrowStackDupWithoutRx\Program.cs:line 10
+```
+
+Notice that each time around the loop, the stack trace gets longer. The problem here is that nothing ever resets the exception dispatch state. Normally that happens when the original `throw` occurs, but in this example we never use `throw`, and so that reset never happens. When you `await` a `Task`, the .NET runtime uses `ExceptionDispatchInfo` to rethrow the exception, and the point of that mechanism is that it preserves the original exceptions state, and appends the current stack so that you get a complete record of the exception's history. It wasn't designed to cope with an exception being rethrown multiple times without corresponding multiple `throw`s.
+
+Since Rx uses the same mechanism when exceptions emerge out of Rx's world, the Rx equivalent of that `Task` example has exactly the same problem:
+
+```cs
+IObservable<int> ts = Observable.Throw<int>(new Exception("Pow!"));
+
+for (int i = 0; i < 3; ++i)
+{
+    Console.WriteLine();
+    Console.WriteLine();
+
+	try
+	{
+		await ts; // Exception thrown here
+
+	}
+	catch (Exception x)
+	{
+        Console.WriteLine(x);
+	}
+}
+```
+
+This will also produce a longer stack trace each time round the loop, because again, we're rethrowing the same exception object multiple times without ever resetting its state.
+
+Remember, Rx deliberately preserves the state because for all it knows, that state was important. (This is how the example with the `FileNotFoundException` above reports the correct information). This is why Rx 6.1 introduced the [`ResetExceptionDispatchState`](#resetexceptiondispatchstate) operator. It enables us to tell Rx that we do in fact want it to reset the exception state each time an error emerges from an observable. If we change the first statement of the example above to this:
+
+```cs
+IObservable<int> ts = Observable
+   .Throw<int>(new Exception("Pow!"))
+   .ResetExceptionDispatchState();
+```
+
+this now conforms to the rules described at the start of this section, which prevents the ever-growing stack trace problem.
+
 
 ## Integration with `async` and `await`
 
@@ -209,6 +383,18 @@ public static IObservable<IDictionary<TKey, TSource>> ToDictionary<TSource, TKey
 
 The `ToLookup` extension offers near-identical-looking overloads, the difference being the return type (and the name, obviously). They all return an `IObservable<ILookup<TKey, TElement>>`. As with LINQ to Objects, the distinction between a dictionary and a lookup is that the `ILookup<TKey, TElement>>` interface allows each key to have any number of values, whereas a dictionary maps each key to one value.
 
+## ResetExceptionDispatchState
+
+The `ResetExceptionDispatchState` instructs Rx to reset the _exception dispatch state_ of any exception passed to `OnError` before passing the exception on to its observer. Other than that, this operator just passes all notifications through unmodified.
+
+The purpose of this operator is to make it easier to conform with the rules described in the [Exception state](#exception-state) section above. You typically only need to use if when an exception originates from an Rx observable without ever having executed a `throw`, and you then go on to cause that exception to leave Rx's world and to be rethrown in the normal .NET way.
+
+For example, [`Observable.Throw`](03_CreatingObservableSequences.md#observablethrow) does not reset the state of the exception you give it (in case there was anything important in that state), but if you might be using a world-crossing mechanism like `await` that will rethrow the exception, and if you might do so multiple times for the same exception object, you will need to use this operator to ensure that the exception state gets reset each time.
+
+This can also be useful when using the [`Replay`](15_PublishingOperators.md#replay) operator, because even if the underlying source is using `throw` (thus resetting the exception state), `Replay` will hold onto the exception, and will deliver the same object to any subsequent subscribers. The use of `Replay` effectively prevents the reset that would otherwise have happened. So if you then go on to use `await` (or any other of the mechanisms in this chapter that would cause an exception delivered to `OnError` to be rethrown) you will no longer be following the rules described in [Exception state](#exception-state), which can result in problems such as ever-longer stack traces.
+
+If you remain entirely within Rx's world, you should not need to use `ResetExceptionDispatchState`. This operator exists only to deal with a problem that can occur when crossing from Rx's world to the world of conventional exception throwing.
+
 ## ToTask
 
 Although Rx provides direct support for using `await` with an `IObservable<T>`, it can sometimes be useful to obtain a `Task<T>` representing an `IObservable<T>`. This is useful because some APIs expect a `Task<T>`. You can call `ToTask()` on any `IObservable<T>`, and this will subscribe to that observable, returning a `Task<T>` that will complete when the task completes, producing the sequence's final output as the task's result. If the source completes without producing an element, the task will enter a faulted state, with an `InvalidOperation` exception complaining that the input sequence contains no elements.

Rx.NET/Documentation/IntroToRx/15_PublishingOperators.md

@@ -278,6 +278,8 @@ The `ReplaySubject<T>` that enables this behaviour will consume memory to store
 
 `Replay` also supports the per-subscription-multicast model I showed for the other `Multicast`-based operators in this section.
 
+Note that if the source reports an error (by calling `OnError`), `Replay` will retain the `Exception`, and provide it to all current subscribers, and also any subsequent subscribers. This should not be a surprise—`Replay`'s job is to replay what the source did—but be aware that this can cause a problem if you use any of the mechanisms described in the [Leaving Rx's World](13_LeavingIObservable.md) chapter. For example if you `await` an observable produced by `Replay` and if the underlying source reported an error, you will no longer be conforming to the rules described in [Exception state](13_LeavingIObservable.md#exception-state). If you need to use `await` (or similar mechanisms that will turn a call to `OnError` into a rethrow) you may need to use the [`ResetExceptionDispatchState`](13_LeavingIObservable.md#resetexceptiondispatchstate) operator to ensure that each rethrowing of the exception gets suitably reset exception state.
+
 ## RefCount
 
 We saw in the preceding section that `Multicast` (and also its various wrappers) supports two usage models:

Rx.NET/Documentation/IntroToRx/B_Disposables.md

@@ -21,7 +21,20 @@ The `CancellationDisposable` class offers an integration point between the .NET
 
 ## `CompositeDisposable`
 
-The `CompositeDisposable` type allows you to treat many disposable resources as one. You can create an instance of `CompositeDisposable` by passing in a <code>params</code> array of disposable resources. Calling `Dispose` on the `CompositeDisposable` will call dispose on each of these resources in the order they were provided. Additionally, the `CompositeDisposable` class implements `ICollection<IDisposable>`; this allows you to add and remove resources from the collection. After the `CompositeDisposable` has been disposed of, any further resources that are added to this collection will be disposed of instantly. Any item that is removed from the collection is also disposed of, regardless of whether the collection itself has been disposed of. This includes usage of both the `Remove` and `Clear` methods.
+The `CompositeDisposable` type allows you to treat many disposable resources as one. You can create an instance of `CompositeDisposable` by passing in a <code>params</code> array of disposable resources. You can also add disposable resources to an existing `CompositeDisposable` instance by calling its `Add` method. Calling `Dispose` on the `CompositeDisposable` will call dispose on each of these resources in the order they were provided. Additionally, the `CompositeDisposable` class implements `ICollection<IDisposable>`; this allows you to add and remove resources from the collection. After the `CompositeDisposable` has been disposed of, any further resources that are added to this collection will be disposed of instantly. Any item that is removed from the collection is also disposed of, regardless of whether the collection itself has been disposed of. This includes usage of both the `Remove` and `Clear` methods.
+
+The `System.Reactive.Disposables.Fluent` namespace defines an extension method called `DisposeWith`, available on any `IDisposable`, for use with `CompositeDisposable`, providing another way to add disposable resources. The typical usage for this is to add subscriptions, e.g.:
+
+```cs
+CompositeDisposable d = new();
+
+someObservable1.Subscribe(myObserver1)
+    .DisposeWith(d);
+someObservable2.Subscribe(myObserver2)
+    .DisposeWith(d);
+```
+
+This has exactly the same effect as if we had used `Add`, but it supports the 'fluent' style of development in which we build up behaviour by chaining together multiple method calls. `DisposeWith` returns its argument, enabling further calls to be chained on if you wish.
 
 ## `ContextDisposable`
 `ContextDisposable` allows you to enforce that disposal of a resource is performed on a given `SynchronizationContext`. The constructor requires both a `SynchronizationContext` and an `IDisposable` resource. When the `Dispose` method is invoked on the `ContextDisposable`, the provided resource will be disposed of on the specified context.

Rx.NET/Documentation/adr/0004-onerror-to-throw.md

@@ -69,7 +69,7 @@ File name: 'c:\temp\test.txt'
    at System.Reactive.Subjects.AsyncSubject`1.GetResult()
    at Program.<Main>$(String[] args) in D:\source\RxThrowExamples\RxThrowExamples\Program.cs:line 13
    at Program.<Main>(String[] args)
-   ```
+```
 
 This is straightforward because the exception here is thrown in the conventional .NET manner. It happens to be caught by Rx—this overload of `Observable.Create` wraps the `Task` returned by the callback in an adapter that detects when the `Task` enters a faulted state, in which case it extracts the exception and passes it to the subscribing `IObserver<T>`. And then the awaiter that Rx provides when you `await` an observable rethrows this same exception.
 
@@ -154,7 +154,7 @@ It occurs as a direct result of the steps Rx takes to produce the stack trace we
 
 This behaviour is not peculiar to Rx. It originates from `ExceptionDispatchInfo.Throw` and we can create a `Task`-based version of this behaviour without using Rx:
 
-```
+```cs
 Exception ox = new("Kaboom!");
 
 for (int i = 0; i < 3; ++i)