support opt-in explicit stream validation in read model listeners

condron · condron · commit 68cbcab451f7 · 2024-12-08T11:59:31.000-05:00
diff --git a/src/ReactiveDomain.Foundation/IListener.cs b/src/ReactiveDomain.Foundation/IListener.cs
@@ -16,24 +16,27 @@ public interface IListener : IDisposable
         /// <param name="stream">the exact stream name</param>
         /// <param name="checkpoint">start point to listen from</param>
         /// <param name="blockUntilLive">wait for the is live event from the catchup subscription before returning</param>
+        /// <param name="validateStream">ensure the stream exists on start</param>
         /// <param name="cancelWaitToken">Cancellation token to cancel waiting if blockUntilLive is true</param>
-        void Start(string stream, long? checkpoint = null, bool blockUntilLive = false, CancellationToken cancelWaitToken = default);
+        void Start(string stream, long? checkpoint = null, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default);
         /// <summary>
         /// Starts listening on an aggregate root stream
         /// </summary>
         /// <typeparam name="TAggregate">The type of aggregate</typeparam>
         /// <param name="id">the aggregate id</param>
         /// <param name="checkpoint">start point to listen from</param>
         /// <param name="blockUntilLive">wait for the is live event from the catchup subscription before returning</param>
+        /// <param name="validateStream">ensure the stream exists on start</param>
         /// <param name="cancelWaitToken">Cancellation token to cancel waiting if blockUntilLive is true</param>
-        void Start<TAggregate>(Guid id, long? checkpoint = null, bool blockUntilLive = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource;
+        void Start<TAggregate>(Guid id, long? checkpoint = null, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource;
         /// <summary>
         /// Starts listening on a Aggregate Category Stream
         /// </summary>
         /// <typeparam name="TAggregate">The type of aggregate</typeparam>
         /// <param name="checkpoint">start point to listen from</param>
         /// <param name="blockUntilLive">wait for the is live event from the catchup subscription before returning</param>
+        /// <param name="validateStream">ensure the stream exists on start</param>
         /// <param name="cancelWaitToken">Cancellation token to cancel waiting if blockUntilLive is true</param>
-        void Start<TAggregate>(long? checkpoint = null, bool blockUntilLive = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource;
+        void Start<TAggregate>(long? checkpoint = null, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource;
     }
 }
diff --git a/src/ReactiveDomain.Foundation/StreamStore/QueuedStreamListener.cs b/src/ReactiveDomain.Foundation/StreamStore/QueuedStreamListener.cs
@@ -53,12 +53,12 @@ public void Handle(IMessage @event)
             }
         }
 
-        public override void Start(string streamName, long? checkpoint = null, bool waitUntilLive = false, CancellationToken cancelWaitToken = default(CancellationToken))
+        public override void Start(string streamName, long? checkpoint = null, bool waitUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default(CancellationToken))
         {
             _isLive.Reset();
 
             SyncQueue?.Start();
-            base.Start(streamName, checkpoint, waitUntilLive, cancelWaitToken);
+            base.Start(streamName, checkpoint, waitUntilLive, validateStream, cancelWaitToken);
 
             Interlocked.Exchange(ref _pendingCount, SyncQueue.MessageCount);
             if (Interlocked.Read(ref _pendingCount) <= 0 || SyncQueue.Idle)
diff --git a/src/ReactiveDomain.Foundation/StreamStore/ReadModelBase.cs b/src/ReactiveDomain.Foundation/StreamStore/ReadModelBase.cs
@@ -1,5 +1,6 @@
 ﻿using System;
 using System.Collections.Generic;
+using System.Diagnostics.SymbolStore;
 using System.Linq;
 using System.Threading;
 using ReactiveDomain.Messaging;
@@ -105,7 +106,7 @@ public List<Tuple<string, long>> GetCheckpoint()
         /// <param name="checkpoint">The event to start with.</param>
         /// <param name="blockUntilLive">If true, blocks returning from this method until the listener has caught up.</param>
         /// <param name="cancelWaitToken">Cancellation token to cancel waiting if blockUntilLive is true.</param>
-        public void Start(string stream, long? checkpoint = null, bool blockUntilLive = false, CancellationToken cancelWaitToken = default)
+        public void Start(string stream, long? checkpoint = null, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default)
         {
             if (_getReader != null)
             {
@@ -115,7 +116,7 @@ public void Start(string stream, long? checkpoint = null, bool blockUntilLive =
                     checkpoint = reader.Position ?? checkpoint;
                 }
             }
-            AddNewListener().Start(stream, checkpoint, blockUntilLive, cancelWaitToken);
+            AddNewListener().Start(stream, checkpoint, blockUntilLive, validateStream, cancelWaitToken);
         }
 
         /// <summary>
@@ -126,7 +127,7 @@ public void Start(string stream, long? checkpoint = null, bool blockUntilLive =
         /// <param name="checkpoint">The event to start with.</param>
         /// <param name="blockUntilLive">If true, blocks returning from this method until the listener has caught up.</param>
         /// <param name="cancelWaitToken">Cancellation token to cancel waiting if blockUntilLive is true.</param>
-        public void Start<TAggregate>(Guid id, long? checkpoint = null, bool blockUntilLive = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource
+        public void Start<TAggregate>(Guid id, long? checkpoint = null, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource
         {
             if (_getReader != null)
             {
@@ -136,7 +137,7 @@ public void Start<TAggregate>(Guid id, long? checkpoint = null, bool blockUntilL
                     checkpoint = reader.Position;
                 }
             }
-            AddNewListener().Start<TAggregate>(id, checkpoint, blockUntilLive, cancelWaitToken);
+            AddNewListener().Start<TAggregate>(id, checkpoint, blockUntilLive, validateStream, cancelWaitToken);
         }
 
         /// <summary>
@@ -146,7 +147,7 @@ public void Start<TAggregate>(Guid id, long? checkpoint = null, bool blockUntilL
         /// <param name="checkpoint">The event to start with.</param>
         /// <param name="blockUntilLive">If true, blocks returning from this method until the listener has caught up.</param>
         /// <param name="cancelWaitToken">Cancellation token to cancel waiting if blockUntilLive is true.</param>
-        public void Start<TAggregate>(long? checkpoint = null, bool blockUntilLive = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource
+        public void Start<TAggregate>(long? checkpoint = null, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource
         {
             if (_getReader != null)
             {
@@ -156,7 +157,7 @@ public void Start<TAggregate>(long? checkpoint = null, bool blockUntilLive = fal
                     checkpoint = reader.Position;
                 }
             }
-            AddNewListener().Start<TAggregate>(checkpoint, blockUntilLive, cancelWaitToken);
+            AddNewListener().Start<TAggregate>(checkpoint, blockUntilLive, validateStream, cancelWaitToken);
         }
 
         /// <summary>
diff --git a/src/ReactiveDomain.Foundation/StreamStore/SnapshotReadModel.cs b/src/ReactiveDomain.Foundation/StreamStore/SnapshotReadModel.cs
@@ -17,6 +17,7 @@ protected virtual void Restore(
                                 ReadModelState snapshot,
                                 bool startListeners = true, 
                                 bool block = false,
+                                bool validateStreams = false,
                                 CancellationToken cancelWaitToken = default(CancellationToken)) {
             if(StartingState != null) {
                 throw new InvalidOperationException("ReadModel has already been restored.");
@@ -27,7 +28,7 @@ protected virtual void Restore(
             if (!startListeners || StartingState.Checkpoints == null) return;
 
             foreach (var stream in StartingState.Checkpoints) {
-                Start(stream.Item1,stream.Item2,block, cancelWaitToken);
+                Start(stream.Item1,stream.Item2, block, validateStreams, cancelWaitToken);
             }
         }
 
diff --git a/src/ReactiveDomain.Foundation/StreamStore/StreamListener.cs b/src/ReactiveDomain.Foundation/StreamStore/StreamListener.cs
@@ -79,6 +79,7 @@ public void Start(
             Type tMessage,
             long? checkpoint = null,
             bool blockUntilLive = false,
+            bool validateStream = false,
             CancellationToken cancelWaitToken = default(CancellationToken))
         {
             if (!tMessage.IsSubclassOf(typeof(Event)))
@@ -89,6 +90,7 @@ public void Start(
                _streamNameBuilder.GenerateForEventType(tMessage.Name),
                 checkpoint,
                 blockUntilLive,
+                validateStream,
                 cancelWaitToken);
         }
         /// <summary>
@@ -102,13 +104,15 @@ public void Start(
         public void Start<TAggregate>(
                         long? checkpoint = null,
                         bool blockUntilLive = false,
+                        bool validateStream = false,
                         CancellationToken cancelWaitToken = default(CancellationToken)) where TAggregate : class, IEventSource
         {
 
             Start(
                 _streamNameBuilder.GenerateForCategory(typeof(TAggregate)),
                 checkpoint,
                 blockUntilLive,
+                validateStream,
                 cancelWaitToken);
         }
 
@@ -125,12 +129,14 @@ public void Start<TAggregate>(
                         Guid id,
                         long? checkpoint = null,
                         bool blockUntilLive = false,
+                        bool validateStream = false,
                         CancellationToken cancelWaitToken = default(CancellationToken)) where TAggregate : class, IEventSource
         {
             Start(
                 _streamNameBuilder.GenerateForAggregate(typeof(TAggregate), id),
                 checkpoint,
                 blockUntilLive,
+                validateStream,
                 cancelWaitToken);
         }
 
@@ -146,14 +152,15 @@ public virtual void Start(
                             string streamName,
                             long? checkpoint = null,
                             bool blockUntilLive = false,
+                            bool validateStream = false,
                             CancellationToken cancelWaitToken = default(CancellationToken))
         {
             _liveLock.Reset();
             lock (_startLock)
             {
                 if (_started)
                     throw new InvalidOperationException("Listener already started.");
-                if (!ValidateStreamName(streamName))
+                if (validateStream && !ValidateStreamName(streamName))
                     throw new ArgumentException("Stream not found.", streamName);
                 StreamName = streamName;
                 _subscription =
@@ -204,8 +211,17 @@ public IDisposable SubscribeToStreamFrom(
 
         public bool ValidateStreamName(string streamName)
         {
-            var isValid = _streamStoreConnection.ReadStreamForward(streamName, 0, 1) != null;
-            return isValid;
+            try
+            {
+                var result = _streamStoreConnection.ReadStreamForward(streamName, 0, 1);
+               
+                return result.GetType() == typeof(StreamEventsSlice);
+            }
+            catch (Exception)
+            {
+                return false;
+            }
+
         }
         protected virtual void GotEvent(RecordedEvent recordedEvent)
         {
diff --git a/src/ReactiveDomain.Testing/Specifications/NullListener.cs b/src/ReactiveDomain.Testing/Specifications/NullListener.cs
@@ -11,6 +11,7 @@ namespace ReactiveDomain.Testing
     /// </summary>
     public class NullListener : IListener, ISubscriber
     {
+#pragma warning disable CS1066 // The default value specified will have no effect because it applies to a member that is used in contexts that do not allow optional arguments
 
         private string _stream;
         private long _position;
@@ -34,7 +35,7 @@ public class NullListener : IListener, ISubscriber
         /// </summary>
         /// <param name="name">This parameter is ignored.</param>
         public NullListener(string name = "")
-        {          
+        {
         }
 
         /// <summary>
@@ -52,7 +53,7 @@ public void Dispose()
         /// <param name="checkpoint">The position at which the listener should start.</param>
         /// <param name="blockUntilLive">This parameter is ignored.</param>
         /// <param name="cancelWaitToken">This parameter is ignored.</param>
-        public void Start(string stream, long? checkpoint = null, bool blockUntilLive = false, CancellationToken cancelWaitToken = default)
+        public void Start(string stream, long? checkpoint = null, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default)
         {
             _stream = stream;
             _position = checkpoint ?? 0;
@@ -87,7 +88,7 @@ bool ISubscriber.HasSubscriberFor<T>(bool includeDerived)
         /// <param name="checkpoint">This parameter is ignored.</param>
         /// <param name="blockUntilLive">This parameter is ignored.</param>
         /// <param name="cancelWaitToken">This parameter is ignored.</param>
-        void IListener.Start<TAggregate>(Guid id, long? checkpoint, bool blockUntilLive, CancellationToken cancelWaitToken)
+        void IListener.Start<TAggregate>(Guid id, long? checkpoint, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default)
         {
             _stream = new PrefixedCamelCaseStreamNameBuilder().GenerateForAggregate(typeof(TAggregate), id);
         }
@@ -99,7 +100,7 @@ void IListener.Start<TAggregate>(Guid id, long? checkpoint, bool blockUntilLive,
         /// <param name="checkpoint">This parameter is ignored.</param>
         /// <param name="blockUntilLive">This parameter is ignored.</param>
         /// <param name="cancelWaitToken">This parameter is ignored.</param>
-        void IListener.Start<TAggregate>(long? checkpoint, bool blockUntilLive, CancellationToken cancelWaitToken)
+        void IListener.Start<TAggregate>(long? checkpoint, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default)
         {
             _stream = nameof(TAggregate);
         }
@@ -124,5 +125,6 @@ IDisposable ISubscriber.Subscribe<T>(IHandle<T> handler, bool includeDerived)
         void ISubscriber.Unsubscribe<T>(IHandle<T> handler)
         {
         }
+#pragma warning restore CS1066 // The default value specified will have no effect because it applies to a member that is used in contexts that do not allow optional arguments
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -53,12 +53,12 @@ public void Handle(IMessage @event)`
`53`	`53`	`}`
`54`	`54`	`}`
`55`	`55`
`56`		`- public override void Start(string streamName, long? checkpoint = null, bool waitUntilLive = false, CancellationToken cancelWaitToken = default(CancellationToken))`
	`56`	`+ public override void Start(string streamName, long? checkpoint = null, bool waitUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default(CancellationToken))`
`57`	`57`	`{`
`58`	`58`	`_isLive.Reset();`
`59`	`59`
`60`	`60`	`SyncQueue?.Start();`
`61`		`- base.Start(streamName, checkpoint, waitUntilLive, cancelWaitToken);`
	`61`	`+ base.Start(streamName, checkpoint, waitUntilLive, validateStream, cancelWaitToken);`
`62`	`62`
`63`	`63`	`Interlocked.Exchange(ref _pendingCount, SyncQueue.MessageCount);`
`64`	`64`	`if (Interlocked.Read(ref _pendingCount) <= 0 \|\| SyncQueue.Idle)`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,6 @@`
`1`	`1`	`using System;`
`2`	`2`	`using System.Collections.Generic;`
	`3`	`+using System.Diagnostics.SymbolStore;`
`3`	`4`	`using System.Linq;`
`4`	`5`	`using System.Threading;`
`5`	`6`	`using ReactiveDomain.Messaging;`
`@@ -105,7 +106,7 @@ public List<Tuple<string, long>> GetCheckpoint()`
`105`	`106`	`/// <param name="checkpoint">The event to start with.</param>`
`106`	`107`	`/// <param name="blockUntilLive">If true, blocks returning from this method until the listener has caught up.</param>`
`107`	`108`	`/// <param name="cancelWaitToken">Cancellation token to cancel waiting if blockUntilLive is true.</param>`
`108`		`- public void Start(string stream, long? checkpoint = null, bool blockUntilLive = false, CancellationToken cancelWaitToken = default)`
	`109`	`+ public void Start(string stream, long? checkpoint = null, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default)`
`109`	`110`	`{`
`110`	`111`	`if (_getReader != null)`
`111`	`112`	`{`
`@@ -115,7 +116,7 @@ public void Start(string stream, long? checkpoint = null, bool blockUntilLive =`
`115`	`116`	`checkpoint = reader.Position ?? checkpoint;`
`116`	`117`	`}`
`117`	`118`	`}`
`118`		`- AddNewListener().Start(stream, checkpoint, blockUntilLive, cancelWaitToken);`
	`119`	`+ AddNewListener().Start(stream, checkpoint, blockUntilLive, validateStream, cancelWaitToken);`
`119`	`120`	`}`
`120`	`121`
`121`	`122`	`/// <summary>`
`@@ -126,7 +127,7 @@ public void Start(string stream, long? checkpoint = null, bool blockUntilLive =`
`126`	`127`	`/// <param name="checkpoint">The event to start with.</param>`
`127`	`128`	`/// <param name="blockUntilLive">If true, blocks returning from this method until the listener has caught up.</param>`
`128`	`129`	`/// <param name="cancelWaitToken">Cancellation token to cancel waiting if blockUntilLive is true.</param>`
`129`		`- public void Start<TAggregate>(Guid id, long? checkpoint = null, bool blockUntilLive = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource`
	`130`	`+ public void Start<TAggregate>(Guid id, long? checkpoint = null, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource`
`130`	`131`	`{`
`131`	`132`	`if (_getReader != null)`
`132`	`133`	`{`
`@@ -136,7 +137,7 @@ public void Start<TAggregate>(Guid id, long? checkpoint = null, bool blockUntilL`
`136`	`137`	`checkpoint = reader.Position;`
`137`	`138`	`}`
`138`	`139`	`}`
`139`		`- AddNewListener().Start<TAggregate>(id, checkpoint, blockUntilLive, cancelWaitToken);`
	`140`	`+ AddNewListener().Start<TAggregate>(id, checkpoint, blockUntilLive, validateStream, cancelWaitToken);`
`140`	`141`	`}`
`141`	`142`
`142`	`143`	`/// <summary>`
`@@ -146,7 +147,7 @@ public void Start<TAggregate>(Guid id, long? checkpoint = null, bool blockUntilL`
`146`	`147`	`/// <param name="checkpoint">The event to start with.</param>`
`147`	`148`	`/// <param name="blockUntilLive">If true, blocks returning from this method until the listener has caught up.</param>`
`148`	`149`	`/// <param name="cancelWaitToken">Cancellation token to cancel waiting if blockUntilLive is true.</param>`
`149`		`- public void Start<TAggregate>(long? checkpoint = null, bool blockUntilLive = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource`
	`150`	`+ public void Start<TAggregate>(long? checkpoint = null, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default) where TAggregate : class, IEventSource`
`150`	`151`	`{`
`151`	`152`	`if (_getReader != null)`
`152`	`153`	`{`
`@@ -156,7 +157,7 @@ public void Start<TAggregate>(long? checkpoint = null, bool blockUntilLive = fal`
`156`	`157`	`checkpoint = reader.Position;`
`157`	`158`	`}`
`158`	`159`	`}`
`159`		`- AddNewListener().Start<TAggregate>(checkpoint, blockUntilLive, cancelWaitToken);`
	`160`	`+ AddNewListener().Start<TAggregate>(checkpoint, blockUntilLive, validateStream, cancelWaitToken);`
`160`	`161`	`}`
`161`	`162`
`162`	`163`	`/// <summary>`
Original file line number	Diff line number	Diff line change
`@@ -17,6 +17,7 @@ protected virtual void Restore(`
`17`	`17`	`ReadModelState snapshot,`
`18`	`18`	`bool startListeners = true,`
`19`	`19`	`bool block = false,`
	`20`	`+ bool validateStreams = false,`
`20`	`21`	`CancellationToken cancelWaitToken = default(CancellationToken)) {`
`21`	`22`	`if(StartingState != null) {`
`22`	`23`	`throw new InvalidOperationException("ReadModel has already been restored.");`
`@@ -27,7 +28,7 @@ protected virtual void Restore(`
`27`	`28`	`if (!startListeners \|\| StartingState.Checkpoints == null) return;`
`28`	`29`
`29`	`30`	`foreach (var stream in StartingState.Checkpoints) {`
`30`		`- Start(stream.Item1,stream.Item2,block, cancelWaitToken);`
	`31`	`+ Start(stream.Item1,stream.Item2, block, validateStreams, cancelWaitToken);`
`31`	`32`	`}`
`32`	`33`	`}`
`33`	`34`
Original file line number	Diff line number	Diff line change
`@@ -11,6 +11,7 @@ namespace ReactiveDomain.Testing`
`11`	`11`	`/// </summary>`
`12`	`12`	`public class NullListener : IListener, ISubscriber`
`13`	`13`	`{`
	`14`	`+#pragma warning disable CS1066 // The default value specified will have no effect because it applies to a member that is used in contexts that do not allow optional arguments`
`14`	`15`
`15`	`16`	`private string _stream;`
`16`	`17`	`private long _position;`
`@@ -34,7 +35,7 @@ public class NullListener : IListener, ISubscriber`
`34`	`35`	`/// </summary>`
`35`	`36`	`/// <param name="name">This parameter is ignored.</param>`
`36`	`37`	`public NullListener(string name = "")`
`37`		`- {`
	`38`	`+ {`
`38`	`39`	`}`
`39`	`40`
`40`	`41`	`/// <summary>`
`@@ -52,7 +53,7 @@ public void Dispose()`
`52`	`53`	`/// <param name="checkpoint">The position at which the listener should start.</param>`
`53`	`54`	`/// <param name="blockUntilLive">This parameter is ignored.</param>`
`54`	`55`	`/// <param name="cancelWaitToken">This parameter is ignored.</param>`
`55`		`- public void Start(string stream, long? checkpoint = null, bool blockUntilLive = false, CancellationToken cancelWaitToken = default)`
	`56`	`+ public void Start(string stream, long? checkpoint = null, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default)`
`56`	`57`	`{`
`57`	`58`	`_stream = stream;`
`58`	`59`	`_position = checkpoint ?? 0;`
`@@ -87,7 +88,7 @@ bool ISubscriber.HasSubscriberFor<T>(bool includeDerived)`
`87`	`88`	`/// <param name="checkpoint">This parameter is ignored.</param>`
`88`	`89`	`/// <param name="blockUntilLive">This parameter is ignored.</param>`
`89`	`90`	`/// <param name="cancelWaitToken">This parameter is ignored.</param>`
`90`		`- void IListener.Start<TAggregate>(Guid id, long? checkpoint, bool blockUntilLive, CancellationToken cancelWaitToken)`
	`91`	`+ void IListener.Start<TAggregate>(Guid id, long? checkpoint, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default)`
`91`	`92`	`{`
`92`	`93`	`_stream = new PrefixedCamelCaseStreamNameBuilder().GenerateForAggregate(typeof(TAggregate), id);`
`93`	`94`	`}`
`@@ -99,7 +100,7 @@ void IListener.Start<TAggregate>(Guid id, long? checkpoint, bool blockUntilLive,`
`99`	`100`	`/// <param name="checkpoint">This parameter is ignored.</param>`
`100`	`101`	`/// <param name="blockUntilLive">This parameter is ignored.</param>`
`101`	`102`	`/// <param name="cancelWaitToken">This parameter is ignored.</param>`
`102`		`- void IListener.Start<TAggregate>(long? checkpoint, bool blockUntilLive, CancellationToken cancelWaitToken)`
	`103`	`+ void IListener.Start<TAggregate>(long? checkpoint, bool blockUntilLive = false, bool validateStream = false, CancellationToken cancelWaitToken = default)`
`103`	`104`	`{`
`104`	`105`	`_stream = nameof(TAggregate);`
`105`	`106`	`}`
`@@ -124,5 +125,6 @@ IDisposable ISubscriber.Subscribe<T>(IHandle<T> handler, bool includeDerived)`
`124`	`125`	`void ISubscriber.Unsubscribe<T>(IHandle<T> handler)`
`125`	`126`	`{`
`126`	`127`	`}`
	`128`	`+#pragma warning restore CS1066 // The default value specified will have no effect because it applies to a member that is used in contexts that do not allow optional arguments`
`127`	`129`	`}`
`128`	`130`	`}`