| name | myth-flow |
| description | Use when you need to orchestrate data processing pipelines. Pipeline.Start(ctx) chains .Step()/.StepAsync()/.StepResultAsync(), .Tap()/.TapAsync() for side effects, .When() for conditional branches, and .Transform() for context type changes. Built-in Result<T> pattern, retry with exponential backoff, OpenTelemetry tracing, and per-step DI resolution. |
SKILL.md - Myth.Flow
Version: 1.0
Target Framework: .NET 8.0
License: Apache 2.0
Table of Contents
Overview
Myth.Flow is a fluent pipeline library for .NET that enables building data processing workflows declaratively. It provides a type-safe, composable way to chain operations with built-in support for telemetry, retry policies, transformations, the Result pattern, and conditional execution.
Key Features
- Fluent Pipeline API: Chain operations with intuitive method calls
- Result Pattern: Built-in support for success/failure handling
- OpenTelemetry Integration: Automatic distributed tracing
- Retry Policies: Exponential backoff with configurable attempts
- Transformations: Transform context between pipeline steps
- Side Effects (Tap): Execute operations without modifying context
- Conditional Execution: Execute pipeline branches based on predicates
- Cancellation Support: Full CancellationToken support
- Exception Filtering: Selective exception propagation
- Async-First: Fully asynchronous execution
Installation
dotnet add package Myth.Flow
Dependencies
- .NET 8.0 or higher
- Myth.Commons
- System.Diagnostics.DiagnosticSource (OpenTelemetry)
- Microsoft.Extensions.Logging.Abstractions
Core Concepts
1. Pipeline
A pipeline is a sequence of operations (steps) executed in order. Each step receives a context and returns a modified context.
var result = await Pipeline.Start(context)
.Step(ctx => Process(ctx))
.StepAsync(ctx => ProcessAsync(ctx))
.ExecuteAsync();
2. Pipeline Data (Context)
The pipeline processes data through a series of steps. The data can be any type - it doesn't need to be named "Context". Each step receives the current state and returns a modified state.
public class CreateUserContext {
public CreateUserRequest Request { get; set; }
public User? CreatedUser { get; set; }
public List<string> Errors { get; set; } = new();
}
var result = await Pipeline.Start(user)
.StepAsync(u => ValidateUserAsync(u))
.StepAsync(u => SaveUserAsync(u))
.ExecuteAsync();
var result = await Pipeline.Start(request)
.StepAsync(r => ProcessRequestAsync(r))
.ExecuteAsync();
var result = await Pipeline.Start(userId)
.StepAsync(id => GetUserAsync(id))
.ExecuteAsync();
Note: The generic type parameter TContext in API signatures is just a type variable name - it works with any type, not just classes named "Context".
3. Result Pattern
Operations can return Result<T> to indicate success or failure without throwing exceptions.
public async Task<Result<User>> CreateUserAsync(CreateUserContext ctx) {
if (string.IsNullOrEmpty(ctx.Request.Email))
return Result<User>.Failure("Email is required");
var user = await _repository.CreateAsync(ctx.Request);
return Result<User>.Success(user);
}
4. Steps vs Taps
- Steps: Transform the context (return modified context)
- Taps: Perform side effects without modifying context (logging, events, metrics)
API Reference
Pipeline Static Class
Namespace: Myth.Flow
Entry point for creating pipelines.
Methods
public static IPipelineBuilder<TInput> Start<TInput>(TInput input)
public static IPipelineBuilder<TInput> Start<TInput>(
TInput input,
Action<PipelineConfiguration> configure)
Examples:
var result = await Pipeline.Start(context)
.StepAsync(ctx => ProcessAsync(ctx))
.ExecuteAsync();
var result = await Pipeline.Start(context, config => {
config.EnableTelemetry = true;
config.DefaultRetryAttempts = 3;
})
.StepAsync(ctx => ProcessAsync(ctx))
.ExecuteAsync();
IPipelineBuilder Interface
Namespace: Myth.Interfaces
Fluent interface for building pipelines.
Step Methods
Synchronous Step
IPipelineBuilder<TContext> Step(
Func<TContext, TContext> handler,
Action<TContext>? onSuccess = null,
Action<Exception>? onError = null)
Example:
.Step(ctx => {
ctx.IsValidated = true;
return ctx;
},
onSuccess: ctx => _logger.LogInformation("Validated"),
onError: ex => _logger.LogError(ex, "Validation failed"))
Asynchronous Step
IPipelineBuilder<TContext> StepAsync(
Func<TContext, Task<TContext>> handler,
Action<TContext>? onSuccess = null,
Action<Exception>? onError = null)
IPipelineBuilder<TContext> StepAsync(
Func<TContext, CancellationToken, Task<TContext>> handler)
Examples:
.StepAsync(async ctx => {
ctx.User = await _repository.GetUserAsync(ctx.UserId);
return ctx;
})
.StepAsync(async (ctx, ct) => {
ctx.Data = await _api.FetchDataAsync(ctx.Id, ct);
return ctx;
})
Result-Based Steps
IPipelineBuilder<TContext> StepResult(
Func<TContext, Result<TContext>> handler)
IPipelineBuilder<TContext> StepResultAsync(
Func<TContext, Task<Result<TContext>>> handler)
IPipelineBuilder<TContext> StepResultAsync(
Func<TContext, CancellationToken, Task<Result<TContext>>> handler)
Behavior: If Result.IsFailure, pipeline stops and ExecuteAsync returns failure.
Examples:
.StepResult(ctx => {
if (ctx.Age < 18)
return Result<MyContext>.Failure("Must be 18 or older");
return Result<MyContext>.Success(ctx);
})
.StepResultAsync(async ctx => {
var isValid = await _validator.ValidateAsync(ctx.Data);
if (!isValid)
return Result<MyContext>.Failure("Validation failed");
return Result<MyContext>.Success(ctx);
})
.StepResultAsync(async (ctx, ct) => {
var result = await _service.ProcessAsync(ctx, ct);
return result.IsSuccess
? Result<MyContext>.Success(ctx)
: Result<MyContext>.Failure(result.Error);
})
Transformation Methods
IPipelineBuilder<TNewContext> Transform<TNewContext>(
Func<TContext, TNewContext> mapper)
IPipelineBuilder<TNewContext> TransformAsync<TNewContext>(
Func<TContext, Task<TNewContext>> mapper)
Use Case: Change the context type mid-pipeline.
Examples:
await Pipeline.Start(createUserContext)
.StepAsync(ctx => CreateUserAsync(ctx))
.Transform(ctx => new UserDto {
Id = ctx.CreatedUser.Id,
Email = ctx.CreatedUser.Email,
Name = ctx.CreatedUser.Name
})
.ExecuteAsync();
.TransformAsync(async ctx => {
var dto = ctx.User.To<UserDto>();
dto.ProfileUrl = await _storage.GetUrlAsync(ctx.User.ProfileImageId);
return dto;
})
Side Effect Methods (Tap)
IPipelineBuilder<TContext> Tap(Action<TContext> action)
IPipelineBuilder<TContext> TapAsync(Func<TContext, Task> action)
Use Case: Execute operations without modifying context (logging, events, metrics).
Examples:
.Tap(ctx => _logger.LogInformation("Processing user {UserId}", ctx.UserId))
.TapAsync(async ctx => {
await _eventBus.PublishAsync(new UserCreatedEvent {
UserId = ctx.CreatedUser.Id,
Email = ctx.CreatedUser.Email
});
})
.TapAsync(async ctx => {
await _metrics.IncrementCounterAsync("users_created");
})
Conditional Execution
IPipelineBuilder<TContext> When(
Func<TContext, bool> predicate,
Action<IPipelineBuilder<TContext>> configurePipeline)
Use Case: Execute pipeline branch only if condition is true.
Example:
await Pipeline.Start(order)
.StepAsync(o => ValidateOrderAsync(o))
.When(
o => o.Amount > 1000,
pipeline => pipeline
.TapAsync(o => SendHighValueNotificationAsync(o))
.TapAsync(o => LogHighValueOrderAsync(o))
.StepAsync(o => ApplySpecialDiscountAsync(o)))
.When(
o => o.IsFirstOrder,
pipeline => pipeline
.TapAsync(o => SendWelcomeEmailAsync(o)))
.StepAsync(o => SaveOrderAsync(o))
.ExecuteAsync();
Configuration Methods
IPipelineBuilder<TContext> WithTelemetry(string operationName)
IPipelineBuilder<TContext> WithRetry(int maxAttempts, int backoffMs = 100)
Examples:
.WithTelemetry("CreateUser")
.WithRetry(maxAttempts: 3, backoffMs: 200)
await Pipeline.Start(context)
.WithTelemetry("ProcessOrder")
.WithRetry(maxAttempts: 5, backoffMs: 100)
.StepResultAsync(ctx => ProcessAsync(ctx))
.ExecuteAsync();
Retry Behavior:
- Backoff time = backoffMs × attempt number
- Attempt 1: 100ms, Attempt 2: 200ms, Attempt 3: 300ms
- Does NOT retry
OperationCanceledException
- Does NOT retry filtered exceptions (see Exception Filter)
Execution Methods
Task<Result<TContext>> ExecuteAsync(CancellationToken cancellationToken = default)
Result<TContext> Execute()
Examples:
var result = await pipeline.ExecuteAsync();
if (result.IsSuccess) {
Console.WriteLine(result.Value);
} else {
Console.WriteLine($"Error: {result.ErrorMessage}");
if (result.Exception != null) {
_logger.LogError(result.Exception, "Pipeline failed");
}
}
var cts = new CancellationTokenSource();
var result = await pipeline.ExecuteAsync(cts.Token);
var result = pipeline.Execute();
Result Pattern
Namespace: Myth.Models
Result Struct
public readonly struct Result<T> {
public bool IsSuccess { get; }
public bool IsFailure { get; }
public T? Value { get; }
public string? ErrorMessage { get; }
public Exception? Exception { get; }
}
Static Factory Methods
public static Result<T> Success(T value)
public static Result<T> Failure(string errorMessage, Exception? exception = null)
Examples:
return Result<User>.Success(user);
return Result<User>.Failure("User not found");
try {
var user = await _repository.GetAsync(id);
return Result<User>.Success(user);
} catch (Exception ex) {
return Result<User>.Failure("Failed to retrieve user", ex);
}
var result = await GetUserAsync(userId);
if (result.IsSuccess) {
var user = result.Value;
Console.WriteLine(user.Name);
} else {
Console.WriteLine($"Error: {result.ErrorMessage}");
if (result.Exception != null) {
_logger.LogError(result.Exception, "Error getting user");
}
}
Configuration
PipelineConfiguration Class
Namespace: Myth.Models
public class PipelineConfiguration {
public bool EnableTelemetry { get; set; } = true;
public bool EnableLogging { get; set; } = true;
public int DefaultRetryAttempts { get; set; } = 0;
public int DefaultBackoffMs { get; set; } = 100;
public ActivitySource? ActivitySource { get; set; }
public HashSet<Type> ExceptionTypesToPropagate { get; set; } = new();
}
FlowConfigurationBuilder Class
Namespace: Myth.Builders
Fluent builder for Flow configuration.
Methods
FlowConfigurationBuilder UseTelemetry()
FlowConfigurationBuilder DisableTelemetry()
FlowConfigurationBuilder UseLogging()
FlowConfigurationBuilder DisableLogging()
FlowConfigurationBuilder UseRetry(int attempts, int backoffMs = 100)
FlowConfigurationBuilder DisableRetry()
FlowConfigurationBuilder UseActivitySource(ActivitySource activitySource)
FlowConfigurationBuilder UseActivitySource(string name, string? version = null)
FlowConfigurationBuilder UseExceptionFilter(params Type[] exceptionTypes)
FlowConfigurationBuilder UseExceptionFilter<TException>() where TException : Exception
PipelineConfiguration Build()
Example:
builder.Services.AddFlow(config => config
.UseTelemetry()
.UseLogging()
.UseRetry(3, 200)
.UseActivitySource("MyApp.Pipeline", "1.0.0")
.UseExceptionFilter<ArgumentException>()
.UseExceptionFilter<InvalidOperationException>());
ServiceCollectionExtensions
Namespace: Myth.Extensions
IServiceCollection AddFlow(this IServiceCollection services)
IServiceCollection AddFlow(
this IServiceCollection services,
Action<FlowConfigurationBuilder> configure)
Examples:
services.AddFlow();
services.AddFlow(builder => builder
.UseTelemetry()
.UseRetry(3, 100)
.DisableLogging());
Exceptions
PipelineException
Namespace: Myth.Exceptions
public class PipelineException : Exception {
public PipelineException(string message)
public PipelineException(string message, Exception? innerException)
}
When Thrown:
- StepResult returns failure
- StepResultAsync returns failure
PipelineConfigurationException
Namespace: Myth.Exceptions
public class PipelineConfigurationException : InvalidOperationException {
public PipelineConfigurationException(string message)
public PipelineConfigurationException(string message, Exception? innerException)
}
When Thrown:
- Configuration errors (always propagated, fail-fast)
Usage Examples
Example 1: Simple CRUD Pipeline
public class CreateUserContext {
public CreateUserRequest Request { get; set; }
public User? CreatedUser { get; set; }
}
public async Task<Result<User>> CreateUserAsync(CreateUserRequest request) {
var context = new CreateUserContext { Request = request };
var result = await Pipeline.Start(context)
.StepResultAsync(async ctx => {
if (string.IsNullOrEmpty(ctx.Request.Email))
return Result<CreateUserContext>.Failure("Email is required");
return Result<CreateUserContext>.Success(ctx);
})
.StepResultAsync(async ctx => {
var exists = await _userRepository.AnyAsync(u => u.Email == ctx.Request.Email);
if (exists)
return Result<CreateUserContext>.Failure("Email already exists");
return Result<CreateUserContext>.Success(ctx);
})
.StepAsync(async ctx => {
ctx.CreatedUser = new User {
Email = ctx.Request.Email,
Name = ctx.Request.Name,
CreatedAt = DateTime.UtcNow
};
await _userRepository.AddAsync(ctx.CreatedUser);
return ctx;
})
.TapAsync(async ctx => {
await _emailService.SendWelcomeEmailAsync(ctx.CreatedUser.Email);
})
.ExecuteAsync();
return result.IsSuccess
? Result<User>.Success(result.Value.CreatedUser!)
: Result<User>.Failure(result.ErrorMessage!);
}
Example 2: Pipeline with Transformation
public async Task<Result<UserDto>> CreateAndReturnDtoAsync(CreateUserRequest request) {
var result = await Pipeline.Start(new CreateUserContext { Request = request })
.WithTelemetry("CreateUser")
.WithRetry(maxAttempts: 2, backoffMs: 100)
.StepResultAsync(ctx => ValidateAsync(ctx))
.StepAsync(ctx => CreateUserAsync(ctx))
.TapAsync(ctx => PublishUserCreatedEventAsync(ctx))
.Transform<UserDto>(ctx => new UserDto {
Id = ctx.CreatedUser!.Id,
Email = ctx.CreatedUser.Email,
Name = ctx.CreatedUser.Name,
CreatedAt = ctx.CreatedUser.CreatedAt
})
.ExecuteAsync();
return result.IsSuccess
? Result<UserDto>.Success(result.Value)
: Result<UserDto>.Failure(result.ErrorMessage!);
}
Example 3: Conditional Pipeline Branches
public async Task<Result<Order>> ProcessOrderAsync(CreateOrderRequest request) {
var result = await Pipeline.Start(new OrderContext { Request = request })
.StepAsync(ctx => ValidateOrderAsync(ctx))
.StepAsync(ctx => CalculateTotalAsync(ctx))
.When(
ctx => ctx.Total > 1000,
pipeline => pipeline
.StepAsync(ctx => ApplyBulkDiscountAsync(ctx))
.TapAsync(ctx => NotifyFinanceTeamAsync(ctx)))
.When(
ctx => ctx.Request.CustomerId != null,
pipeline => pipeline
.StepAsync(ctx => LoadCustomerHistoryAsync(ctx))
.When(
ctx => ctx.IsFirstOrder,
p => p.TapAsync(ctx => SendWelcomeGiftAsync(ctx))))
.StepAsync(ctx => SaveOrderAsync(ctx))
.TapAsync(ctx => SendConfirmationEmailAsync(ctx))
.ExecuteAsync();
return result.IsSuccess
? Result<Order>.Success(result.Value.Order!)
: Result<Order>.Failure(result.ErrorMessage!);
}
Example 4: Exception Filtering
services.AddFlow(config => config
.UseExceptionFilter<ArgumentException>()
.UseExceptionFilter<InvalidOperationException>());
try {
var result = await Pipeline.Start(data)
.StepAsync(async ctx => {
if (ctx.Value < 0)
throw new ArgumentException("Value cannot be negative");
return ctx;
})
.ExecuteAsync();
} catch (ArgumentException ex) {
_logger.LogError(ex, "Validation error");
}
try {
var result = await Pipeline.Start(data)
.TapAsync(ctx => ThrowIfInvalid(ctx))
.Transform(ctx => new OutputContext(ctx))
.ExecuteAsync();
} catch (PipelineException ex) when (ex.InnerException is ArgumentException argEx) {
_logger.LogError(argEx, "Validation error propagated through Transform");
}
var result = await Pipeline.Start(data)
.StepAsync(async ctx => {
throw new IOException("File not found");
})
.ExecuteAsync();
Example 5: Multiple Side Effects
public async Task<Result<ProcessedData>> ProcessDataAsync(InputData input) {
var result = await Pipeline.Start(input)
.WithTelemetry("ProcessData")
.StepAsync(ctx => EnrichDataAsync(ctx))
.Tap(ctx => _logger.LogInformation("Data enriched: {Count} items", ctx.Items.Count))
.StepAsync(ctx => ValidateDataAsync(ctx))
.TapAsync(async ctx => {
await _metrics.RecordGaugeAsync("data.items", ctx.Items.Count);
})
.StepAsync(ctx => TransformDataAsync(ctx))
.TapAsync(async ctx => {
await _cache.SetAsync($"processed:{ctx.Id}", ctx);
})
.TapAsync(async ctx => {
await _eventBus.PublishAsync(new DataProcessedEvent {
Id = ctx.Id,
ItemCount = ctx.Items.Count,
ProcessedAt = DateTime.UtcNow
});
})
.Transform<ProcessedData>(ctx => new ProcessedData {
Id = ctx.Id,
Items = ctx.Items,
ProcessedAt = DateTime.UtcNow
})
.ExecuteAsync();
return result.IsSuccess
? Result<ProcessedData>.Success(result.Value)
: Result<ProcessedData>.Failure(result.ErrorMessage!);
}
Advanced Patterns
Pattern 1: Reusable Pipeline Steps
public class UserPipelineSteps {
private readonly IUserRepository _repository;
private readonly IValidator _validator;
public async Task<Result<TContext>> ValidateUserAsync<TContext>(TContext ctx)
where TContext : IUserContext {
var validationResult = await _validator.ValidateAsync(ctx.User);
return validationResult.IsValid
? Result<TContext>.Success(ctx)
: Result<TContext>.Failure(validationResult.ErrorMessage);
}
public async Task<TContext> LoadUserDetailsAsync<TContext>(TContext ctx)
where TContext : IUserContext {
ctx.User = await _repository.GetByIdAsync(ctx.UserId);
return ctx;
}
}
var result = await Pipeline.Start(context)
.StepResultAsync(ctx => _userSteps.ValidateUserAsync(ctx))
.StepAsync(ctx => _userSteps.LoadUserDetailsAsync(ctx))
.ExecuteAsync();
Pattern 2: Pipeline Factory
public class OrderPipelineFactory {
public IPipelineBuilder<OrderContext> CreateOrderPipeline(OrderContext context) {
return Pipeline.Start(context)
.WithTelemetry("CreateOrder")
.WithRetry(3, 100)
.StepResultAsync(ctx => _validator.ValidateAsync(ctx))
.StepAsync(ctx => _calculator.CalculateTotalAsync(ctx))
.When(ctx => ctx.Total > 1000,
p => p.StepAsync(ctx => _discount.ApplyBulkDiscountAsync(ctx)))
.StepAsync(ctx => _repository.SaveOrderAsync(ctx))
.TapAsync(ctx => _events.PublishOrderCreatedAsync(ctx));
}
public IPipelineBuilder<OrderContext> UpdateOrderPipeline(OrderContext context) {
return Pipeline.Start(context)
.WithTelemetry("UpdateOrder")
.StepAsync(ctx => _repository.LoadOrderAsync(ctx))
.StepResultAsync(ctx => _validator.ValidateUpdateAsync(ctx))
.StepAsync(ctx => _repository.UpdateOrderAsync(ctx))
.TapAsync(ctx => _events.PublishOrderUpdatedAsync(ctx));
}
}
var result = await _factory.CreateOrderPipeline(context).ExecuteAsync();
Pattern 3: Nested Pipelines
public async Task<Result<CompleteOrderContext>> ProcessCompleteOrderAsync(OrderRequest request) {
var result = await Pipeline.Start(new CompleteOrderContext { Request = request })
.StepResultAsync(async ctx => {
var inventoryResult = await Pipeline.Start(ctx.Request.Items)
.StepAsync(items => CheckInventoryAsync(items))
.StepAsync(items => ReserveInventoryAsync(items))
.ExecuteAsync();
if (inventoryResult.IsFailure)
return Result<CompleteOrderContext>.Failure("Inventory check failed");
ctx.ReservedItems = inventoryResult.Value;
return Result<CompleteOrderContext>.Success(ctx);
})
.StepAsync(ctx => CreateOrderAsync(ctx))
.StepResultAsync(async ctx => {
var paymentResult = await Pipeline.Start(ctx.Order)
.StepAsync(order => ValidatePaymentAsync(order))
.StepAsync(order => ProcessPaymentAsync(order))
.ExecuteAsync();
if (paymentResult.IsFailure)
return Result<CompleteOrderContext>.Failure("Payment failed");
ctx.PaymentConfirmation = paymentResult.Value;
return Result<CompleteOrderContext>.Success(ctx);
})
.ExecuteAsync();
return result.IsSuccess
? Result<CompleteOrderContext>.Success(result.Value)
: Result<CompleteOrderContext>.Failure(result.ErrorMessage!);
}
Best Practices
1. Use Result Pattern for Validation
✅ DO:
.StepResultAsync(async ctx => {
var isValid = await _validator.ValidateAsync(ctx.Data);
return isValid
? Result<MyContext>.Success(ctx)
: Result<MyContext>.Failure("Validation failed");
})
❌ DON'T:
.StepAsync(async ctx => {
var isValid = await _validator.ValidateAsync(ctx.Data);
if (!isValid)
throw new ValidationException("Validation failed");
return ctx;
})
2. Use Tap for Side Effects
✅ DO:
.TapAsync(async ctx => {
await _logger.LogInformationAsync("User created: {UserId}", ctx.UserId);
await _eventBus.PublishAsync(new UserCreatedEvent { UserId = ctx.UserId });
})
❌ DON'T:
.StepAsync(async ctx => {
await _logger.LogInformationAsync("User created: {UserId}", ctx.UserId);
return ctx;
})
3. Configure Globally in Startup
✅ DO:
services.AddFlow(config => config
.UseTelemetry()
.UseRetry(3, 100)
.UseLogging());
❌ DON'T:
Pipeline.Start(ctx, config => {
config.EnableTelemetry = true;
})
4. Use Meaningful Telemetry Names
✅ DO:
.WithTelemetry("CreateUser")
.WithTelemetry("ProcessOrder")
.WithTelemetry("CalculateInvoice")
❌ DON'T:
.WithTelemetry("Pipeline1")
.WithTelemetry("Process")
5. Handle Cancellation
✅ DO:
.StepAsync(async (ctx, ct) => {
ctx.Data = await _api.FetchDataAsync(ctx.Id, ct);
return ctx;
})
var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
var result = await pipeline.ExecuteAsync(cts.Token);
6. Transform Context Types Appropriately
✅ DO:
await Pipeline.Start(context)
.StepAsync(ctx => ProcessAsync(ctx))
.Transform<ResponseDto>(ctx => new ResponseDto {
Id = ctx.Result.Id,
Status = ctx.Result.Status
})
.ExecuteAsync();
Troubleshooting
Issue 1: MythServiceProvider.Current is null
Problem:
NullReferenceException when Pipeline.Start tries to resolve services
Cause: Global service provider not initialized.
Solution:
var app = builder.BuildApp();
var provider = services.BuildServiceProvider();
MythServiceProvider.Initialize(provider);
Issue 2: Retry Not Working
Problem: Retry not executing on failure.
Possible Causes:
-
Exception is filtered
services.AddFlow(c => c.UseExceptionFilter<ArgumentException>());
-
OperationCanceledException - Never retried
Solution: Check exception type and filter configuration.
Issue 3: StepResult Throws PipelineException
Problem: StepResult with failure throws exception.
Expected Behavior: This is by design. Use ExecuteAsync to get Result.
Solution:
var result = await pipeline.ExecuteAsync();
if (result.IsFailure) {
Console.WriteLine(result.ErrorMessage);
}
Issue 4: Telemetry Not Appearing
Problem: No telemetry activities in traces.
Checklist:
- Enable telemetry:
.UseTelemetry()
- Add operation name:
.WithTelemetry("OperationName")
- Configure ActivitySource
- Ensure OpenTelemetry is configured in app
Solution:
services.AddFlow(config => config
.UseTelemetry()
.UseActivitySource("MyApp", "1.0.0"));
await Pipeline.Start(ctx)
.WithTelemetry("CreateUser")
.ExecuteAsync();
Performance Considerations
- Async All the Way: Use StepAsync, not Step for I/O operations
- Retry Overhead: Each retry adds backoff delay
- Telemetry: Minimal overhead with OpenTelemetry
- Transform: Creates new pipeline builder, minimal cost
- Tap: Lightweight, use freely for side effects
Integration with Myth Ecosystem
With Myth.Commons
using Myth.Extensions;
using Myth.ServiceProvider;
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddFlow();
var app = builder.BuildApp();
With Myth.Guard
.StepResultAsync(async ctx => {
var validationResult = await _validator.ValidateAndReturnAsync(ctx.Data);
return validationResult.IsValid
? Result<MyContext>.Success(ctx)
: Result<MyContext>.Failure(validationResult.Errors.First().Message);
})
With Myth.Repository
.StepAsync(async ctx => {
ctx.User = await _userRepository.FirstOrDefaultAsync(u => u.Id == ctx.UserId);
return ctx;
})
Summary
Myth.Flow provides:
- ✅ Fluent Pipeline API: Intuitive, chainable method calls
- ✅ Result Pattern: Type-safe success/failure handling
- ✅ OpenTelemetry: Built-in distributed tracing
- ✅ Retry Policies: Automatic retry with exponential backoff
- ✅ Transformations: Change context types mid-pipeline
- ✅ Side Effects: Tap for non-modifying operations
- ✅ Conditional Logic: Execute branches based on predicates
- ✅ Cancellation: Full CancellationToken support
- ✅ Exception Filtering: Control exception propagation
Additional Resources
This documentation is maintained for AI agents and developers. For questions or contributions, please refer to the repository.