An Observable Cache

Cache.cs

using System;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Reactive.Subjects;

namespace Stuff
{
    public class Cache<T> : IDisposable
    {
        private readonly IConnectableObservable<T> _source;
        private IDisposable _connection;

        public Cache(IObservable<T> source)
        {
            _source = Observable.Create<T>(o => source.Subscribe(o))
                                .Replay(1, Scheduler.Immediate);
        }

        public IObservable<T> GetValue()
        {
            return Observable.Create<T>(o =>
                                         {
                                             var subscription = _source.Subscribe(o);
                                             _connection = _source.Connect();
                                             return subscription;
                                         });
        }

        public void Clear()
        {
            if (_connection != null)
            {
                _connection.Dispose();
                _connection = null;
            }
        }

        public void Dispose()
        {
            Clear();
        }
    }
}

CacheTests.cs

using System;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using Microsoft.Reactive.Testing;
using NUnit.Framework;

namespace Stuff
{
    [TestFixture]
    public class CacheTests
    {
        private TestScheduler _testScheduler;
        private int _calculationStartedCount;
        private TimeSpan _calculationDuration;

        [SetUp]
        public void Setup()
        {
            _testScheduler = new TestScheduler();
            _calculationDuration = TimeSpan.FromMinutes(1);
            
            _calculationStartedCount = 0;
        }

        private Cache<string> GetCalculator()
        {
            var calculation = Observable.Create<string>(o => {
                                                                _calculationStartedCount++;

                                                                return Observable.Timer(_calculationDuration, _testScheduler)
                                                                                 .Select(_ => "Hello World!")
                                                                                 .Subscribe(o);
                                                             });
                
            return new Cache<string>(calculation);
        }

        [Test]
        public void Can_instantiate_Calculator()
        {
            var calculator = GetCalculator();
            Assert.IsNotNull(calculator);
        }

        [Test]
        public void Calling_GetResult_zero_times_should_not_call_calculation()
        {
            var calculator = GetCalculator();
            _testScheduler.Start();

            Assert.That(_calculationStartedCount, Is.EqualTo(0));
        }

        [Test]
        public void Calling_GetResult_first_time_should_call_calculation()
        {
            // ARRANGE
            var calculator = GetCalculator();
            
            // ACT
            string result = null;
            calculator.GetValue().Subscribe(r => result = r);

            bool assertsPerformed = false;
            int calculationsStartedCount = 0;
            string result2 = null;

            _testScheduler.Schedule(TimeSpan.FromSeconds(2).Ticks, (_, __) =>
                                {
                                    assertsPerformed = true;
                                    calculationsStartedCount = _calculationStartedCount;
                                    result2 = result;
                                });
            _testScheduler.Start();

            // ASSERT
            Assert.IsTrue(assertsPerformed);
            Assert.That(calculationsStartedCount, Is.EqualTo(1));
            Assert.IsNull(result2);
        }

        [Test]
        public void Calling_GetResult_should_return_result_after_calculation()
        {
            // ARRANGE
            var calculator = GetCalculator();

            // ACT
            string result = null;
            calculator.GetValue().Subscribe(r => result = r);
            _testScheduler.AdvanceBy(_calculationDuration.Add(TimeSpan.FromSeconds(1)).Ticks);

            // ASSERT
            Assert.That(_calculationStartedCount, Is.EqualTo(1));
            Assert.IsNotNull(result);
        }

        [Test]
        public void Calling_GetResult_two_times_before_calculation_finished_should_perform_calculation_once()
        {
            // ARRANGE
            var calculator = GetCalculator();

            // ACT
            string result1 = null;
            string result2 = null;
            calculator.GetValue().Subscribe(r => result1 = r);

            _testScheduler.AdvanceBy(TimeSpan.FromSeconds(3).Ticks);

            calculator.GetValue().Subscribe(r => result2 = r);
            
            _testScheduler.Start();

            // ASSERT
            Assert.That(_calculationStartedCount, Is.EqualTo(1));
            Assert.IsNotNull(result1);
            Assert.IsNotNull(result2);
        }

        [Test]
        public void Calling_GetResult_after_calculation_finished_should_not_perform_calculation_again()
        {
            // ARRANGE
            var calculator = GetCalculator();

            string result1 = null;
            string result2 = null;
            calculator.GetValue().Subscribe(r => result1 = r);

            _testScheduler.Start();

            // ACT
            calculator.GetValue().Subscribe(r => result2 = r);

            _testScheduler.Start();

            // ASSERT
            Assert.That(_calculationStartedCount, Is.EqualTo(1));
            Assert.IsNotNull(result1);
            Assert.IsNotNull(result2);
        }

        [Test]
        public void After_Calling_GetResult_Calling_ClearResult_and_GetResult_should_perform_calculation_again()
        {
            // ARRANGE
            var calculator = GetCalculator();

            calculator.GetValue().Subscribe();
            _testScheduler.Start();

            // ACT
            calculator.Clear();

            string result = null;
            calculator.GetValue().Subscribe(r => result = r);
            _testScheduler.Start();

            // ASSERT
            Assert.That(_calculationStartedCount, Is.EqualTo(2));
            Assert.IsNotNull(result);
        }

        [Test]
        public void After_Calling_GetResult_x2_after_a_ClearResult_should_perform_calculation_once_more()
        {
            // ARRANGE
            var calculator = GetCalculator();

            calculator.GetValue().Subscribe();
            _testScheduler.Start();

            // ACT
            calculator.Clear();

            string result1 = null;
            string result2 = null;
            calculator.GetValue().Subscribe(r => result1 = r);
            _testScheduler.AdvanceBy(TimeSpan.FromSeconds(2).Ticks);
            
            calculator.GetValue().Subscribe(r => result2 = r);
            _testScheduler.Start();

            // ASSERT
            Assert.That(_calculationStartedCount, Is.EqualTo(2));
            Assert.IsNotNull(result1);
            Assert.IsNotNull(result2);
        }
    }
}