Merge pull request #4 from fluentassertions/fix-unit-test

fix unit test

Author: rose-a

Src/FluentAssertions.Reactive/FluentTestObserver.cs

@@ -15,9 +15,9 @@ namespace FluentAssertions.Reactive
     /// <typeparam name="TPayload"></typeparam>
     public class FluentTestObserver<TPayload> : IObserver<TPayload>, IDisposable
     {
-        private readonly IDisposable subscription;
-        private readonly IScheduler observeScheduler;
-        private readonly RollingReplaySubject<Recorded<Notification<TPayload>>> rollingReplaySubject = new RollingReplaySubject<Recorded<Notification<TPayload>>>();
+        private readonly IDisposable _subscription;
+        private readonly IScheduler _observeScheduler;
+        private readonly RollingReplaySubject<Recorded<Notification<TPayload>>> _rollingReplaySubject = new RollingReplaySubject<Recorded<Notification<TPayload>>>();
 
         /// <summary>
         /// The observable which is observed by this instance
@@ -27,13 +27,13 @@ public class FluentTestObserver<TPayload> : IObserver<TPayload>, IDisposable
         /// <summary>
         /// The stream of recorded <see cref="Notification{T}"/>s
         /// </summary>
-        public IObservable<Recorded<Notification<TPayload>>> RecordedNotificationStream => rollingReplaySubject.AsObservable();
+        public IObservable<Recorded<Notification<TPayload>>> RecordedNotificationStream => _rollingReplaySubject.AsObservable();
 
         /// <summary>
         /// The recorded <see cref="Notification{T}"/>s
         /// </summary>
         public IEnumerable<Recorded<Notification<TPayload>>> RecordedNotifications =>
-            rollingReplaySubject.GetSnapshot();
+            _rollingReplaySubject.GetSnapshot();
 
         /// <summary>
         /// The recorded messages
@@ -64,8 +64,8 @@ public class FluentTestObserver<TPayload> : IObserver<TPayload>, IDisposable
         public FluentTestObserver(IObservable<TPayload> subject)
         {
             Subject = subject;
-            observeScheduler = new EventLoopScheduler();
-            subscription = new CompositeDisposable(); subject.ObserveOn(observeScheduler).Subscribe(this);
+            _observeScheduler = new EventLoopScheduler();
+            _subscription = subject.ObserveOn(_observeScheduler).Subscribe(this);
         }
 
         /// <summary>
@@ -76,8 +76,8 @@ public FluentTestObserver(IObservable<TPayload> subject)
         public FluentTestObserver(IObservable<TPayload> subject, IScheduler scheduler)
         {
             Subject = subject;
-            observeScheduler = scheduler;
-            subscription = subject.ObserveOn(scheduler).Subscribe(this);
+            _observeScheduler = scheduler;
+            _subscription = subject.ObserveOn(scheduler).Subscribe(this);
         }
 
         /// <summary>
@@ -88,35 +88,35 @@ public FluentTestObserver(IObservable<TPayload> subject, IScheduler scheduler)
         public FluentTestObserver(IObservable<TPayload> subject, TestScheduler testScheduler)
         {
             Subject = subject;
-            observeScheduler = testScheduler;
-            subscription = subject.ObserveOn(Scheduler.CurrentThread).Subscribe(this);
+            _observeScheduler = testScheduler;
+            _subscription = subject.ObserveOn(Scheduler.CurrentThread).Subscribe(this);
         }
 
         /// <summary>
         /// Clears the recorded notifications and messages as well as the recorded notifications stream buffer
         /// </summary>
-        public void Clear() => rollingReplaySubject.Clear();
+        public void Clear() => _rollingReplaySubject.Clear();
 
         /// <inheritdoc />
         public void OnNext(TPayload value)
         {
-            rollingReplaySubject.OnNext(
-                new Recorded<Notification<TPayload>>(observeScheduler.Now.UtcTicks, Notification.CreateOnNext(value)));
+            _rollingReplaySubject.OnNext(
+                new Recorded<Notification<TPayload>>(_observeScheduler.Now.UtcTicks, Notification.CreateOnNext(value)));
         }
 
         /// <inheritdoc />
         public void OnError(Exception exception) =>
-            rollingReplaySubject.OnNext(new Recorded<Notification<TPayload>>(observeScheduler.Now.UtcTicks, Notification.CreateOnError<TPayload>(exception)));
+            _rollingReplaySubject.OnNext(new Recorded<Notification<TPayload>>(_observeScheduler.Now.UtcTicks, Notification.CreateOnError<TPayload>(exception)));
 
         /// <inheritdoc />
         public void OnCompleted() =>
-            rollingReplaySubject.OnNext(new Recorded<Notification<TPayload>>(observeScheduler.Now.UtcTicks, Notification.CreateOnCompleted<TPayload>()));
+            _rollingReplaySubject.OnNext(new Recorded<Notification<TPayload>>(_observeScheduler.Now.UtcTicks, Notification.CreateOnCompleted<TPayload>()));
 
         /// <inheritdoc />
         public void Dispose()
         {
-            subscription?.Dispose();
-            rollingReplaySubject?.Dispose();
+            _subscription?.Dispose();
+            _rollingReplaySubject?.Dispose();
         }
 
         /// <summary>

Src/FluentAssertions.Reactive/RollingReplaySubject.cs

@@ -1,53 +1,100 @@
 using System;
 using System.Collections.Generic;
+using System.Reactive.Disposables;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
 
 namespace FluentAssertions.Reactive
 {
     /// <summary>
-    /// Clearable <see cref="ReplaySubject{T}"/> taken from James World: https://stackoverflow.com/a/28945444/4340541
+    /// Clearable <see cref="ReplaySubject{T}"/> taken from James World: https://gist.github.com/james-world/c46f09f32e2d4f338b07
     /// </summary>
     /// <typeparam name="T"></typeparam>
-    public class RollingReplaySubject<T> : ISubject<T>, IDisposable
+    public class RollingReplaySubject
     {
-        private readonly ReplaySubject<IObservable<T>> subjects;
-        private readonly IObservable<T> concatenatedSubjects;
-        private ISubject<T> currentSubject;
+        protected class NopSubject<TSource> : ISubject<TSource>
+        {
+            public static readonly NopSubject<TSource> Default = new NopSubject<TSource>();
+
+            public void OnCompleted()
+            {
+            }
+
+            public void OnError(Exception error)
+            {
+            }
+
+            public void OnNext(TSource value)
+            {
+            }
+
+            public IDisposable Subscribe(IObserver<TSource> observer)
+            {
+                return Disposable.Empty;
+            }
+        }
+    }
+
+    public class RollingReplaySubject<TSource> : RollingReplaySubject, ISubject<TSource>
+    {
+        private readonly ReplaySubject<IObservable<TSource>> _subjects;
+        private readonly IObservable<TSource> _concatenatedSubjects;
+        private ISubject<TSource> _currentSubject;
+        private readonly object _gate = new object();
 
         public RollingReplaySubject()
         {
-            subjects = new ReplaySubject<IObservable<T>>(1);
-            concatenatedSubjects = subjects.Concat();
-            currentSubject = new ReplaySubject<T>();
-            subjects.OnNext(currentSubject);
+            _subjects = new ReplaySubject<IObservable<TSource>>(1);
+            _concatenatedSubjects = _subjects.Concat();
+            _currentSubject = new ReplaySubject<TSource>();
+            _subjects.OnNext(_currentSubject);
         }
-
+        
         public void Clear()
         {
-            currentSubject.OnCompleted();
-            currentSubject = new ReplaySubject<T>();
-            subjects.OnNext(currentSubject);
+            lock (_gate)
+            {
+                _currentSubject.OnCompleted();
+                _currentSubject = new ReplaySubject<TSource>();
+                _subjects.OnNext(_currentSubject);
+            }
         }
 
-        public void OnNext(T value) => currentSubject.OnNext(value);
+        public void OnNext(TSource value)
+        {
+            lock (_gate)
+            {
+                _currentSubject.OnNext(value);
+            }
+        }
 
-        public void OnError(Exception error) => currentSubject.OnError(error);
+        public void OnError(Exception error)
+        {
+            lock (_gate)
+            {
+                _currentSubject.OnError(error);
+                _currentSubject = NopSubject<TSource>.Default;
+            }
+        }
 
         public void OnCompleted()
         {
-            currentSubject.OnCompleted();
-            subjects.OnCompleted();
-            // a quick way to make the current ReplaySubject unreachable
-            // except to in-flight observers, and not hold up collection
-            currentSubject = new Subject<T>();
+            lock (_gate)
+            {
+                _currentSubject.OnCompleted();
+                _subjects.OnCompleted();
+                _currentSubject = NopSubject<TSource>.Default;
+            }
         }
 
-        public IDisposable Subscribe(IObserver<T> observer) => concatenatedSubjects.Subscribe(observer);
+        public IDisposable Subscribe(IObserver<TSource> observer)
+        {
+            return _concatenatedSubjects.Subscribe(observer);
+        }
 
-        public IEnumerable<T> GetSnapshot()
+        public IEnumerable<TSource> GetSnapshot()
         {
-            var snapshot = new List<T>();
+            var snapshot = new List<TSource>();
             using (this.Subscribe(item => snapshot.Add(item)))
             {
                 // Deliberately empty; subscribing will add everything to the list.
@@ -58,7 +105,7 @@ public IEnumerable<T> GetSnapshot()
         public void Dispose()
         {
             OnCompleted();
-            subjects?.Dispose();
+            _subjects?.Dispose();
         }
     }
 }

Tests/FluentAssertions.Reactive.Specs/ReactiveAssertionSpecs.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Diagnostics;
 using System.Linq;
 using System.Reactive;
 using System.Reactive.Linq;
@@ -50,20 +51,20 @@ public void When_the_expected_number_of_notifications_where_not_pushed_it_should
 
             // assert a single notification
             // Act
-            Action act = () => observer.Should().Push(1, TimeSpan.Zero);
+            Action act = () => observer.Should().Push(1, TimeSpan.FromMilliseconds(1));
             // Assert
             act.Should().Throw<XunitException>().WithMessage(
-                $"Expected observable to push at least 1 notification, but 0 were received within {Formatter.ToString(TimeSpan.Zero)}.");
+                $"Expected observable to push at least 1 notification, but 0 were received within {Formatter.ToString(TimeSpan.FromMilliseconds(1))}.");
             observer.RecordedNotifications.Should().BeEmpty("because no messages have been pushed");
 
             // assert multiple notifications
             scheduler.AdvanceTo(250);
 
             // Act
-            act = () => observer.Should().Push(3, TimeSpan.Zero);
+            act = () => observer.Should().Push(3, TimeSpan.FromMilliseconds(1));
             // Assert
             act.Should().Throw<XunitException>().WithMessage(
-                $"Expected observable to push at least 3 notifications, but 2 were received within {Formatter.ToString(TimeSpan.Zero)}.");
+                $"Expected observable to push at least 3 notifications, but 2 were received within {Formatter.ToString(TimeSpan.FromMilliseconds(1))}.");
             observer.RecordedNotifications.Should().BeEquivalentTo(observable.Messages.Take(2));
         }
 
@@ -117,7 +118,7 @@ public void When_the_observable_is_expected_to_fail_but_does_not_it_should_throw
             using var observer = observable.Observe();
 
             // Act
-            Action act = () => observer.Should().Throw<ArgumentException>(TimeSpan.Zero);
+            Action act = () => observer.Should().Throw<ArgumentException>(TimeSpan.FromMilliseconds(1));
 
             // Assert
             act.Should().Throw<XunitException>().WithMessage(
@@ -151,10 +152,10 @@ public void When_the_observable_is_expected_to_complete_but_does_not_it_should_t
             using var observer = observable.Observe();
 
             // Act
-            Action act = () => observer.Should().Complete(TimeSpan.Zero);
+            Action act = () => observer.Should().Complete(TimeSpan.FromMilliseconds(1));
             // Assert
             act.Should().Throw<XunitException>().WithMessage(
-                $"Expected observable to complete within {Formatter.ToString(TimeSpan.Zero)}, but it did not.");
+                $"Expected observable to complete within {Formatter.ToString(TimeSpan.FromMilliseconds(1))}, but it did not.");
             observer.Error.Should().BeNull();
         }