Skip to content

Simplify StreamBuilder API to use a single type parameter #186

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
eneshoxha merged 1 commit into from
Jan 28, 2026
Merged

Simplify StreamBuilder API to use a single type parameter #186

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 5 additions & 3 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<int, int>.CreateNewStream("ExampleStream")
var stream = StreamBuilder<int>.CreateNewStream("ExampleStream")
.Stream()
.Map(x => x * 2)
.Filter(x => x > 10)
.Sink(Console.WriteLine)
Expand All @@ -203,9 +204,10 @@ Console.WriteLine(stateStore.Get("key1"));

```csharp
var telemetryProvider = new OpenTelemetryProvider();
var stream = StreamBuilder<int, int>
var stream = StreamBuilder<int>
.CreateNewStream("TelemetryStream")
.WithTelemetry(telemetryProvider)
.Stream()
.Map(x => x * 2)
.Sink(Console.WriteLine)
.Build();
Expand Down Expand Up @@ -239,7 +241,7 @@ public class ClickEvent
static void Main(string[] args)
{
// Build the stream
var stream = StreamBuilder<ClickEvent, ClickEvent>.CreateNewStream("ClickStream")
var stream = StreamBuilder<ClickEvent>.CreateNewStream("ClickStream")
.Stream()
.Filter(e => !string.IsNullOrEmpty(e.PageUrl))
.GroupBySilently(
Expand Down
18 changes: 8 additions & 10 deletions src/Cortex.Streams.Mediator/Extensions/InitialStreamBuilderMediatorExtensions.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,21 +15,20 @@ public static class InitialStreamBuilderMediatorExtensions
/// Starts a stream using a Mediator streaming query as the source.
/// </summary>
/// <typeparam name="TIn">The initial input type of the stream.</typeparam>
/// <typeparam name="TCurrent">The current type of data in the stream (same as TIn for initial builders).</typeparam>
/// <typeparam name="TQuery">The type of streaming query.</typeparam>
/// <param name="builder">The initial stream builder instance.</param>
/// <param name="mediator">The mediator instance.</param>
/// <param name="query">The streaming query to execute.</param>
/// <param name="errorHandler">Optional handler for errors during query execution.</param>
/// <returns>A stream builder for further configuration.</returns>
public static IStreamBuilder<TIn, TCurrent> StreamFromQuery<TIn, TCurrent, TQuery>(
this IInitialStreamBuilder<TIn, TCurrent> builder,
public static IStreamBuilder<TIn, TIn> StreamFromQuery<TIn, TQuery>(
this IInitialStreamBuilder<TIn> builder,
IMediator mediator,
TQuery query,
Action<Exception> errorHandler = null)
where TQuery : IStreamQuery<TCurrent>
where TQuery : IStreamQuery<TIn>
{
var sourceOperator = new MediatorStreamQuerySourceOperator<TQuery, TCurrent>(
var sourceOperator = new MediatorStreamQuerySourceOperator<TQuery, TIn>(
mediator,
query,
errorHandler);
Expand All @@ -42,21 +41,20 @@ public static IStreamBuilder<TIn, TCurrent> StreamFromQuery<TIn, TCurrent, TQuer
/// This is useful when the query needs to be created lazily or with current context.
/// </summary>
/// <typeparam name="TIn">The initial input type of the stream.</typeparam>
/// <typeparam name="TCurrent">The current type of data in the stream.</typeparam>
/// <typeparam name="TQuery">The type of streaming query.</typeparam>
/// <param name="builder">The initial stream builder instance.</param>
/// <param name="mediator">The mediator instance.</param>
/// <param name="queryFactory">A factory function to create the streaming query.</param>
/// <param name="errorHandler">Optional handler for errors during query execution.</param>
/// <returns>A stream builder for further configuration.</returns>
public static IStreamBuilder<TIn, TCurrent> StreamFromQueryFactory<TIn, TCurrent, TQuery>(
this IInitialStreamBuilder<TIn, TCurrent> builder,
public static IStreamBuilder<TIn, TIn> StreamFromQueryFactory<TIn, TQuery>(
this IInitialStreamBuilder<TIn> builder,
IMediator mediator,
Func<TQuery> queryFactory,
Action<Exception> errorHandler = null)
where TQuery : IStreamQuery<TCurrent>
where TQuery : IStreamQuery<TIn>
{
var sourceOperator = new MediatorStreamQueryFactorySourceOperator<TQuery, TCurrent>(
var sourceOperator = new MediatorStreamQueryFactorySourceOperator<TQuery, TIn>(
mediator,
queryFactory,
errorHandler);
Expand Down
14 changes: 9 additions & 5 deletions src/Cortex.Streams/Abstractions/IInitialStreamBuilder.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,38 +5,42 @@

namespace Cortex.Streams.Abstractions
{
public interface IInitialStreamBuilder<TIn, TCurrent>
/// <summary>
/// Initial builder interface for creating a stream processing pipeline.
/// </summary>
/// <typeparam name="TIn">The type of the initial input to the stream.</typeparam>
public interface IInitialStreamBuilder<TIn>
{
/// <summary>
/// Start the stream inside the application, in-app streaming
/// </summary>
/// <returns></returns>
/// <exception cref="InvalidOperationException"></exception>
IStreamBuilder<TIn, TCurrent> Stream();
IStreamBuilder<TIn, TIn> Stream();

/// <summary>
/// Start configuring the Stream
/// </summary>
/// <param name="sourceOperator">Type of the Source Operator</param>
/// <returns></returns>
/// <exception cref="InvalidOperationException"></exception>
IStreamBuilder<TIn, TCurrent> Stream(ISourceOperator<TCurrent> sourceOperator);
IStreamBuilder<TIn, TIn> Stream(ISourceOperator<TIn> sourceOperator);

/// <summary>
/// Configure Telemetry for the Stream
/// </summary>
/// <param name="telemetryProvider">Telemetry provider like OpenTelemetryProvider</param>
/// <returns></returns>
/// <exception cref="InvalidOperationException"></exception>
IInitialStreamBuilder<TIn, TCurrent> WithTelemetry(ITelemetryProvider telemetryProvider);
IInitialStreamBuilder<TIn> WithTelemetry(ITelemetryProvider telemetryProvider);


/// <summary>
/// Configure global error handling for the stream.
/// </summary>
/// <param name="executionOptions">Execution options controlling error handling strategy and callbacks.</param>
/// <returns>The initial builder for chaining.</returns>
IInitialStreamBuilder<TIn, TCurrent> WithErrorHandling(StreamExecutionOptions executionOptions);
IInitialStreamBuilder<TIn> WithErrorHandling(StreamExecutionOptions executionOptions);

}
}
49 changes: 28 additions & 21 deletions src/Cortex.Streams/StreamBuilder.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,29 @@

namespace Cortex.Streams
{
/// <summary>
/// Entry point for creating a stream processing pipeline.
/// </summary>
/// <typeparam name="TIn">The type of the initial input to the stream.</typeparam>
public static class StreamBuilder<TIn>
{
/// <summary>
/// Creates a new stream with the specified name.
/// </summary>
/// <param name="name">The name of the stream.</param>
/// <returns>An initial stream builder.</returns>
public static IInitialStreamBuilder<TIn> CreateNewStream(string name)
{
return new StreamBuilder<TIn, TIn>(name);
}
}

/// <summary>
/// Builds a stream processing pipeline with optional branches.
/// </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>
public class StreamBuilder<TIn, TCurrent> : IInitialStreamBuilder<TIn, TCurrent>, IStreamBuilder<TIn, TCurrent>
internal class StreamBuilder<TIn, TCurrent> : IInitialStreamBuilder<TIn>, IStreamBuilder<TIn, TCurrent>
{
private readonly string _name;
private IOperator _firstOperator;
Expand All @@ -29,12 +46,12 @@ public class StreamBuilder<TIn, TCurrent> : IInitialStreamBuilder<TIn, TCurrent>



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;
Expand All @@ -44,24 +61,14 @@ private StreamBuilder(string name, IOperator firstOperator, IOperator lastOperat
_executionOptions = executionOptions ?? StreamExecutionOptions.Default;
}

/// <summary>
/// Creates a new stream with the specified name.
/// </summary>
/// <param name="name">The name of the stream.</param>
/// <returns>An initial stream builder.</returns>
public static IInitialStreamBuilder<TIn, TIn> CreateNewStream(string name)
{
return new StreamBuilder<TIn, TIn>(name);
}

/// <summary>
/// Creates a new stream with the specified name.
/// </summary>
/// <param name="name">The name of the stream.</param>
/// <param name="firstOperator">The first operator in the pipeline</param>
/// <param name="lastOperator">The last operator in the pipeline</param>
/// <returns>An initial stream builder.</returns>
public static IStreamBuilder<TIn, TCurrent> CreateNewStream(string name, IOperator firstOperator, IOperator lastOperator)
internal static IStreamBuilder<TIn, TCurrent> CreateNewStream(string name, IOperator firstOperator, IOperator lastOperator)
{
return new StreamBuilder<TIn, TCurrent>(name, firstOperator, lastOperator, false, null);
}
Expand Down Expand Up @@ -163,14 +170,14 @@ public ISinkBuilder<TIn, TCurrent> Sink(ISinkOperator<TCurrent> sinkOperator)
/// <param name="sourceOperator">Type of the Source Operator</param>
/// <returns></returns>
/// <exception cref="InvalidOperationException"></exception>
public IStreamBuilder<TIn, TCurrent> Stream(ISourceOperator<TCurrent> sourceOperator)
IStreamBuilder<TIn, TIn> IInitialStreamBuilder<TIn>.Stream(ISourceOperator<TIn> sourceOperator)
{
if (_sourceAdded)
{
throw new InvalidOperationException("Source operator already added.");
}

var sourceAdapter = new SourceOperatorAdapter<TCurrent>(sourceOperator);
var sourceAdapter = new SourceOperatorAdapter<TIn>(sourceOperator);

if (_firstOperator == null)
{
Expand All @@ -183,15 +190,15 @@ public IStreamBuilder<TIn, TCurrent> Stream(ISourceOperator<TCurrent> sourceOper
}

_sourceAdded = true;
return this; // Returns IStreamBuilder<TIn, TCurrent>
return (IStreamBuilder<TIn, TIn>)(object)this;
}

/// <summary>
/// Start the stream inside the application, in-app streaming
/// </summary>
/// <returns></returns>
/// <exception cref="InvalidOperationException"></exception>
public IStreamBuilder<TIn, TCurrent> Stream()
IStreamBuilder<TIn, TIn> IInitialStreamBuilder<TIn>.Stream()
{
// In memory source added.
if (_sourceAdded)
Expand All @@ -200,7 +207,7 @@ public IStreamBuilder<TIn, TCurrent> Stream()
}

_sourceAdded = true;
return this; // Returns IStreamBuilder<TIn, TCurrent>
return (IStreamBuilder<TIn, TIn>)(object)this;
}


Expand Down Expand Up @@ -375,7 +382,7 @@ public IStreamBuilder<TIn, KeyValuePair<TKey, TAggregate>> Aggregate<TKey, TAggr
return new StreamBuilder<TIn, KeyValuePair<TKey, TAggregate>>(_name, _firstOperator, _lastOperator, _sourceAdded, _telemetryProvider, _executionOptions);
}

public IInitialStreamBuilder<TIn, TCurrent> WithTelemetry(ITelemetryProvider telemetryProvider)
IInitialStreamBuilder<TIn> IInitialStreamBuilder<TIn>.WithTelemetry(ITelemetryProvider telemetryProvider)
{
_telemetryProvider = telemetryProvider;
return this;
Expand Down Expand Up @@ -723,7 +730,7 @@ public IStreamBuilder<TIn, WindowResult<string, TCurrent>> AdvancedSessionWindow
return new StreamBuilder<TIn, WindowResult<string, TCurrent>>(_name, _firstOperator, _lastOperator, _sourceAdded, _telemetryProvider, _executionOptions);
}

public IInitialStreamBuilder<TIn, TCurrent> WithErrorHandling(StreamExecutionOptions executionOptions)
IInitialStreamBuilder<TIn> IInitialStreamBuilder<TIn>.WithErrorHandling(StreamExecutionOptions executionOptions)
{
_executionOptions = executionOptions ?? StreamExecutionOptions.Default;
_executionOptions.StreamName = _name;
Expand Down
Loading