implemented config watching

Author: Erwinvandervalk

src/Config.SqlStreamStore.Tests/ConfigRepositoryTests.cs

@@ -1,3 +1,4 @@
+using System;
 using System.Threading;
 using System.Threading.Tasks;
 using SqlStreamStore;
@@ -73,6 +74,31 @@ private async Task<ConfigurationSettings> SaveSettings(IConfigurationSettings se
             return await _configRepository.GetLatest(CancellationToken.None);
         }
 
+        [Fact]
+        public async Task Can_subscribe_to_changes()
+        {
+            var settings = await SaveSettings(BuildNewSettings());
+
+            var tcs = new TaskCompletionSource<ConfigurationSettings>();
+
+            Task OnSettingsChanged(ConfigurationSettings configurationSettings, CancellationToken ct)
+            {
+                tcs.SetResult(configurationSettings);
+                return Task.CompletedTask;
+            }
+
+            var subscription = _configRepository.SubscribeToChanges(settings.Version, OnSettingsChanged,
+                ct: CancellationToken.None);
+
+            var modified = await _configRepository.WriteChanges(settings.Modify(("setting1", "newValue")), CancellationToken.None);
+
+            var noftifiedSettings = await Task.WhenAny(tcs.Task, Task.Delay(TimeSpan.FromSeconds(1)));
+
+            Assert.True(tcs.Task.IsCompleted);
+            Assert.Equal(modified, tcs.Task.Result);
+        }
+
+
         private static IConfigurationSettings BuildNewSettings()
         {
             var settings = new ModifiedConfigurationSettings(

src/Config.SqlStreamStore/ConfigRepository.cs

@@ -5,6 +5,7 @@
 using Newtonsoft.Json;
 using SqlStreamStore;
 using SqlStreamStore.Streams;
+using SqlStreamStore.Subscriptions;
 
 namespace Config.SqlStreamStore
 {
@@ -15,6 +16,8 @@ public interface IConfigRepository
 
     public class ConfigRepository : IConfigRepository
     {
+        public delegate Task OnSettingsChanged(ConfigurationSettings settings, CancellationToken ct);
+
         private readonly IStreamStore _streamStore;
         private readonly StreamId _streamId;
 
@@ -38,11 +41,17 @@ public async Task<ConfigurationSettings> GetLatest(CancellationToken ct)
             }
 
             var lastMessage = lastPage.Messages.First();
-            var data = await lastMessage.GetJsonData(ct);
+            return await BuildConfigurationSettingsFromMessage(lastMessage, ct);
+        }
+
+        private static async Task<ConfigurationSettings> BuildConfigurationSettingsFromMessage(
+            StreamMessage message, CancellationToken ct)
+        {
+            var data = await message.GetJsonData(ct);
 
             var configChanged = JsonConvert.DeserializeObject<ConfigChanged>(data);
 
-            return new ConfigurationSettings(lastMessage.StreamVersion, lastMessage.CreatedUtc, configChanged.AllSettings);
+            return new ConfigurationSettings(message.StreamVersion, message.CreatedUtc, configChanged.AllSettings);
         }
 
         public async Task<IConfigurationSettings> WriteChanges(IConfigurationSettings settings, CancellationToken ct)
@@ -61,6 +70,56 @@ public async Task<IConfigurationSettings> WriteChanges(IConfigurationSettings se
 
             return new ConfigurationSettings(result.CurrentVersion, null, changes.AllSettings);
         }
-        
+
+        public IDisposable SubscribeToChanges(int version, 
+            OnSettingsChanged onSettingsChanged,
+            CancellationToken ct)
+        {
+            IStreamSubscription subscription = null;
+
+
+            async Task StreamMessageReceived(IStreamSubscription _, StreamMessage streamMessage,
+                CancellationToken cancellationToken)
+            {
+                var settings = await BuildConfigurationSettingsFromMessage(streamMessage, ct);
+                await onSettingsChanged(settings, ct);
+            };
+
+            void SubscriptionDropped(IStreamSubscription _, SubscriptionDroppedReason reason,
+                Exception exception = null)
+            {
+                if (reason != SubscriptionDroppedReason.Disposed)
+                {
+                    SetupSubscription();
+                }
+            };
+
+            void SetupSubscription()
+            {
+                subscription = _streamStore.SubscribeToStream(
+                    streamId: _streamId,
+                    continueAfterVersion: version,
+                    streamMessageReceived: StreamMessageReceived,
+                    subscriptionDropped: SubscriptionDropped);
+            }
+
+            SetupSubscription();
+
+            return subscription;
+        }
+
+        private class DelegateDisposable : IDisposable
+        {
+            public readonly Action Action;
+            public DelegateDisposable(Action action)
+            {
+                Action = action;
+            }
+
+            public void Dispose()
+            {
+                Action();
+            }
+        }
     }
 }

src/Config.SqlStreamStore/StreamStoreConfigurationSource.cs

@@ -6,12 +6,13 @@ namespace Config.SqlStreamStore
 {
     public class StreamStoreConfigurationSource : IConfigurationSource
     {
-        private readonly BuildStreamStoreFromConfig _buildStreamStoreFromConfig;
 
         public delegate IStreamStore BuildStreamStoreFromConnectionString(string connectionString);
         public delegate IStreamStore BuildStreamStoreFromConfig(IConfigurationRoot configurationRoot);
-
         public delegate IConfigRepository BuildConfigRepository();
+        private readonly BuildStreamStoreFromConfig _buildStreamStoreFromConfig;
+
+
 
         private readonly BuildConfigRepository _getConfigRepository;
 
@@ -69,6 +70,5 @@ public IConfigurationProvider Build(IConfigurationBuilder builder)
 
         }
 
-
     }
 }