Feature: TestObserver<T>

[JakenVeina]

Implemented TestObserver<T> as a testing utility for aggregating notifications from any arbitrary observable stream.

This is something I ended up wanting to support testing for the ExpireAfter() operator (the version that operates on ISourceCache<TObject, TKey> and ISourceList<T> directly, and emits sets of removed items, rather than full changesets).

I tried using the ITestingObserver<T> API from Microsoft.Reactive.Testing first, but I found it way to cumbersome, in the fact that it doesn't actually aggregate notifications to allow easy checking for errors or completions, and it is only usable if you setup ALL your test arrangements and actions to run through a TestScheduler.

I wanted an easy way to:

• Inspect the sequence of emitted items.
• Check if an error occurred.
• Check if completion occurred.
• Validate general correctness of the stream, I.E. only one error or completion should ever occur, and no notifications should ever follow them.

In other words, I wanted what ChangeSetAggregator does.

Right now, I just put this in the testing project. I figured it doesn't really belong in the main package, if it's not something that supports DynamicData operators or testing, directly, but I could definitely move it, if anyone thinks it's useful as part of the public API. If so, I'd remove the dependency on Microsoft.Reactive.Testing.Recorded<T>.

[JakenVeina]

So, I re-designed my whole concept for this testing observer, from scratch, after thinking some more. The testing functionality I was going for is now split up across several different reusable classes, to be more flexible.

I went back to using Microsoft.Reactive.Testing's concept of ITestingObserver<T>, although I did still have to create a custom TestingObserver<T> implementation of it, since their own implementation is internal and can't be created directly in their API. I might go ahead and make a PR over there to see if they'll expose it. To accommodate my previous issues with the cumbersomeness with using this interface, I added extension methods to support easily doing assertions against it.

There is now a .ValidateSynchronization() operator which will immediately detect and throw any unsynchronized notifications it receives, rather than hoping that a downstream race condition will cause an error.

To implement this operator, there are now the RawAnonymousObserver<T> and RawAnonymousObservable<T> classes which mimic the Observer.Create<T> and Observable.Create<T> behavior, except without any of the internal safeguards that prevent out-of-sequence notifications and do automatic cleanup. These may be useful in other future tests, and will allow us to better test whether our custom operators are actually compliant with the RX contract.

Altogether, here's an example of how you'd write a test with these utilities:

var source = RawAnonymousObservable.Create<int>(observer =>
{
    Parallel.For(1, 10_000, observer.OnNext);

    observer.OnCompleted();

    return Disposable.Empty;
});

var results = TestableObserver.Create<int>();

using var subscription = source
    .ValidateSynchronization()
    .Subscribe(results);

results.TryGetRecordedError().Should().BeOfType<UnsynchronizedNotificationException<int>>();
results.TryGetRecordedCompletion().Should().BeTrue();
results.EnumerateInvalidNotifications().Should().NotBeEmpty();

[RolandPheasant]

This PR is super cool, and I look forward to seeing these concepts in action.

[github-actions]

This pull request has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.