Implemented change tracking

Author: Erwinvandervalk

src/Config.SqlStreamStore.Tests/StreamStoreConfigurationProviderTests.cs

@@ -1,7 +1,9 @@
-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Primitives;
 using SqlStreamStore;
 using Xunit;
 
@@ -91,16 +93,51 @@ IStreamStore FakeStreamStoreFactory(string providedConnectionString)
         [Fact]
         public async Task Can_reload_when_data_chagnes()
         {
+            // Create a SSS instance with some settings
             var instance = await BuildSteamStoreWithSettings(("setting1", "value1"));
 
             var config = new ConfigurationBuilder()
-                .Add(new StreamStoreConfigurationSource(() => instance))
+                .Add(new StreamStoreConfigurationSource(() => instance)
+                {
+                    SubscribeToChanges = true
+                })
+                .Build();
+
+            await new ConfigRepository(instance).Modify(CancellationToken.None,
+                ("setting1", "modified"));
+
+            await WaitUntil(() => config["setting1"] == "modified");
+
+            Assert.Equal("modified", config["setting1"]);
+
+        }
+
+        [Fact]
+        public async Task Triggers_reload_token_on_change()
+        {
+            // Create a SSS instance with some settings
+            var instance = await BuildSteamStoreWithSettings(("setting1", "value1"));
+
+            var config = new ConfigurationBuilder()
+                .Add(new StreamStoreConfigurationSource(() => instance)
+                {
+                    SubscribeToChanges = true
+                })
                 .Build();
 
+            var tcs = new TaskCompletionSource<bool>();
+            ChangeToken.OnChange(config.GetReloadToken, () => tcs.SetResult(true));
+
+            await new ConfigRepository(instance).Modify(CancellationToken.None,
+                ("setting1", "modified"));
 
+            var noftifiedSettings = await Task.WhenAny(tcs.Task, Task.Delay(TimeSpan.FromSeconds(1)));
 
+            Assert.True(tcs.Task.IsCompleted);
+            Assert.Equal("modified", config["setting1"]);
         }
 
+
         private static async Task<InMemoryStreamStore> BuildSteamStoreWithSettings(params (string key, string value)[] settings)
         {
             var instance = new InMemoryStreamStore();
@@ -111,5 +148,15 @@ await repo.WriteChanges(new ModifiedConfigurationSettings(
 
             return instance;
         }
+
+        private async Task WaitUntil(Func<bool> isTrue, TimeSpan? timeout = null)
+        {
+            var actualTimeout = timeout ?? TimeSpan.FromSeconds(1);
+            for (int i = 0; i < actualTimeout.TotalMilliseconds / 10; i++)
+            {
+                if(isTrue()) break;
+                await Task.Delay(10);
+            }
+        }
     }
 }

src/Config.SqlStreamStore/ConfigRepository.cs

@@ -12,6 +12,11 @@ namespace Config.SqlStreamStore
     public interface IConfigRepository
     {
         Task<ConfigurationSettings> GetLatest(CancellationToken ct);
+        Task<IConfigurationSettings> WriteChanges(IConfigurationSettings settings, CancellationToken ct);
+
+        IDisposable SubscribeToChanges(int version, 
+            ConfigRepository.OnSettingsChanged onSettingsChanged,
+            CancellationToken ct);
     }
 
     public class ConfigRepository : IConfigRepository
@@ -54,6 +59,15 @@ private static async Task<ConfigurationSettings> BuildConfigurationSettingsFromM
             return new ConfigurationSettings(message.StreamVersion, message.CreatedUtc, configChanged.AllSettings);
         }
 
+        public async Task<IConfigurationSettings> Modify(CancellationToken ct,
+            params (string Key, string Value)[] modifications)
+        {
+            var currentData = await GetLatest(ct);
+            var modified = currentData.Modify(modifications);
+            return await WriteChanges(modified, ct);
+        }
+
+
         public async Task<IConfigurationSettings> WriteChanges(IConfigurationSettings settings, CancellationToken ct)
         {
             var changes = (settings as ModifiedConfigurationSettings)?.GetChanges();

src/Config.SqlStreamStore/StreamStoreConfigurationProvider.cs

@@ -2,25 +2,54 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
+using System.Threading.Tasks;
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.Primitives;
 
 namespace Config.SqlStreamStore
 {
     public class StreamStoreConfigurationProvider : ConfigurationProvider
     {
+        private readonly StreamStoreConfigurationSource _source;
         private readonly IConfigRepository _configRepository;
+        private ConfigurationSettings _configurationSettings;
 
-        public StreamStoreConfigurationProvider(IConfigRepository configRepository)
+        public StreamStoreConfigurationProvider(StreamStoreConfigurationSource source,
+            IConfigRepository configRepository)
         {
+            _source = source;
             _configRepository = configRepository;
+
         }
         public override void Load()
         {
-            var settings = _configRepository
+            _configurationSettings = _configRepository
                 .GetLatest(CancellationToken.None).GetAwaiter().GetResult();
 
+            Data = _configurationSettings.ToDictionary(x => x.Key, x => x.Value, StringComparer.OrdinalIgnoreCase);
+
+            if (_source.SubscribeToChanges)
+            {
+                SubscribeToChanges(CancellationToken.None);
+            }
+
+        }
+
+        public IDisposable SubscribeToChanges(CancellationToken ct)
+        {
+            return _configRepository.SubscribeToChanges(
+                version: _configurationSettings?.Version ?? 0, 
+                onSettingsChanged: OnChanged, 
+                ct: CancellationToken.None);
+        }
+
+        private Task OnChanged(ConfigurationSettings settings, CancellationToken ct)
+        {
             Data = settings.ToDictionary(x => x.Key, x => x.Value, StringComparer.OrdinalIgnoreCase);
+
+            OnReload();
+
+            return Task.CompletedTask;
         }
     }
 }

src/Config.SqlStreamStore/StreamStoreConfigurationSource.cs

@@ -12,7 +12,7 @@ public class StreamStoreConfigurationSource : IConfigurationSource
         public delegate IConfigRepository BuildConfigRepository();
         private readonly BuildStreamStoreFromConfig _buildStreamStoreFromConfig;
 
-
+        public bool SubscribeToChanges { get; set; }
 
         private readonly BuildConfigRepository _getConfigRepository;
 
@@ -52,7 +52,7 @@ public IConfigurationProvider Build(IConfigurationBuilder builder)
         {
             if (_getConfigRepository != null)
             {
-                return new StreamStoreConfigurationProvider(_getConfigRepository());
+                return new StreamStoreConfigurationProvider(this, _getConfigRepository());
             }
 
             var innerBuilder = new ConfigurationBuilder();
@@ -66,7 +66,7 @@ public IConfigurationProvider Build(IConfigurationBuilder builder)
 
             var repo = new ConfigRepository(_buildStreamStoreFromConfig(innerBuilder.Build()));
 
-            return new StreamStoreConfigurationProvider(repo);
+            return new StreamStoreConfigurationProvider(this, repo);
 
         }