Remove MediatR package

.github/workflows/dotnet.yml

@@ -24,4 +24,4 @@ jobs:
     - name: Build
       run: dotnet build --no-restore ConductorSharp.sln
     - name: Test
-      run: dotnet test --no-restore --verbosity normal ConductorSharp.sln
+      run: dotnet test --no-restore --verbosity normal test/ConductorSharp.Engine.Tests

ConductorSharp.sln

@@ -30,6 +30,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "ConductorSharp.Patterns", "
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ConductorSharp.KafkaCancellationNotifier", "src\ConductorSharp.KafkaCancellationNotifier\ConductorSharp.KafkaCancellationNotifier.csproj", "{A94EE48D-17F3-432A-A47D-BCB9B1EF2670}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ConductorSharp.Engine.IntegrationTests", "test\ConductorSharp.Engine.IntegrationTests\ConductorSharp.Engine.IntegrationTests.csproj", "{30EACB1C-FB15-4294-AF46-3AF7ABEB2121}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -76,6 +78,10 @@ Global
 		{A94EE48D-17F3-432A-A47D-BCB9B1EF2670}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{A94EE48D-17F3-432A-A47D-BCB9B1EF2670}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{A94EE48D-17F3-432A-A47D-BCB9B1EF2670}.Release|Any CPU.Build.0 = Release|Any CPU
+		{30EACB1C-FB15-4294-AF46-3AF7ABEB2121}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{30EACB1C-FB15-4294-AF46-3AF7ABEB2121}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{30EACB1C-FB15-4294-AF46-3AF7ABEB2121}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{30EACB1C-FB15-4294-AF46-3AF7ABEB2121}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE

examples/ConductorSharp.ApiEnabled/Extensions/ServiceCollectionExtensions.cs

@@ -1,5 +1,7 @@
-using ConductorSharp.ApiEnabled.Handlers;
-using ConductorSharp.ApiEnabled.Services;
+using ConductorSharp.ApiEnabled.Services;
+using ConductorSharp.ApiEnabled.Workers;
+using ConductorSharp.ApiEnabled.Workers;
+using ConductorSharp.ApiEnabled.Workflows;
 using ConductorSharp.Engine.Extensions;
 using ConductorSharp.Engine.Health;
 
@@ -16,23 +18,23 @@ public static IServiceCollection ConfigureApiEnabled(this IServiceCollection hos
                 maxConcurrentWorkers: configuration.GetValue<int>("Conductor:MaxConcurrentWorkers"),
                 sleepInterval: configuration.GetValue<int>("Conductor:SleepInterval"),
                 longPollInterval: configuration.GetValue<int>("Conductor:LongPollInterval"),
-                domain: configuration.GetValue<string>("Conductor:WorkerDomain"),
-                typeof(ServiceCollectionExtensions).Assembly
+                domain: configuration.GetValue<string>("Conductor:WorkerDomain")
             )
             .SetHealthCheckService<FileHealthService>()
             .AddPipelines(pipelines =>
             {
                 pipelines.AddExecutionTaskTracking();
-                pipelines.AddContextLogging();
-                pipelines.AddRequestResponseLogging();
                 pipelines.AddValidation();
             });
 
         hostBuilder.AddSingleton<ITaskExecutionCounterService, TaskExecutionCounterService>();
-        hostBuilder.RegisterWorkerTask<PrepareEmailHandler>(options =>
+        hostBuilder.RegisterWorkerTask<PrepareEmailWorker>(options =>
         {
             options.OwnerEmail = "[email protected]";
         });
+        hostBuilder.RegisterWorkerTask<FirstTestWorker.Worker>();
+        hostBuilder.RegisterWorkerTask<SecondTestWorker.Worker>();
+        hostBuilder.RegisterWorkflow<TestWorkflow.Workflow>();
 
         return hostBuilder;
     }

examples/ConductorSharp.ApiEnabled/Program.cs

@@ -34,3 +34,5 @@
 app.MapControllers();
 app.MapHealthChecks("/health");
 app.Run();
+
+public partial class Program { }

examples/ConductorSharp.ApiEnabled/Workers/FirstTestWorker.cs

@@ -0,0 +1,33 @@
+using ConductorSharp.Engine;
+using ConductorSharp.Engine.Interface;
+using ConductorSharp.Engine.Util;
+
+namespace ConductorSharp.ApiEnabled.Workers
+{
+    public class FirstTestWorker : ITaskInput<FirstTestWorker.Response>
+    {
+        public string Input { get; set; }
+
+        public class Response
+        {
+            public string Output { get; set; }
+        }
+
+        public class Worker : Worker<FirstTestWorker, Response>
+        {
+            private readonly ILogger<Worker> _logger;
+
+            public Worker(ILogger<Worker> logger)
+            {
+                _logger = logger;
+            }
+
+            public override Task<Response> Handle(FirstTestWorker test, WorkerExecutionContext context, CancellationToken cancellationToken)
+            {
+                _logger.LogInformation("First test worker");
+
+                return Task.FromResult<Response>(new() { Output = test.Input });
+            }
+        }
+    }
+}

examples/ConductorSharp.ApiEnabled/Handlers/PrepareEmailHandler.cs → examples/ConductorSharp.ApiEnabled/Workers/PrepareEmailWorker.cs

@@ -2,11 +2,10 @@
 using ConductorSharp.Engine.Builders.Metadata;
 using ConductorSharp.Engine.Interface;
 using ConductorSharp.Engine.Util;
-using MediatR;
 
-namespace ConductorSharp.ApiEnabled.Handlers;
+namespace ConductorSharp.ApiEnabled.Workers;
 
-public class PrepareEmailRequest : IRequest<PrepareEmailResponse>
+public class PrepareEmailRequest : ITaskInput<PrepareEmailResponse>
 {
     public string CustomerName { get; set; }
     public string Address { get; set; }
@@ -18,18 +17,16 @@ public class PrepareEmailResponse
 }
 
 [OriginalName("EMAIL_prepare")]
-public class PrepareEmailHandler : ITaskRequestHandler<PrepareEmailRequest, PrepareEmailResponse>
+public class PrepareEmailWorker : IWorker<PrepareEmailRequest, PrepareEmailResponse>
 {
-    private readonly ConductorSharpExecutionContext _context;
-    private readonly ILogger<PrepareEmailHandler> _logger;
+    private readonly ILogger<PrepareEmailWorker> _logger;
 
-    public PrepareEmailHandler(ConductorSharpExecutionContext context, ILogger<PrepareEmailHandler> logger)
+    public PrepareEmailWorker(ILogger<PrepareEmailWorker> logger)
     {
-        _context = context;
         _logger = logger;
     }
 
-    public async Task<PrepareEmailResponse> Handle(PrepareEmailRequest request, CancellationToken cancellationToken)
+    public async Task<PrepareEmailResponse> Handle(PrepareEmailRequest request, WorkerExecutionContext context, CancellationToken cancellationToken)
     {
         var emailBodyBuilder = new StringBuilder();
 
@@ -39,8 +36,8 @@ public async Task<PrepareEmailResponse> Handle(PrepareEmailRequest request, Canc
         emailBodyBuilder.AppendLine($"Customer: {request.CustomerName}");
         emailBodyBuilder.AppendLine($"Address: {request.Address}");
         emailBodyBuilder.AppendLine("------------------");
-        emailBodyBuilder.AppendLine($"WorkflowId : {_context.WorkflowId}");
-        emailBodyBuilder.AppendLine($"WorkflowName: {_context.WorkflowName}");
+        emailBodyBuilder.AppendLine($"WorkflowId : {context.WorkflowId}");
+        emailBodyBuilder.AppendLine($"WorkflowName: {context.WorkflowName}");
 
         _logger.LogInformation("Prepared email");
 

examples/ConductorSharp.ApiEnabled/Workers/SecondTestWorker.cs

@@ -0,0 +1,31 @@
+using ConductorSharp.Engine;
+using ConductorSharp.Engine.Interface;
+using ConductorSharp.Engine.Util;
+
+namespace ConductorSharp.ApiEnabled.Workers;
+
+public class SecondTestWorker : ITaskInput<SecondTestWorker.Response>
+{
+    public string Input { get; set; }
+
+    public class Response
+    {
+        public string Output { get; set; }
+    }
+
+    public class Worker : Worker<SecondTestWorker, Response>
+    {
+        private readonly ILogger<Worker> _logger;
+
+        public Worker(ILogger<Worker> logger)
+        {
+            _logger = logger;
+        }
+
+        public override Task<Response> Handle(SecondTestWorker test, WorkerExecutionContext context, CancellationToken cancellationToken)
+        {
+            _logger.LogInformation("Second test worker");
+            return Task.FromResult<Response>(new() { Output = test.Input });
+        }
+    }
+}

examples/ConductorSharp.ApiEnabled/Workflows/TestWorkflow.cs

@@ -0,0 +1,36 @@
+using ConductorSharp.ApiEnabled.Workers;
+using ConductorSharp.ApiEnabled.Workers;
+using ConductorSharp.Engine.Builders;
+using ConductorSharp.Engine.Builders.Metadata;
+
+namespace ConductorSharp.ApiEnabled.Workflows;
+
+public class TestWorkflow : WorkflowInput<TestWorkflow.Output>
+{
+    public string Input { get; set; }
+
+    public class Output : WorkflowOutput
+    {
+        public string Out { get; set; }
+    }
+
+    [OriginalName("TEST_workflow")]
+    public class Workflow : Workflow<Workflow, TestWorkflow, Output>
+    {
+        public Workflow(WorkflowDefinitionBuilder<Workflow, TestWorkflow, Output> builder)
+            : base(builder) { }
+
+        public FirstTestWorker.Worker FirstWorker { get; set; }
+
+        public SecondTestWorker.Worker SecondWorker { get; set; }
+
+        public override void BuildDefinition()
+        {
+            _builder.AddTask(wf => wf.FirstWorker, wf => new() { Input = wf.Input.Input });
+
+            _builder.AddTask(wf => wf.SecondWorker, wf => new() { Input = wf.FirstWorker.Output.Output });
+
+            _builder.SetOutput(wf => new() { Out = wf.SecondWorker.Output.Output });
+        }
+    }
+}

examples/ConductorSharp.Definitions/Generated/Task.cs

@@ -1,12 +1,12 @@
 using ConductorSharp.Engine.Builders.Metadata;
+using ConductorSharp.Engine.Interface;
 using ConductorSharp.Engine.Model;
 using ConductorSharp.Engine.Util;
-using MediatR;
 using Newtonsoft.Json;
 
 namespace ConductorSharp.Definitions.Generated
 {
-    public partial class EmailPrepareV1Input : IRequest<EmailPrepareV1Output>
+    public partial class EmailPrepareV1Input : ITaskInput<EmailPrepareV1Output>
     {
         /// <originalName>
         /// address
@@ -33,7 +33,7 @@ public partial class EmailPrepareV1Output
     [OriginalName("EMAIL_prepare")]
     public partial class EmailPrepareV1 : SimpleTaskModel<EmailPrepareV1Input, EmailPrepareV1Output> { }
 
-    public partial class CustomerGetV1Input : IRequest<CustomerGetV1Output>
+    public partial class CustomerGetV1Input : ITaskInput<CustomerGetV1Output>
     {
         /// <originalName>
         /// customer_id
@@ -61,7 +61,7 @@ public partial class CustomerGetV1Output
     [OriginalName("CUSTOMER_get")]
     public partial class CustomerGetV1 : SimpleTaskModel<CustomerGetV1Input, CustomerGetV1Output> { }
 
-    public partial class EnumTaskInput : IRequest<EnumTaskOutput>
+    public partial class EnumTaskInput : ITaskInput<EnumTaskOutput>
     {
         /// <originalName>
         /// status

examples/ConductorSharp.Definitions/Middlewares/CustomMiddleware.cs

@@ -0,0 +1,29 @@
+using ConductorSharp.Engine.Interface;
+using ConductorSharp.Engine.Util;
+using Microsoft.Extensions.Logging;
+
+namespace ConductorSharp.Definitions.Middlewares;
+
+internal class CustomMiddleware<TRequest, TResponse> : IWorkerMiddleware<TRequest, TResponse>
+    where TRequest : ITaskInput<TResponse>, new()
+{
+    private readonly ILogger<CustomMiddleware<TRequest, TResponse>> _logger;
+
+    public CustomMiddleware(ILogger<CustomMiddleware<TRequest, TResponse>> logger)
+    {
+        _logger = logger;
+    }
+
+    public async Task<TResponse> Handle(
+        TRequest request,
+        WorkerExecutionContext context,
+        Func<Task<TResponse>> next,
+        CancellationToken cancellationToken
+    )
+    {
+        _logger.LogInformation("Executed before all middlewares");
+        var response = await next();
+        _logger.LogInformation("Executed after all middlewares");
+        return response;
+    }
+}

examples/ConductorSharp.Definitions/Program.cs

@@ -1,4 +1,4 @@
-using ConductorSharp.Definitions.Behaviors;
+using ConductorSharp.Definitions.Middlewares;
 using ConductorSharp.Definitions.Workflows;
 using ConductorSharp.Engine.Extensions;
 using ConductorSharp.Engine.Health;
@@ -25,15 +25,13 @@
                     maxConcurrentWorkers: configuration.GetValue<int>("Conductor:MaxConcurrentWorkers"),
                     sleepInterval: configuration.GetValue<int>("Conductor:SleepInterval"),
                     longPollInterval: configuration.GetValue<int>("Conductor:LongPollInterval"),
-                    domain: configuration.GetValue<string>("Conductor:WorkerDomain"),
-                    typeof(Program).Assembly
+                    domain: configuration.GetValue<string>("Conductor:WorkerDomain")
                 )
                 .AddConductorSharpPatterns()
                 .SetHealthCheckService<FileHealthService>()
                 .AddPipelines(pipelines =>
                 {
-                    pipelines.AddCustomBehavior(typeof(CustomBehavior<,>));
-                    pipelines.AddRequestResponseLogging();
+                    pipelines.AddCustomMiddleware(typeof(CustomMiddleware<,>));
                     pipelines.AddValidation();
                 })
                 .AddCSharpLambdaTasks();

examples/ConductorSharp.Definitions/Workflows/CSharpLambdaWorkflow.cs

@@ -1,9 +1,9 @@
 using ConductorSharp.Engine.Builders;
 using ConductorSharp.Engine.Builders.Metadata;
+using ConductorSharp.Engine.Interface;
 using ConductorSharp.Engine.Model;
 using ConductorSharp.Patterns.Builders;
 using ConductorSharp.Patterns.Model;
-using MediatR;
 
 namespace ConductorSharp.Definitions.Workflows
 {
@@ -18,7 +18,7 @@ public class CSharpLambdaWorkflowOutput : WorkflowOutput { }
     [WorkflowMetadata(OwnerEmail = "[email protected]")]
     public class CSharpLambdaWorkflow : Workflow<CSharpLambdaWorkflow, CSharpLambdaWorkflowInput, CSharpLambdaWorkflowOutput>
     {
-        public class LambdaTaskInput : IRequest<LambdaTaskOutput>
+        public class LambdaTaskInput : ITaskInput<LambdaTaskOutput>
         {
             public string LambdaInput { get; set; }
         }

examples/ConductorSharp.Definitions/Workflows/SendCustomerNotification.cs

@@ -1,10 +1,9 @@
 using ConductorSharp.Definitions.Generated;
 using ConductorSharp.Engine.Builders;
 using ConductorSharp.Engine.Builders.Metadata;
+using ConductorSharp.Engine.Interface;
 using ConductorSharp.Engine.Model;
-using ConductorSharp.Engine.Util;
 using ConductorSharp.Patterns.Tasks;
-using MediatR;
 
 namespace ConductorSharp.Definitions.Workflows
 {
@@ -21,7 +20,7 @@ public class SendCustomerNotificationOutput : WorkflowOutput
         public object Constant { get; set; }
     }
 
-    public class ExpectedDynamicInput : CustomerGetV1Input, IRequest<ExpectedDynamicOutput> { }
+    public class ExpectedDynamicInput : CustomerGetV1Input, ITaskInput<ExpectedDynamicOutput> { }
 
     public class ExpectedDynamicOutput : CustomerGetV1Output { }
 

examples/ConductorSharp.Definitions/appsettings.json

@@ -1,6 +1,6 @@
 {
   "Conductor": {
-    "BaseUrl": "http://localhost:8080",
+    "BaseUrl": "http://localhost:8127",
     "LongPollInterval": 100,
     "MaxConcurrentWorkers": 10,
     "SleepInterval": 500