Normalize listener subscriptions on missing/future streams

src/ReactiveDomain.Foundation.Tests/Domain/AggregateRootEntityTests.cs

@@ -1,9 +1,12 @@
-using System;
+using ReactiveDomain.Foundation.Tests.Domain;
+using System;
 using Xunit;
 
 // ReSharper disable once CheckNamespace
-namespace ReactiveDomain.Domain.Tests {
-    namespace AggregateRootEntityTests {
+namespace ReactiveDomain.Domain.Tests
+{
+    namespace AggregateRootEntityTests
+    {
         public class AnyInstance {
             // IEventSource behavior
 

src/ReactiveDomain.Foundation.Tests/Domain/CapturingRoute.cs

@@ -1,4 +1,4 @@
-namespace ReactiveDomain.Domain.Tests
+namespace ReactiveDomain.Foundation.Tests.Domain
 {
     public class CapturingRoute
     {

src/ReactiveDomain.Foundation.Tests/Domain/CorrelatedAggregate.cs

@@ -3,7 +3,7 @@
 using ReactiveDomain.Messaging;
 
 // ReSharper disable once CheckNamespace
-namespace ReactiveDomain.Foundation.Tests
+namespace ReactiveDomain.Foundation.Tests.Domain
 {
     public class CorrelatedAggregate : AggregateRoot
     {

src/ReactiveDomain.Foundation.Tests/Domain/with_correlated_aggregate.cs

@@ -3,7 +3,7 @@
 using Xunit;
 
 // ReSharper disable once CheckNamespace
-namespace ReactiveDomain.Foundation.Tests
+namespace ReactiveDomain.Foundation.Tests.Domain
 {
     // ReSharper disable once InconsistentNaming
     public class with_correlated_aggregate
@@ -28,7 +28,7 @@ public void can_raise_correlated_events_from_constructor_source()
                                                 },
                                                  e =>{
                                                    if( e is CorrelatedAggregate.CorrelatedEvent @event){
-                                                        Assert.Equal(_command.CorrelationId, @event.CorrelationId);                                                        
+                                                        Assert.Equal(_command.CorrelationId, @event.CorrelationId);
                                                        }
                                                    else{
                                                         throw new Exception("wrong event");

src/ReactiveDomain.Foundation.Tests/Domain/with_correlated_repository.cs

@@ -4,7 +4,7 @@
 using Xunit;
 
 // ReSharper disable once CheckNamespace
-namespace ReactiveDomain.Foundation.Tests
+namespace ReactiveDomain.Foundation.Tests.Domain
 {
     [Collection(nameof(EmbeddedStreamStoreConnectionCollection))]
     // ReSharper disable once InconsistentNaming
@@ -83,36 +83,40 @@ public void correlation_and_causation_are_injected()
 
             var raisedEvents = ((IEventSource)agg).TakeEvents();
             Assert.Collection(raisedEvents,
-                @event => {
+                @event =>
+                {
                     var created = @event as CorrelatedAggregate.Created;
                     Assert.NotNull(created);
-                    Assert.Equal(command1.MsgId,created.CausationId);
-                    Assert.Equal(command1.CorrelationId,created.CorrelationId);
+                    Assert.Equal(command1.MsgId, created.CausationId);
+                    Assert.Equal(command1.CorrelationId, created.CorrelationId);
                 },
-                @event => {
+                @event =>
+                {
                     var corrEvent = @event as CorrelatedAggregate.CorrelatedEvent;
                     Assert.NotNull(corrEvent);
-                    Assert.Equal(command1.MsgId,corrEvent.CausationId);
-                    Assert.Equal(command1.CorrelationId,corrEvent.CorrelationId);
+                    Assert.Equal(command1.MsgId, corrEvent.CausationId);
+                    Assert.Equal(command1.CorrelationId, corrEvent.CorrelationId);
                 },
-                @event => {
+                @event =>
+                {
                     var corrEvent = @event as CorrelatedAggregate.CorrelatedEvent;
                     Assert.NotNull(corrEvent);
-                    Assert.Equal(command1.MsgId,corrEvent.CausationId);
-                    Assert.Equal(command1.CorrelationId,corrEvent.CorrelationId);
+                    Assert.Equal(command1.MsgId, corrEvent.CausationId);
+                    Assert.Equal(command1.CorrelationId, corrEvent.CorrelationId);
                 });
             var command2 = MessageBuilder.New(() => new TestCommands.Command2());
-            ((ICorrelatedEventSource) agg).Source = command2;
-            
+            ((ICorrelatedEventSource)agg).Source = command2;
+
             agg.RaiseCorrelatedEvent();
             raisedEvents = ((IEventSource)agg).TakeEvents();
             Assert.Collection(raisedEvents,
-                    @event => {
-                    var corrEvent = @event as CorrelatedAggregate.CorrelatedEvent;
-                    Assert.NotNull(corrEvent);
-                    Assert.Equal(command2.MsgId,corrEvent.CausationId);
-                    Assert.Equal(command2.CorrelationId,corrEvent.CorrelationId);
-                });
+                    @event =>
+                    {
+                        var corrEvent = @event as CorrelatedAggregate.CorrelatedEvent;
+                        Assert.NotNull(corrEvent);
+                        Assert.Equal(command2.MsgId, corrEvent.CausationId);
+                        Assert.Equal(command2.CorrelationId, corrEvent.CorrelationId);
+                    });
         }
     }
 }

src/ReactiveDomain.Foundation.Tests/Domain/with_repository.cs

@@ -7,7 +7,7 @@
 using Xunit;
 
 // ReSharper disable once CheckNamespace
-namespace ReactiveDomain.Foundation.Tests
+namespace ReactiveDomain.Foundation.Tests.Domain
 {
     [Collection(nameof(EmbeddedStreamStoreConnectionCollection))]
     // ReSharper disable once InconsistentNaming
@@ -25,20 +25,20 @@ public with_repository(StreamStoreConnectionFixture fixture)
                 _streamNameBuilder,
                 fixture.Connection,
                 _serializer,
-                GetPolicyUserId);           
+                GetPolicyUserId);
         }
         private Guid GetPolicyUserId() { return _policyUserId; }
         private Guid _policyUserId = Guid.NewGuid();
         [Fact]
         public void policy_user_id_is_saved()
         {
-           
+
             var id = Guid.NewGuid();
-            var agg = new TestAggregate(id);     
+            var agg = new TestAggregate(id);
             var userId = Guid.NewGuid();
             _policyUserId = userId;
             _repo.Save(agg);
-           
+
 
             agg.RaiseBy(10);
             var nextUser = Guid.NewGuid();
@@ -52,13 +52,13 @@ public void policy_user_id_is_saved()
             var md = newAggregate.ReadMetadatum<AuditRecord>();
             Assert.NotNull(md);
             Assert.Equal(userId, md.PolicyUserId);
-            Assert.True((DateTime.UtcNow - md.EventDateUTC) < TimeSpan.FromSeconds(5)); // just check for a fresh timestamp, not trying to test built in .Now 
+            Assert.True(DateTime.UtcNow - md.EventDateUTC < TimeSpan.FromSeconds(5)); // just check for a fresh timestamp, not trying to test built in .Now 
             var evt2 = slice.Events[1];
             var incremented = _serializer.Deserialize(evt2) as TestAggregateMessages.Increment;
             var md2 = incremented.ReadMetadatum<AuditRecord>();
             Assert.NotNull(md2);
             Assert.Equal(nextUser, md2.PolicyUserId);
-            Assert.True((DateTime.UtcNow - md2.EventDateUTC) < TimeSpan.FromSeconds(5)); // just check for a fresh timestamp, not trying to test built in .Now 
+            Assert.True(DateTime.UtcNow - md2.EventDateUTC < TimeSpan.FromSeconds(5)); // just check for a fresh timestamp, not trying to test built in .Now 
 #if NETCOREAPP
             Assert.NotEqual(md.EventDateUTC, md2.EventDateUTC);
 #endif

src/ReactiveDomain.Foundation.Tests/ReactiveDomain.Foundation.Tests.csproj

@@ -15,9 +15,9 @@
   </ItemGroup>
   <ItemGroup>
     <PackageReference Include="Newtonsoft.Json" Version="13.0.3" />
-    <PackageReference Include="xunit" Version="2.9.0" />
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.10.0" />
-    <PackageReference Include="xunit.runner.console" Version="2.9.0">
+    <PackageReference Include="xunit" Version="2.9.2" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.12.0" />
+    <PackageReference Include="xunit.runner.console" Version="2.9.2">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
     </PackageReference>

src/ReactiveDomain.Foundation.Tests/SynchronizedStreamListenerTests/Common/CommonHelpers.cs → src/ReactiveDomain.Foundation.Tests/StreamListenerTests/Common/CommonHelpers.cs

@@ -2,7 +2,7 @@
 using ReactiveDomain.Messaging;
 using ReactiveDomain.Testing;
 
-namespace ReactiveDomain.Foundation.Tests.SynchronizedStreamListenerTests.Common
+namespace ReactiveDomain.Foundation.Tests.StreamListenerTests.Common
 {
     internal static class CommonHelpers
     {

src/ReactiveDomain.Foundation.Tests/SynchronizedStreamListenerTests/Common/TestStreamNameBuilder.cs → src/ReactiveDomain.Foundation.Tests/StreamListenerTests/Common/TestStreamNameBuilder.cs

@@ -1,13 +1,13 @@
 using System;
 
-namespace ReactiveDomain.Foundation.Tests.SynchronizedStreamListenerTests.Common
+namespace ReactiveDomain.Foundation.Tests.StreamListenerTests.Common
 {
     /// <summary>
     /// Generate stream names for testing. 
     /// </summary>
     /// <remarks>
     /// todo:
-    /// The use of the extra Guid doesn't match the generation by the <see cref="ReactiveDomain.Foundation.IStreamNameBuilder"/>
+    /// The use of the extra Guid doesn't match the generation by the <see cref="IStreamNameBuilder"/>
     /// and the checks for existing category and event streams fail. <see cref="PrefixedCamelCaseStreamNameBuilderTests"/>
     /// tests the stream name generation. So switching to the PrefixedCamelCaseStreamNameBuilder. Leaving these here so Chris can
     /// agree and remove or correct me.

src/ReactiveDomain.Foundation.Tests/SynchronizedStreamListenerTests/when_using_listener_start_with_aggregate_and_guid.cs → src/ReactiveDomain.Foundation.Tests/StreamListenerTests/when_using_listener_start_with_aggregate_and_guid.cs

@@ -3,31 +3,33 @@
 using ReactiveDomain.Messaging;
 using ReactiveDomain.Messaging.Bus;
 using ReactiveDomain.Testing;
-using ReactiveDomain.Foundation.Tests.SynchronizedStreamListenerTests.Common;
 using Xunit;
+using ReactiveDomain.Foundation.Tests.StreamListenerTests.Common;
 
-namespace ReactiveDomain.Foundation.Tests.SynchronizedStreamListenerTests {
+namespace ReactiveDomain.Foundation.Tests.StreamListenerTests
+{
     // ReSharper disable once InconsistentNaming
     [Collection(nameof(EmbeddedStreamStoreConnectionCollection))]
     public class when_using_listener_start_with_aggregate_and_guid
     {
         private readonly IStreamNameBuilder _streamNameBuilder = new PrefixedCamelCaseStreamNameBuilder();
         private readonly IEventSerializer _eventSerializer = new JsonMessageSerializer();
 
-        public when_using_listener_start_with_aggregate_and_guid(StreamStoreConnectionFixture fixture){
+        public when_using_listener_start_with_aggregate_and_guid(StreamStoreConnectionFixture fixture)
+        {
             var conn = fixture.Connection;
             conn.Connect();
 
-            // Build an origin stream to which the the events are appended. We are testing this stream directly
+            // Build an origin stream to which the events are appended. We are testing this stream directly
             var originalStreamGuid = Guid.NewGuid();
             var originStreamName = _streamNameBuilder.GenerateForAggregate(typeof(AggregateIdTestAggregate), originalStreamGuid);
 
-            
+
             // Drop an event into the stream testAggregate-guid
             var result = conn.AppendToStream(
-                                originStreamName, 
-                                ExpectedVersion.NoStream, 
-                                null, 
+                                originStreamName,
+                                ExpectedVersion.NoStream,
+                                null,
                                 _eventSerializer.Serialize(new TestEvent()));
             Assert.True(result.NextExpectedVersion == 0);
 
@@ -49,16 +51,18 @@ public when_using_listener_start_with_aggregate_and_guid(StreamStoreConnectionFi
         [Fact]
         public void can_get_events_from_aggregate_id_stream()
         {
-            AssertEx.IsOrBecomesTrue(() => Interlocked.Read(ref _testEventCount) == 1,3000);
+            AssertEx.IsOrBecomesTrue(() => Interlocked.Read(ref _testEventCount) == 1, 3000);
         }
 
-        private void Handle(IMessage message) {
+        private void Handle(IMessage message)
+        {
             dynamic evt = message;
-            if (evt is TestEvent) {
+            if (evt is TestEvent)
+            {
                 Interlocked.Increment(ref _testEventCount);
             }
 
         }
-        public class AggregateIdTestAggregate:EventDrivenStateMachine{}
+        public class AggregateIdTestAggregate : EventDrivenStateMachine { }
     }
 }

src/ReactiveDomain.Foundation.Tests/SynchronizedStreamListenerTests/when_using_listener_start_with_category_aggregate.cs → src/ReactiveDomain.Foundation.Tests/StreamListenerTests/when_using_listener_start_with_category_aggregate.cs

@@ -3,10 +3,11 @@
 using ReactiveDomain.Messaging;
 using ReactiveDomain.Messaging.Bus;
 using ReactiveDomain.Testing;
-using ReactiveDomain.Foundation.Tests.SynchronizedStreamListenerTests.Common;
 using Xunit;
+using ReactiveDomain.Foundation.Tests.StreamListenerTests.Common;
 
-namespace ReactiveDomain.Foundation.Tests.SynchronizedStreamListenerTests {
+namespace ReactiveDomain.Foundation.Tests.StreamListenerTests
+{
     // ReSharper disable once InconsistentNaming
     [Collection(nameof(EmbeddedStreamStoreConnectionCollection))]
     public class when_using_listener_start_with_category_aggregate
@@ -24,38 +25,41 @@ public when_using_listener_start_with_category_aggregate(StreamStoreConnectionFi
 
             // Drop an event into the stream testAggregate-guid
             var result = conn.AppendToStream(
-                aggStream, 
-                ExpectedVersion.NoStream, 
-                null, 
+                aggStream,
+                ExpectedVersion.NoStream,
+                null,
                 _eventSerializer.Serialize(new TestEvent()));
             Assert.True(result.NextExpectedVersion == 0);
 
             //wait for the projection to be written.
             CommonHelpers.WaitForStream(conn, categoryStream);
-            
+
             // Now set up the projection listener, and start it. 
             var listener = new QueuedStreamListener(
                 "category listener",
                 conn,
                streamNameBuilder,
                 new JsonMessageSerializer());
-            listener.EventStream.Subscribe<TestEvent>(new AdHocHandler<TestEvent>(Handle));
+            listener.EventStream.Subscribe(new AdHocHandler<TestEvent>(Handle));
             listener.Start<AggregateCategoryTestAggregate>();
         }
 
         private long _testEventCount;
         [Fact]
-        public void can_get_events_from_category_projection() {
+        public void can_get_events_from_category_projection()
+        {
             AssertEx.IsOrBecomesTrue(() => Interlocked.Read(ref _testEventCount) == 1, 4000);
         }
 
-        private void Handle(IMessage message) {
+        private void Handle(IMessage message)
+        {
             dynamic evt = message;
-            if (evt is TestEvent) {
+            if (evt is TestEvent)
+            {
                 Interlocked.Increment(ref _testEventCount);
             }
 
         }
-        public class AggregateCategoryTestAggregate:EventDrivenStateMachine{}
+        public class AggregateCategoryTestAggregate : EventDrivenStateMachine { }
     }
 }

src/ReactiveDomain.Foundation.Tests/SynchronizedStreamListenerTests/when_using_listener_start_with_custom_stream_not_synched.cs → src/ReactiveDomain.Foundation.Tests/StreamListenerTests/when_using_listener_start_with_custom_stream_not_synched.cs

@@ -3,10 +3,11 @@
 using ReactiveDomain.Messaging;
 using ReactiveDomain.Messaging.Bus;
 using ReactiveDomain.Testing;
-using ReactiveDomain.Foundation.Tests.SynchronizedStreamListenerTests.Common;
 using Xunit;
+using ReactiveDomain.Foundation.Tests.StreamListenerTests.Common;
 
-namespace ReactiveDomain.Foundation.Tests.SynchronizedStreamListenerTests {
+namespace ReactiveDomain.Foundation.Tests.StreamListenerTests
+{
     // ReSharper disable once InconsistentNaming
     [Collection(nameof(EmbeddedStreamStoreConnectionCollection))]
     public class when_using_listener_start_with_custom_stream_not_synched
@@ -23,11 +24,11 @@ public when_using_listener_start_with_custom_stream_not_synched(StreamStoreConne
             var originStreamName = $"testStream-{Guid.NewGuid():N}";
 
             // Generate event and save it to the custom stream
-            
+
             var result = fixture.Connection.AppendToStream(
-                                               originStreamName, 
-                                               ExpectedVersion.NoStream, 
-                                               null, 
+                                               originStreamName,
+                                               ExpectedVersion.NoStream,
+                                               null,
                                                _eventSerializer.Serialize(new TestEvent()));
             Assert.True(result.NextExpectedVersion == 0);
 
@@ -46,7 +47,7 @@ public when_using_listener_start_with_custom_stream_not_synched(StreamStoreConne
         private long _testEventCount;
 
         [Fact]
-        public void can_get_events_from_custom_stream() 
+        public void can_get_events_from_custom_stream()
         {
             AssertEx.IsOrBecomesTrue(() => Interlocked.Read(ref _testEventCount) == 1, 3000);
         }