From d060fb0d1bd7a57b5314ab202e1408f1a2569e47 Mon Sep 17 00:00:00 2001 From: Enes Hoxha Date: Wed, 28 Jan 2026 22:16:58 +0100 Subject: [PATCH] Simplify StreamBuilder API to use a single type parameter Refactored the stream builder API to remove the second type parameter (TCurrent) from IInitialStreamBuilder and StreamBuilder. Stream creation now starts with StreamBuilder.CreateNewStream("Name"), returning IInitialStreamBuilder. All builder methods now operate on TIn as the initial/current type. Updated all usages, extension methods, tests, and documentation to use the new API. This change makes the API more intuitive, reduces boilerplate, and improves usability while preserving type safety and flexibility. --- README.md | 8 +-- .../InitialStreamBuilderMediatorExtensions.cs | 18 +++---- .../Abstractions/IInitialStreamBuilder.cs | 14 ++++-- src/Cortex.Streams/StreamBuilder.cs | 49 ++++++++++-------- .../Streams/Tests/ErrorHandlingTests.cs | 50 +++++++++---------- .../Streams/Tests/FlatMapOperatorTests.cs | 10 ++-- .../Tests/SessionWindowOperatorTests.cs | 2 +- .../Tests/SlidingWindowOperatorTests.cs | 2 +- .../Streams/Tests/StreamBuilderTests.cs | 4 +- .../Streams/Tests/StreamIntegrationTests.cs | 4 +- .../Streams/Tests/TelemetryTests.cs | 20 ++++---- .../Tests/TumblingWindowOperatorTests.cs | 2 +- .../StreamBuilderMediatorExtensionsTests.cs | 14 +++--- 13 files changed, 104 insertions(+), 93 deletions(-) diff --git a/README.md b/README.md index 3638cd9..f6fa9ee 100644 --- a/README.md +++ b/README.md @@ -181,7 +181,8 @@ Cortex Data Framework makes it easy to set up and run real-time data processing ### 1. Creating a Stream ```csharp -var stream = StreamBuilder.CreateNewStream("ExampleStream") +var stream = StreamBuilder.CreateNewStream("ExampleStream") + .Stream() .Map(x => x * 2) .Filter(x => x > 10) .Sink(Console.WriteLine) @@ -203,9 +204,10 @@ Console.WriteLine(stateStore.Get("key1")); ```csharp var telemetryProvider = new OpenTelemetryProvider(); -var stream = StreamBuilder +var stream = StreamBuilder .CreateNewStream("TelemetryStream") .WithTelemetry(telemetryProvider) + .Stream() .Map(x => x * 2) .Sink(Console.WriteLine) .Build(); @@ -239,7 +241,7 @@ public class ClickEvent static void Main(string[] args) { // Build the stream - var stream = StreamBuilder.CreateNewStream("ClickStream") + var stream = StreamBuilder.CreateNewStream("ClickStream") .Stream() .Filter(e => !string.IsNullOrEmpty(e.PageUrl)) .GroupBySilently( diff --git a/src/Cortex.Streams.Mediator/Extensions/InitialStreamBuilderMediatorExtensions.cs b/src/Cortex.Streams.Mediator/Extensions/InitialStreamBuilderMediatorExtensions.cs index 38886ac..3fbc9cd 100644 --- a/src/Cortex.Streams.Mediator/Extensions/InitialStreamBuilderMediatorExtensions.cs +++ b/src/Cortex.Streams.Mediator/Extensions/InitialStreamBuilderMediatorExtensions.cs @@ -15,21 +15,20 @@ public static class InitialStreamBuilderMediatorExtensions /// Starts a stream using a Mediator streaming query as the source. /// /// The initial input type of the stream. - /// The current type of data in the stream (same as TIn for initial builders). /// The type of streaming query. /// The initial stream builder instance. /// The mediator instance. /// The streaming query to execute. /// Optional handler for errors during query execution. /// A stream builder for further configuration. - public static IStreamBuilder StreamFromQuery( - this IInitialStreamBuilder builder, + public static IStreamBuilder StreamFromQuery( + this IInitialStreamBuilder builder, IMediator mediator, TQuery query, Action errorHandler = null) - where TQuery : IStreamQuery + where TQuery : IStreamQuery { - var sourceOperator = new MediatorStreamQuerySourceOperator( + var sourceOperator = new MediatorStreamQuerySourceOperator( mediator, query, errorHandler); @@ -42,21 +41,20 @@ public static IStreamBuilder StreamFromQuery /// The initial input type of the stream. - /// The current type of data in the stream. /// The type of streaming query. /// The initial stream builder instance. /// The mediator instance. /// A factory function to create the streaming query. /// Optional handler for errors during query execution. /// A stream builder for further configuration. - public static IStreamBuilder StreamFromQueryFactory( - this IInitialStreamBuilder builder, + public static IStreamBuilder StreamFromQueryFactory( + this IInitialStreamBuilder builder, IMediator mediator, Func queryFactory, Action errorHandler = null) - where TQuery : IStreamQuery + where TQuery : IStreamQuery { - var sourceOperator = new MediatorStreamQueryFactorySourceOperator( + var sourceOperator = new MediatorStreamQueryFactorySourceOperator( mediator, queryFactory, errorHandler); diff --git a/src/Cortex.Streams/Abstractions/IInitialStreamBuilder.cs b/src/Cortex.Streams/Abstractions/IInitialStreamBuilder.cs index 32bd119..571424d 100644 --- a/src/Cortex.Streams/Abstractions/IInitialStreamBuilder.cs +++ b/src/Cortex.Streams/Abstractions/IInitialStreamBuilder.cs @@ -5,14 +5,18 @@ namespace Cortex.Streams.Abstractions { - public interface IInitialStreamBuilder + /// + /// Initial builder interface for creating a stream processing pipeline. + /// + /// The type of the initial input to the stream. + public interface IInitialStreamBuilder { /// /// Start the stream inside the application, in-app streaming /// /// /// - IStreamBuilder Stream(); + IStreamBuilder Stream(); /// /// Start configuring the Stream @@ -20,7 +24,7 @@ public interface IInitialStreamBuilder /// Type of the Source Operator /// /// - IStreamBuilder Stream(ISourceOperator sourceOperator); + IStreamBuilder Stream(ISourceOperator sourceOperator); /// /// Configure Telemetry for the Stream @@ -28,7 +32,7 @@ public interface IInitialStreamBuilder /// Telemetry provider like OpenTelemetryProvider /// /// - IInitialStreamBuilder WithTelemetry(ITelemetryProvider telemetryProvider); + IInitialStreamBuilder WithTelemetry(ITelemetryProvider telemetryProvider); /// @@ -36,7 +40,7 @@ public interface IInitialStreamBuilder /// /// Execution options controlling error handling strategy and callbacks. /// The initial builder for chaining. - IInitialStreamBuilder WithErrorHandling(StreamExecutionOptions executionOptions); + IInitialStreamBuilder WithErrorHandling(StreamExecutionOptions executionOptions); } } diff --git a/src/Cortex.Streams/StreamBuilder.cs b/src/Cortex.Streams/StreamBuilder.cs index bd2a614..b24d902 100644 --- a/src/Cortex.Streams/StreamBuilder.cs +++ b/src/Cortex.Streams/StreamBuilder.cs @@ -9,12 +9,29 @@ namespace Cortex.Streams { + /// + /// Entry point for creating a stream processing pipeline. + /// + /// The type of the initial input to the stream. + public static class StreamBuilder + { + /// + /// Creates a new stream with the specified name. + /// + /// The name of the stream. + /// An initial stream builder. + public static IInitialStreamBuilder CreateNewStream(string name) + { + return new StreamBuilder(name); + } + } + /// /// Builds a stream processing pipeline with optional branches. /// /// The type of the initial input to the stream. /// The current type of data in the stream. - public class StreamBuilder : IInitialStreamBuilder, IStreamBuilder + internal class StreamBuilder : IInitialStreamBuilder, IStreamBuilder { private readonly string _name; private IOperator _firstOperator; @@ -29,12 +46,12 @@ public class StreamBuilder : IInitialStreamBuilder - private StreamBuilder(string name) + internal StreamBuilder(string name) { _name = name; } - private StreamBuilder(string name, IOperator firstOperator, IOperator lastOperator, bool sourceAdded, ITelemetryProvider telemetryProvider = null, StreamExecutionOptions executionOptions = null) + internal StreamBuilder(string name, IOperator firstOperator, IOperator lastOperator, bool sourceAdded, ITelemetryProvider telemetryProvider = null, StreamExecutionOptions executionOptions = null) { _name = name; _firstOperator = firstOperator; @@ -44,16 +61,6 @@ private StreamBuilder(string name, IOperator firstOperator, IOperator lastOperat _executionOptions = executionOptions ?? StreamExecutionOptions.Default; } - /// - /// Creates a new stream with the specified name. - /// - /// The name of the stream. - /// An initial stream builder. - public static IInitialStreamBuilder CreateNewStream(string name) - { - return new StreamBuilder(name); - } - /// /// Creates a new stream with the specified name. /// @@ -61,7 +68,7 @@ public static IInitialStreamBuilder CreateNewStream(string name) /// The first operator in the pipeline /// The last operator in the pipeline /// An initial stream builder. - public static IStreamBuilder CreateNewStream(string name, IOperator firstOperator, IOperator lastOperator) + internal static IStreamBuilder CreateNewStream(string name, IOperator firstOperator, IOperator lastOperator) { return new StreamBuilder(name, firstOperator, lastOperator, false, null); } @@ -163,14 +170,14 @@ public ISinkBuilder Sink(ISinkOperator sinkOperator) /// Type of the Source Operator /// /// - public IStreamBuilder Stream(ISourceOperator sourceOperator) + IStreamBuilder IInitialStreamBuilder.Stream(ISourceOperator sourceOperator) { if (_sourceAdded) { throw new InvalidOperationException("Source operator already added."); } - var sourceAdapter = new SourceOperatorAdapter(sourceOperator); + var sourceAdapter = new SourceOperatorAdapter(sourceOperator); if (_firstOperator == null) { @@ -183,7 +190,7 @@ public IStreamBuilder Stream(ISourceOperator sourceOper } _sourceAdded = true; - return this; // Returns IStreamBuilder + return (IStreamBuilder)(object)this; } /// @@ -191,7 +198,7 @@ public IStreamBuilder Stream(ISourceOperator sourceOper /// /// /// - public IStreamBuilder Stream() + IStreamBuilder IInitialStreamBuilder.Stream() { // In memory source added. if (_sourceAdded) @@ -200,7 +207,7 @@ public IStreamBuilder Stream() } _sourceAdded = true; - return this; // Returns IStreamBuilder + return (IStreamBuilder)(object)this; } @@ -375,7 +382,7 @@ public IStreamBuilder> Aggregate>(_name, _firstOperator, _lastOperator, _sourceAdded, _telemetryProvider, _executionOptions); } - public IInitialStreamBuilder WithTelemetry(ITelemetryProvider telemetryProvider) + IInitialStreamBuilder IInitialStreamBuilder.WithTelemetry(ITelemetryProvider telemetryProvider) { _telemetryProvider = telemetryProvider; return this; @@ -723,7 +730,7 @@ public IStreamBuilder> AdvancedSessionWindow return new StreamBuilder>(_name, _firstOperator, _lastOperator, _sourceAdded, _telemetryProvider, _executionOptions); } - public IInitialStreamBuilder WithErrorHandling(StreamExecutionOptions executionOptions) + IInitialStreamBuilder IInitialStreamBuilder.WithErrorHandling(StreamExecutionOptions executionOptions) { _executionOptions = executionOptions ?? StreamExecutionOptions.Default; _executionOptions.StreamName = _name; diff --git a/src/Cortex.Tests/Streams/Tests/ErrorHandlingTests.cs b/src/Cortex.Tests/Streams/Tests/ErrorHandlingTests.cs index 89f4567..fdbd7a0 100644 --- a/src/Cortex.Tests/Streams/Tests/ErrorHandlingTests.cs +++ b/src/Cortex.Tests/Streams/Tests/ErrorHandlingTests.cs @@ -29,7 +29,7 @@ public void SkipStrategy_ContinuesProcessingAfterError() ErrorHandlingStrategy = ErrorHandlingStrategy.Skip }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("SkipStrategyTest") .WithErrorHandling(executionOptions) .Stream() @@ -63,7 +63,7 @@ public void SkipStrategy_InFilterOperator_SkipsOnPredicateError() ErrorHandlingStrategy = ErrorHandlingStrategy.Skip }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("SkipFilterTest") .WithErrorHandling(executionOptions) .Stream() @@ -97,7 +97,7 @@ public void SkipStrategy_InSinkOperator_SkipsFailedSink() ErrorHandlingStrategy = ErrorHandlingStrategy.Skip }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("SkipSinkTest") .WithErrorHandling(executionOptions) .Stream() @@ -131,7 +131,7 @@ public void SkipStrategy_InFlatMapOperator_SkipsFailedTransformation() ErrorHandlingStrategy = ErrorHandlingStrategy.Skip }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("SkipFlatMapTest") .WithErrorHandling(executionOptions) .Stream() @@ -171,7 +171,7 @@ public void RetryStrategy_RetriesFailedOperation() RetryDelay = TimeSpan.Zero }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("RetryTest") .WithErrorHandling(executionOptions) .Stream() @@ -207,7 +207,7 @@ public void RetryStrategy_StopsGracefully_WhenMaxRetriesExceeded() RetryDelay = TimeSpan.Zero }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("RetryExceededTest") .WithErrorHandling(executionOptions) .Stream() @@ -245,7 +245,7 @@ public void RetryStrategy_RespectsRetryDelay() }; var attemptCount = 0; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("RetryDelayTest") .WithErrorHandling(executionOptions) .Stream() @@ -288,7 +288,7 @@ public void StopStrategy_GracefullyStopsStreamAndStopsProcessing() ErrorHandlingStrategy = ErrorHandlingStrategy.Stop }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("StopTest") .WithErrorHandling(executionOptions) .Stream() @@ -321,7 +321,7 @@ public void StopStrategy_StopsStreamAfterError() ErrorHandlingStrategy = ErrorHandlingStrategy.Stop }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("StopGracefulTest") .WithErrorHandling(executionOptions) .Stream() @@ -355,7 +355,7 @@ public void StopStrategy_StopsStreamAfterError() public void RethrowStrategy_PropagatesOriginalException() { // Arrange - No error handling configured means Rethrow - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("RethrowTest") .Stream() .Map(x => @@ -382,7 +382,7 @@ public void NoneStrategy_BehavesLikeRethrow() ErrorHandlingStrategy = ErrorHandlingStrategy.None }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("NoneTest") .WithErrorHandling(executionOptions) .Stream() @@ -423,7 +423,7 @@ public void CustomErrorHandler_CanDecidePerError() } }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("CustomHandlerTest") .WithErrorHandling(executionOptions) .Stream() @@ -462,7 +462,7 @@ public void CustomErrorHandler_ReceivesCorrectContext() } }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("ContextTest") .WithErrorHandling(executionOptions) .Stream() @@ -506,7 +506,7 @@ public void CustomErrorHandler_CanRetryWithAttemptTracking() }; var attemptCount = 0; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("AttemptTrackingTest") .WithErrorHandling(executionOptions) .Stream() @@ -540,7 +540,7 @@ public void CustomErrorHandler_CanForceStop() OnError = ctx => ErrorHandlingDecision.Stop }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("ForceStopTest") .WithErrorHandling(executionOptions) .Stream() @@ -592,7 +592,7 @@ public void StopStrategy_StopsStreamOnError() ErrorHandlingStrategy = ErrorHandlingStrategy.Stop }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("OperatorNameTest") .WithErrorHandling(executionOptions) .Stream() @@ -634,7 +634,7 @@ public void ErrorHandling_PropagatesAcrossOperatorChain() } }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("PropagationTest") .WithErrorHandling(executionOptions) .Stream() @@ -689,7 +689,7 @@ public void ErrorHandling_HandlesNullInput() }; var processedItems = new List(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("NullInputTest") .WithErrorHandling(executionOptions) .Stream() @@ -714,7 +714,7 @@ public void ErrorHandling_WorksWithMultipleStreams() var options1 = new StreamExecutionOptions { ErrorHandlingStrategy = ErrorHandlingStrategy.Skip }; var options2 = new StreamExecutionOptions { ErrorHandlingStrategy = ErrorHandlingStrategy.Stop }; - var stream1 = StreamBuilder + var stream1 = StreamBuilder .CreateNewStream("Stream1") .WithErrorHandling(options1) .Stream() @@ -726,7 +726,7 @@ public void ErrorHandling_WorksWithMultipleStreams() .Sink(x => results1.Add(x)) .Build(); - var stream2 = StreamBuilder + var stream2 = StreamBuilder .CreateNewStream("Stream2") .WithErrorHandling(options2) .Stream() @@ -766,7 +766,7 @@ public void ErrorHandling_RetryWithZeroMaxRetries_StopsGracefully() MaxRetries = 0 // No retries allowed }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("ZeroRetriesTest") .WithErrorHandling(executionOptions) .Stream() @@ -800,7 +800,7 @@ public void ErrorHandling_StopStrategy_StopsStreamGracefully() ErrorHandlingStrategy = ErrorHandlingStrategy.Stop }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("StackTraceTest") .WithErrorHandling(executionOptions) .Stream() @@ -832,7 +832,7 @@ public void ErrorHandling_WorksWithComplexPipeline() ErrorHandlingStrategy = ErrorHandlingStrategy.Skip }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("ComplexPipelineTest") .WithErrorHandling(executionOptions) .Stream() @@ -884,7 +884,7 @@ public async Task ErrorHandling_WorksWithAsyncEmit() ErrorHandlingStrategy = ErrorHandlingStrategy.Skip }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("AsyncEmitTest") .WithErrorHandling(executionOptions) .Stream() @@ -1045,7 +1045,7 @@ public async Task ErrorHandling_IsThreadSafe_UnderConcurrentEmits() } }; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("ThreadSafetyTest") .WithErrorHandling(executionOptions) .Stream() diff --git a/src/Cortex.Tests/Streams/Tests/FlatMapOperatorTests.cs b/src/Cortex.Tests/Streams/Tests/FlatMapOperatorTests.cs index 22ba542..772e9e2 100644 --- a/src/Cortex.Tests/Streams/Tests/FlatMapOperatorTests.cs +++ b/src/Cortex.Tests/Streams/Tests/FlatMapOperatorTests.cs @@ -28,7 +28,7 @@ public void Stream_FlatMap_SplitsInputIntoMultipleOutputs() // Build the stream: // Start a stream without a dedicated source, we will just Emit into it. - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("TestStream") .Stream() .FlatMap(line => line.Split(' ')) // Use FlatMap to split a sentence into words @@ -56,7 +56,7 @@ public void Stream_FlatMap_EmptyResult_EmitsNoOutput() // Arrange var collectingSink = new CollectingSink(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("EmptyResultStream") .Stream() .FlatMap(num => new int[0]) // Always empty @@ -80,7 +80,7 @@ public void Stream_FlatMap_NullResult_TreatedAsEmpty() // Arrange var collectingSink = new CollectingSink(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("NullResultStream") .Stream() .FlatMap(num => null) // Always null @@ -104,7 +104,7 @@ public void Stream_FlatMap_ExceptionInFunction_BubblesUp() // Arrange var collectingSink = new CollectingSink(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("ExceptionStream") .Stream() .FlatMap(num => throw new InvalidOperationException("Test exception")) @@ -126,7 +126,7 @@ public void Stream_FlatMap_SingleOutputEmittedForEachInput() // Arrange var collectingSink = new CollectingSink(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("SingleOutputStream") .Stream() .FlatMap(line => new[] { line.ToUpper() }) // One-to-one mapping but via flatmap diff --git a/src/Cortex.Tests/Streams/Tests/SessionWindowOperatorTests.cs b/src/Cortex.Tests/Streams/Tests/SessionWindowOperatorTests.cs index 0ce2116..47b98e4 100644 --- a/src/Cortex.Tests/Streams/Tests/SessionWindowOperatorTests.cs +++ b/src/Cortex.Tests/Streams/Tests/SessionWindowOperatorTests.cs @@ -289,7 +289,7 @@ public void SessionWindowOperator_IntegrationWithStreamBuilder() var inactivityGap = TimeSpan.FromSeconds(2); var emittedResults = new List>(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("Test Session Window Stream") .Stream() .SessionWindow( diff --git a/src/Cortex.Tests/Streams/Tests/SlidingWindowOperatorTests.cs b/src/Cortex.Tests/Streams/Tests/SlidingWindowOperatorTests.cs index 30556fe..9288882 100644 --- a/src/Cortex.Tests/Streams/Tests/SlidingWindowOperatorTests.cs +++ b/src/Cortex.Tests/Streams/Tests/SlidingWindowOperatorTests.cs @@ -210,7 +210,7 @@ public void SlidingWindowOperator_IntegrationWithStreamBuilder() var slideInterval = TimeSpan.FromSeconds(1); var emittedResults = new List>(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("Test Sliding Window Stream") .Stream() .SlidingWindow( diff --git a/src/Cortex.Tests/Streams/Tests/StreamBuilderTests.cs b/src/Cortex.Tests/Streams/Tests/StreamBuilderTests.cs index 1ca856e..4c7513b 100644 --- a/src/Cortex.Tests/Streams/Tests/StreamBuilderTests.cs +++ b/src/Cortex.Tests/Streams/Tests/StreamBuilderTests.cs @@ -7,7 +7,7 @@ public void StreamBuilder_CreatesAndRunsStreamCorrectly() { // Arrange var receivedData = new List(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("TestStream") .Stream() .Map(x => x * 2) @@ -31,7 +31,7 @@ public void StreamBuilder_CreatesAndRunsStreamCorrectly() public void Build_ShouldCreateStreamSuccessfully() { // Arrange - var builder = StreamBuilder.CreateNewStream("TestStream") + var builder = StreamBuilder.CreateNewStream("TestStream") .Stream() .Map(x => x * 2) .Filter(x => x > 5); diff --git a/src/Cortex.Tests/Streams/Tests/StreamIntegrationTests.cs b/src/Cortex.Tests/Streams/Tests/StreamIntegrationTests.cs index 4569942..6d9b8d8 100644 --- a/src/Cortex.Tests/Streams/Tests/StreamIntegrationTests.cs +++ b/src/Cortex.Tests/Streams/Tests/StreamIntegrationTests.cs @@ -8,7 +8,7 @@ public void FullPipeline_ShouldProcessDataCorrectly() // Arrange string result = null; - var stream = StreamBuilder.CreateNewStream("TestStream") + var stream = StreamBuilder.CreateNewStream("TestStream") .Stream() .Filter(x => x > 5) .Map(x => x * 2) @@ -29,7 +29,7 @@ public void Pipeline_ShouldFilterOutInvalidData() // Arrange string result = null; - var stream = StreamBuilder.CreateNewStream("TestStream") + var stream = StreamBuilder.CreateNewStream("TestStream") .Stream() .Filter(x => x > 10) .Sink(x => result = $"Result: {x}") diff --git a/src/Cortex.Tests/Streams/Tests/TelemetryTests.cs b/src/Cortex.Tests/Streams/Tests/TelemetryTests.cs index 2f619fb..31ada7c 100644 --- a/src/Cortex.Tests/Streams/Tests/TelemetryTests.cs +++ b/src/Cortex.Tests/Streams/Tests/TelemetryTests.cs @@ -200,7 +200,7 @@ public void Stream_WorksWithoutTelemetry() { // Arrange var receivedData = new List(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("TestStreamWithoutTelemetry") .Stream() .Map(x => x * 2) @@ -225,7 +225,7 @@ public void Stream_WorksWithNullTelemetryProvider() { // Arrange var receivedData = new List(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("TestStreamNullTelemetry") .WithTelemetry(null!) .Stream() @@ -254,7 +254,7 @@ public void MapOperator_WithTelemetry_RecordsMetrics() var (mockProvider, state) = CreateMockTelemetryProvider(); int result = 0; - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("MapTelemetryTest") .WithTelemetry(mockProvider.Object) .Stream() @@ -311,7 +311,7 @@ public void FilterOperator_WithTelemetry_RecordsMetrics() var (mockProvider, state) = CreateMockTelemetryProvider(); var receivedData = new List(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("FilterTelemetryTest") .WithTelemetry(mockProvider.Object) .Stream() @@ -410,7 +410,7 @@ public void FlatMapOperator_WithTelemetry_RecordsMetrics() var (mockProvider, state) = CreateMockTelemetryProvider(); var receivedData = new List(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("FlatMapTelemetryTest") .WithTelemetry(mockProvider.Object) .Stream() @@ -444,7 +444,7 @@ public void GroupByKeyOperator_WithTelemetry_RecordsMetrics() var (mockProvider, state) = CreateMockTelemetryProvider(); var receivedGroups = new List>>(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("GroupByTelemetryTest") .WithTelemetry(mockProvider.Object) .Stream() @@ -479,7 +479,7 @@ public void AggregateOperator_WithTelemetry_RecordsMetrics() var (mockProvider, state) = CreateMockTelemetryProvider(); var receivedAggregates = new List>(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("AggregateTelemetryTest") .WithTelemetry(mockProvider.Object) .Stream() @@ -516,7 +516,7 @@ public void BranchOperator_WithTelemetry_RecordsMetrics() var branch1Data = new List(); var branch2Data = new List(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("BranchTelemetryTest") .WithTelemetry(mockProvider.Object) .Stream() @@ -560,7 +560,7 @@ public void Telemetry_PropagatesThroughEntirePipeline() var (mockProvider, state) = CreateMockTelemetryProvider(); var receivedData = new List(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("E2ETelemetryTest") .WithTelemetry(mockProvider.Object) .Stream() @@ -709,7 +709,7 @@ public void Telemetry_IsThreadSafe() var (mockProvider, state) = CreateMockTelemetryProvider(); var receivedData = new System.Collections.Concurrent.ConcurrentBag(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("ThreadSafeTelemetryTest") .WithTelemetry(mockProvider.Object) .Stream() diff --git a/src/Cortex.Tests/Streams/Tests/TumblingWindowOperatorTests.cs b/src/Cortex.Tests/Streams/Tests/TumblingWindowOperatorTests.cs index 1156d4a..7403623 100644 --- a/src/Cortex.Tests/Streams/Tests/TumblingWindowOperatorTests.cs +++ b/src/Cortex.Tests/Streams/Tests/TumblingWindowOperatorTests.cs @@ -248,7 +248,7 @@ public void TumblingWindowOperator_IntegrationWithStreamBuilder() var windowSize = TimeSpan.FromSeconds(2); var emittedResults = new List>(); - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("Test Tumbling Window Stream") .Stream() .TumblingWindow( diff --git a/src/Cortex.Tests/StreamsMediator/Tests/StreamBuilderMediatorExtensionsTests.cs b/src/Cortex.Tests/StreamsMediator/Tests/StreamBuilderMediatorExtensionsTests.cs index 0556457..9baeda5 100644 --- a/src/Cortex.Tests/StreamsMediator/Tests/StreamBuilderMediatorExtensionsTests.cs +++ b/src/Cortex.Tests/StreamsMediator/Tests/StreamBuilderMediatorExtensionsTests.cs @@ -44,7 +44,7 @@ public void SinkToCommand_CreatesSinkWithCorrectBehavior() .ReturnsAsync("result"); // Build a stream using the extension method - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("TestStream") .Stream() .SinkToCommand( @@ -80,7 +80,7 @@ public void SinkToVoidCommand_CreatesSinkWithCorrectBehavior() .Returns(Task.CompletedTask); // Build a stream using the extension method - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("TestStream") .Stream() .SinkToVoidCommand( @@ -115,7 +115,7 @@ public void SinkToNotification_CreatesSinkWithCorrectBehavior() .Returns(Task.CompletedTask); // Build a stream using the extension method - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("TestStream") .Stream() .SinkToNotification( @@ -150,7 +150,7 @@ public void PublishNotification_WorksWithNotificationTypeDirectly() .Returns(Task.CompletedTask); // Build a stream using the extension method - starts with notification type - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("NotificationStream") .Stream() .PublishNotification(mockMediator.Object) @@ -182,7 +182,7 @@ public void SinkToCommand_InvokesResultHandler() .ReturnsAsync((StreamExtensionTestCommand cmd, CancellationToken _) => $"processed-{cmd.Input}"); // Build a stream using the extension method - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("TestStream") .Stream() .SinkToCommand( @@ -216,7 +216,7 @@ public void SinkToCommand_InvokesErrorHandler_OnException() .ThrowsAsync(new InvalidOperationException("Test error")); // Build a stream using the extension method - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("TestStream") .Stream() .SinkToCommand( @@ -252,7 +252,7 @@ public void SinkToNotification_InvokesCompletionHandler() .Returns(Task.CompletedTask); // Build a stream using the extension method - var stream = StreamBuilder + var stream = StreamBuilder .CreateNewStream("TestStream") .Stream() .SinkToNotification(