update reactive assertions

Author: rose-a

tests/GraphQL.Client.Tests.Common/FluentAssertions.Reactive/FluentTestObserver.cs

@@ -1,10 +1,10 @@
 using System;
 using System.Collections.Generic;
-using System.Diagnostics;
+using System.Linq;
 using System.Reactive;
 using System.Reactive.Concurrency;
+using System.Reactive.Disposables;
 using System.Reactive.Linq;
-using System.Threading;
 using Microsoft.Reactive.Testing;
 
 namespace GraphQL.Client.Tests.Common.FluentAssertions.Reactive
@@ -16,7 +16,7 @@ namespace GraphQL.Client.Tests.Common.FluentAssertions.Reactive
     public class FluentTestObserver<TPayload> : IObserver<TPayload>, IDisposable
     {
         private readonly IDisposable _subscription;
-        private readonly EventLoopScheduler _observeScheduler = new EventLoopScheduler();
+        private readonly IScheduler _observeScheduler;
         private readonly RollingReplaySubject<Recorded<Notification<TPayload>>> _rollingReplaySubject = new RollingReplaySubject<Recorded<Notification<TPayload>>>();
 
         /// <summary>
@@ -33,13 +33,29 @@ public class FluentTestObserver<TPayload> : IObserver<TPayload>, IDisposable
         /// The recorded <see cref="Notification{T}"/>s
         /// </summary>
         public IEnumerable<Recorded<Notification<TPayload>>> RecordedNotifications =>
-            _rollingReplaySubject.Snapshot();
+            _rollingReplaySubject.GetSnapshot();
 
         /// <summary>
         /// The recorded messages
         /// </summary>
         public IEnumerable<TPayload> RecordedMessages =>
             RecordedNotifications.GetMessages();
+        
+        /// <summary>
+        /// The exception 
+        /// </summary>
+        public Exception Error =>
+            RecordedNotifications
+                .Where(r => r.Value.Kind == NotificationKind.OnError)
+                .Select(r => r.Value.Exception)
+                .FirstOrDefault();
+        
+        /// <summary>
+        /// The recorded messages
+        /// </summary>
+        public bool Completed =>
+            RecordedNotifications
+                .Any(r => r.Value.Kind == NotificationKind.OnCompleted);
 
         /// <summary>
         /// Creates a new <see cref="FluentTestObserver{TPayload}"/> which subscribes to the supplied <see cref="IObservable{T}"/>
@@ -48,10 +64,30 @@ public class FluentTestObserver<TPayload> : IObserver<TPayload>, IDisposable
         public FluentTestObserver(IObservable<TPayload> subject)
         {
             Subject = subject;
-            _observeScheduler.Schedule(() =>
-                Debug.WriteLine($"Observe scheduler thread id: {Thread.CurrentThread.ManagedThreadId}"));
+            _observeScheduler = new EventLoopScheduler();
+            _subscription = new CompositeDisposable(); subject.ObserveOn(_observeScheduler).Subscribe(this);
+        }
+
+        /// <summary>
+        /// Creates a new <see cref="FluentTestObserver{TPayload}"/> which subscribes to the supplied <see cref="IObservable{T}"/>
+        /// </summary>
+        /// <param name="subject">the <see cref="IObservable{T}"/> under test</param>
+        public FluentTestObserver(IObservable<TPayload> subject, IScheduler scheduler)
+        {
+            Subject = subject;
+            _observeScheduler = scheduler;
+            _subscription = subject.ObserveOn(scheduler).Subscribe(this);
+        }
 
-            _subscription = subject.ObserveOn(_observeScheduler).Subscribe(this);
+        /// <summary>
+        /// Creates a new <see cref="FluentTestObserver{TPayload}"/> which subscribes to the supplied <see cref="IObservable{T}"/>
+        /// </summary>
+        /// <param name="subject">the <see cref="IObservable{T}"/> under test</param>
+        public FluentTestObserver(IObservable<TPayload> subject, TestScheduler testScheduler)
+        {
+            Subject = subject;
+            _observeScheduler = testScheduler;
+            _subscription = subject.ObserveOn(Scheduler.CurrentThread).Subscribe(this);
         }
 
         /// <summary>
@@ -60,8 +96,11 @@ public FluentTestObserver(IObservable<TPayload> subject)
         public void Clear() => _rollingReplaySubject.Clear();
 
         /// <inheritdoc />
-        public void OnNext(TPayload value) =>
-            _rollingReplaySubject.OnNext(new Recorded<Notification<TPayload>>(_observeScheduler.Now.UtcTicks, Notification.CreateOnNext(value)));
+        public void OnNext(TPayload value)
+        {
+            _rollingReplaySubject.OnNext(
+                new Recorded<Notification<TPayload>>(_observeScheduler.Now.UtcTicks, Notification.CreateOnNext(value)));
+        }
 
         /// <inheritdoc />
         public void OnError(Exception exception) =>
@@ -75,14 +114,13 @@ public void OnCompleted() =>
         public void Dispose()
         {
             _subscription?.Dispose();
-            _observeScheduler.Dispose();
             _rollingReplaySubject?.Dispose();
         }
 
         /// <summary>
-        /// Returns an <see cref="ObservableAssertions{TPayload}"/> object that can be used to assert the observed <see cref="IObservable{T}"/>
+        /// Returns an <see cref="ReactiveAssertions{TPayload}"/> object that can be used to assert the observed <see cref="IObservable{T}"/>
         /// </summary>
         /// <returns></returns>
-        public ObservableAssertions<TPayload> Should() => new ObservableAssertions<TPayload>(this);
+        public ReactiveAssertions<TPayload> Should() => new ReactiveAssertions<TPayload>(this);
     }
 }