Merge pull request #208 from buildersoftio/v3/bug/207

Refactor IStream to single generic parameter interface

Author: eneshoxha

src/Cortex.Streams.Mediator/Behaviors/StreamEmittingCommandBehavior.cs

@@ -14,7 +14,7 @@ namespace Cortex.Streams.Mediator.Behaviors
     public class StreamEmittingCommandBehavior<TCommand, TResult> : ICommandPipelineBehavior<TCommand, TResult>
         where TCommand : ICommand<TResult>
     {
-        private readonly IStream<CommandExecutionEvent<TCommand, TResult>, CommandExecutionEvent<TCommand, TResult>> _stream;
+        private readonly IStream<CommandExecutionEvent<TCommand, TResult>> _stream;
         private readonly bool _emitBeforeExecution;
         private readonly bool _emitAfterExecution;
 
@@ -25,7 +25,7 @@ public class StreamEmittingCommandBehavior<TCommand, TResult> : ICommandPipeline
         /// <param name="emitBeforeExecution">If true, emit an event before command execution.</param>
         /// <param name="emitAfterExecution">If true, emit an event after command execution.</param>
         public StreamEmittingCommandBehavior(
-            IStream<CommandExecutionEvent<TCommand, TResult>, CommandExecutionEvent<TCommand, TResult>> stream,
+            IStream<CommandExecutionEvent<TCommand, TResult>> stream,
             bool emitBeforeExecution = false,
             bool emitAfterExecution = true)
         {

src/Cortex.Streams.Mediator/Behaviors/StreamEmittingNotificationBehavior.cs

@@ -18,7 +18,7 @@ namespace Cortex.Streams.Mediator.Behaviors
     public class StreamEmittingNotificationBehavior<TNotification> : INotificationPipelineBehavior<TNotification>
         where TNotification : INotification
     {
-        private readonly IStream<NotificationEvent<TNotification>, NotificationEvent<TNotification>> _stream;
+        private readonly IStream<NotificationEvent<TNotification>> _stream;
         private readonly bool _emitBeforeHandling;
         private readonly bool _emitAfterHandling;
 
@@ -29,7 +29,7 @@ public class StreamEmittingNotificationBehavior<TNotification> : INotificationPi
         /// <param name="emitBeforeHandling">If true, emit an event before notification handling.</param>
         /// <param name="emitAfterHandling">If true, emit an event after notification handling.</param>
         public StreamEmittingNotificationBehavior(
-            IStream<NotificationEvent<TNotification>, NotificationEvent<TNotification>> stream,
+            IStream<NotificationEvent<TNotification>> stream,
             bool emitBeforeHandling = false,
             bool emitAfterHandling = true)
         {

src/Cortex.Streams.Mediator/DependencyInjection/ServiceCollectionExtensions.cs

@@ -22,7 +22,7 @@ public static class ServiceCollectionExtensions
         /// <returns>The service collection for chaining.</returns>
         public static IServiceCollection AddStreamEmittingNotificationHandler<TNotification>(
             this IServiceCollection services,
-            Func<IServiceProvider, IStream<TNotification, TNotification>> streamFactory,
+            Func<IServiceProvider, IStream<TNotification>> streamFactory,
             Action<TNotification, Exception> errorHandler = null)
             where TNotification : INotification
         {
@@ -47,7 +47,7 @@ public static IServiceCollection AddStreamEmittingNotificationHandler<TNotificat
         /// <returns>The service collection for chaining.</returns>
         public static IServiceCollection AddTransformingStreamNotificationHandler<TNotification, TStreamInput>(
             this IServiceCollection services,
-            Func<IServiceProvider, IStream<TStreamInput, TStreamInput>> streamFactory,
+            Func<IServiceProvider, IStream<TStreamInput>> streamFactory,
             Func<TNotification, TStreamInput> transformer,
             Action<TNotification, Exception> errorHandler = null)
             where TNotification : INotification

src/Cortex.Streams.Mediator/Handlers/StreamBackedStreamQueryHandler.cs

@@ -17,15 +17,15 @@ namespace Cortex.Streams.Mediator.Handlers
     public abstract class StreamBackedStreamQueryHandler<TQuery, TResult> : IStreamQueryHandler<TQuery, TResult>
         where TQuery : IStreamQuery<TResult>
     {
-        private readonly IStream<TResult, TResult> _stream;
+        private readonly IStream<TResult> _stream;
         private readonly int _channelCapacity;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="StreamBackedStreamQueryHandler{TQuery, TResult}"/> class.
         /// </summary>
         /// <param name="stream">The Cortex Stream to read data from.</param>
         /// <param name="channelCapacity">The capacity of the internal channel buffer. Default is 100.</param>
-        protected StreamBackedStreamQueryHandler(IStream<TResult, TResult> stream, int channelCapacity = 100)
+        protected StreamBackedStreamQueryHandler(IStream<TResult> stream, int channelCapacity = 100)
         {
             _stream = stream ?? throw new ArgumentNullException(nameof(stream));
             _channelCapacity = channelCapacity;

src/Cortex.Streams.Mediator/Handlers/StreamEmittingNotificationHandler.cs

@@ -13,7 +13,7 @@ namespace Cortex.Streams.Mediator.Handlers
     public class StreamEmittingNotificationHandler<TNotification> : INotificationHandler<TNotification>
         where TNotification : INotification
     {
-        private readonly IStream<TNotification, TNotification> _stream;
+        private readonly IStream<TNotification> _stream;
         private readonly Action<TNotification, Exception> _errorHandler;
 
         /// <summary>
@@ -22,7 +22,7 @@ public class StreamEmittingNotificationHandler<TNotification> : INotificationHan
         /// <param name="stream">The stream to emit notifications to.</param>
         /// <param name="errorHandler">Optional handler for errors during emission.</param>
         public StreamEmittingNotificationHandler(
-            IStream<TNotification, TNotification> stream,
+            IStream<TNotification> stream,
             Action<TNotification, Exception> errorHandler = null)
         {
             _stream = stream ?? throw new ArgumentNullException(nameof(stream));
@@ -62,7 +62,7 @@ public async Task Handle(TNotification notification, CancellationToken cancellat
     public class TransformingStreamNotificationHandler<TNotification, TStreamInput> : INotificationHandler<TNotification>
         where TNotification : INotification
     {
-        private readonly IStream<TStreamInput, TStreamInput> _stream;
+        private readonly IStream<TStreamInput> _stream;
         private readonly Func<TNotification, TStreamInput> _transformer;
         private readonly Action<TNotification, Exception> _errorHandler;
 
@@ -73,7 +73,7 @@ public class TransformingStreamNotificationHandler<TNotification, TStreamInput>
         /// <param name="transformer">A function to transform notifications into stream input.</param>
         /// <param name="errorHandler">Optional handler for errors during emission.</param>
         public TransformingStreamNotificationHandler(
-            IStream<TStreamInput, TStreamInput> stream,
+            IStream<TStreamInput> stream,
             Func<TNotification, TStreamInput> transformer,
             Action<TNotification, Exception> errorHandler = null)
         {

src/Cortex.Streams/Abstractions/IBranchInfo.cs

@@ -0,0 +1,15 @@
+namespace Cortex.Streams.Abstractions
+{
+    /// <summary>
+    /// Provides information about a branch in the stream processing pipeline.
+    /// This is a non-generic interface that allows accessing branch metadata
+    /// without exposing internal type parameters.
+    /// </summary>
+    public interface IBranchInfo
+    {
+        /// <summary>
+        /// Gets the name of the branch.
+        /// </summary>
+        string BranchName { get; }
+    }
+}

src/Cortex.Streams/Abstractions/IFanOutBuilder.cs

@@ -126,6 +126,6 @@ public interface IFanOutBuilder<TIn, TCurrent>
         /// </summary>
         /// <returns>The built stream instance ready to be started.</returns>
         /// <exception cref="InvalidOperationException">Thrown when no sinks have been configured.</exception>
-        IStream<TIn, TCurrent> Build();
+        IStream<TIn> Build();
     }
 }

src/Cortex.Streams/Abstractions/ISinkBuilder.cs

@@ -4,13 +4,13 @@
     /// Provides a method to build the stream after adding a sink.
     /// </summary>
     /// <typeparam name="TIn">The type of the initial input to the stream.</typeparam>
-    /// <typeparam name="TCurrent">The current type of data in the stream.</typeparam>
+    /// <typeparam name="TCurrent">The current type of data in the stream (internal use only).</typeparam>
     public interface ISinkBuilder<TIn, TCurrent>
     {
         /// <summary>
         /// Builds the stream and returns a stream instance that can be started and stopped.
         /// </summary>
         /// <returns>A stream instance.</returns>
-        IStream<TIn, TCurrent> Build();
+        IStream<TIn> Build();
     }
 }

src/Cortex.Streams/Abstractions/IStream.cs

@@ -1,13 +1,13 @@
 using Cortex.States;
-using Cortex.Streams.Operators;
+using Cortex.Streams.Abstractions;
 using Cortex.Streams.Performance;
 using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 
 namespace Cortex.Streams
 {
-    public interface IStream<TIn, TCurrent>
+    public interface IStream<TIn>
     {
         /// <summary>
         /// Start the stream processing.
@@ -66,7 +66,7 @@ public interface IStream<TIn, TCurrent>
 
         StreamStatuses GetStatus();
 
-        IReadOnlyDictionary<string, BranchOperator<TCurrent>> GetBranches();
+        IReadOnlyDictionary<string, IBranchInfo> GetBranches();
 
         TStateStore GetStateStoreByName<TStateStore>(string name) where TStateStore : IDataStore;
         IEnumerable<TStateStore> GetStateStoresByType<TStateStore>() where TStateStore : IDataStore;

src/Cortex.Streams/Abstractions/IStreamBuilder.cs

@@ -107,7 +107,7 @@ public interface IStreamBuilder<TIn, TCurrent>
         /// Builds the stream
         /// </summary>
         /// <returns></returns>
-        IStream<TIn, TCurrent> Build();
+        IStream<TIn> Build();
 
 
         /// <summary>