<repo_name>bubdm/Broker<file_sep>/samples/Broker.Samples/Handlers/BaseHandler.cs
using System.Threading.Tasks;
using Broker.Samples.Messages;

namespace Broker.Samples.Handlers
{
    public abstract class BaseHandler : IHandle<GreetingMessage>
    {
        public abstract Task HandleAsync(GreetingMessage message);
    }
}
<file_sep>/src/Broker/IPipeline.cs
using System;
using System.Threading.Tasks;

namespace Broker
{
    public interface IPipeline<in TMessage>
    {
        Task ExecuteAsync(TMessage message, Func<Task> next);
    }

    public interface IPipeline<in TMessage, TResult>
    {
        Task<TResult> ExecuteAsync(TMessage message, Func<Task<TResult>> next);
    }
}
<file_sep>/samples/Broker.Samples/Pipelines/GenericPipeline.cs
using System;
using System.Threading.Tasks;

namespace Broker.Samples.Pipelines
{
    public class GenericPipeline<TMessage> : IPipeline<TMessage>
    {
        public async Task ExecuteAsync(TMessage message, Func<Task> next)
        {
            Console.WriteLine("Before generic");
            await next().ConfigureAwait(false);
            Console.WriteLine("After generic");
        }
    }
}
<file_sep>/samples/Broker.Samples/Messages/GreetingMessage.cs
namespace Broker.Samples.Messages
{
    public class GreetingMessage : IAudit
    {
        public string Name { get; set; }

        public string User { get; set; }
    }
}
<file_sep>/samples/Broker.Samples/Registars/AutofacRegistar.cs
using Autofac;
using Broker.Extensions.Autofac.DependencyInjection;
using Broker.Samples.Messages;
using Broker.Samples.Pipelines;

namespace Broker.Samples.Registars
{
    internal class AutofacRegistar : IRegistar
    {
        public IBroker Register()
        {
            var builder = new ContainerBuilder();
            builder.AddBroker();
            builder.RegisterGeneric(typeof(GenericPipeline<>)).As(typeof(IPipeline<>));
            builder.RegisterType(typeof(GreetingPipeline)).As(typeof(IPipeline<GreetingMessage>));
            builder.RegisterGeneric(typeof(GenericQueryPipeline<,>)).As(typeof(IPipeline<,>));
            builder.RegisterType(typeof(GreetingQueryPipeline)).As(typeof(IPipeline<GreetingMessage, string>));

            var container = builder.Build();
            var broker = container.Resolve<IBroker>();

            return broker;
        }
    }
}
<file_sep>/src/Broker.Extensions.Microsoft.DependencyInjection/ServiceFactory.cs
using System;
using System.Collections.Generic;
using Microsoft.Extensions.DependencyInjection;

namespace Broker.Extensions.Microsoft.DependencyInjection
{
    internal class ServiceFactory : IServiceFactory
    {
        private readonly IServiceProvider _provider;

        public ServiceFactory(IServiceProvider provider)
        {
            _provider = provider;
        }

        public T GetService<T>()
        {
            return _provider.GetService<T>();
        }

        public IEnumerable<T> GetServices<T>()
        {
            return _provider.GetServices<T>();
        }
    }
}
<file_sep>/src/Broker.Extensions.Autofac.DependencyInjection/ServiceFactory.cs
using System.Collections.Generic;
using Autofac;

namespace Broker.Extensions.Autofac.DependencyInjection
{
    public class ServiceFactory : IServiceFactory
    {
        private readonly IComponentContext _context;

        public ServiceFactory(IComponentContext context)
        {
            _context = context;
        }

        public T GetService<T>()
        {
            return _context.Resolve<T>();
        }

        public IEnumerable<T> GetServices<T>()
        {
            return _context.Resolve<IEnumerable<T>>();
        }
    }
}
<file_sep>/src/Broker.Extensions.Autofac.DependencyInjection/ContainerBuilderExtensions.cs
using Autofac;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;

namespace Broker.Extensions.Autofac.DependencyInjection
{
    public static class ContainerBuilderExtensions
    {
        public static ContainerBuilder AddBroker(this ContainerBuilder builder) =>
            AddBrokerInternal(builder, AppDomain.CurrentDomain.GetAssemblies().Where(a => !a.IsDynamic));

        public static ContainerBuilder AddBroker(this ContainerBuilder builder, params Assembly[] assemblies) =>
            AddBrokerInternal(builder, assemblies.AsEnumerable());

        public static ContainerBuilder AddBroker(this ContainerBuilder builder, IEnumerable<Assembly> assemblies) =>
            AddBrokerInternal(builder, assemblies);

        private static ContainerBuilder AddBrokerInternal(ContainerBuilder builder, IEnumerable<Assembly> assemblies)
        {
            var assembliesToScan = (assemblies as Assembly[] ?? assemblies).Distinct().ToArray();
            builder.RegisterModule(new BrokerAutofacModule(assembliesToScan));
            return builder;
        }
    }
}
<file_sep>/src/Broker/Broker.cs
using System;
using System.Linq;
using System.Threading.Tasks;

namespace Broker
{
    public class Broker : IBroker
    {
        private readonly IServiceFactory _factory;

        public Broker(IServiceFactory factory)
        {
            _factory = factory;
        }

        public async Task SendAsync<TMessage>(TMessage message)
        {
            if (message == null)
            {
                throw new ArgumentNullException(nameof(message));
            }

            var handler = _factory.GetService<IHandle<TMessage>>();
            if (handler == null)
            {
                throw new InvalidOperationException($"Message {message.GetType()} has no handlers registered");
            }

            var pipelines = _factory.GetServices<IPipeline<TMessage>>();

            Task HandlerAction() => handler.HandleAsync(message);

            var runner = pipelines
                .Reverse()
                .Aggregate((Func<Task>) HandlerAction,
                    (next, pipeline) => () => pipeline.ExecuteAsync(message, next));

            await runner().ConfigureAwait(false);
        }

        public async Task PublishAsync<TMessage>(TMessage message)
        {
            if (message == null)
            {
                throw new ArgumentNullException(nameof(message));
            }

            var handlers = _factory.GetServices<IHandle<TMessage>>();
            var pipelines = _factory.GetServices<IPipeline<TMessage>>().Reverse().ToList();

            foreach (var handler in handlers)
            {
                Task HandlerAction() => handler.HandleAsync(message);

                var runner = pipelines
                    .Aggregate((Func<Task>) HandlerAction,
                        (next, pipeline) => () => pipeline.ExecuteAsync(message, next));

                await runner().ConfigureAwait(false);
            }
        }

        public async Task<TResult> SendAsync<TMessage, TResult>(TMessage message)
        {
            if (message == null)
            {
                throw new ArgumentNullException(nameof(message));
            }

            var handler = _factory.GetService<IHandle<TMessage, TResult>>();
            if (handler == null)
            {
                throw new InvalidOperationException($"Message {message.GetType()} has no handlers registered");
            }

            var pipelines = _factory.GetServices<IPipeline<TMessage, TResult>>();

            Task<TResult> HandlerAction() => handler.HandleAsync(message);

            var runner = pipelines
                .Reverse()
                .Aggregate((Func<Task<TResult>>) HandlerAction,
                    (next, pipeline) => () => pipeline.ExecuteAsync(message, next));

            return await runner().ConfigureAwait(false);
        }
    }
}
<file_sep>/samples/Broker.Samples/Handlers/DerivedHandler.cs
using System;
using System.Threading.Tasks;
using Broker.Samples.Messages;

namespace Broker.Samples.Handlers
{
    public class DerivedHandler : BaseHandler
    {
        public override Task HandleAsync(GreetingMessage message)
        {
            Console.WriteLine("Derived");
            return Task.CompletedTask;
        }
    }
}
<file_sep>/src/Broker/IServiceFactory.cs
﻿using System.Collections.Generic;

namespace Broker
{
    public interface IServiceFactory
    {
        T GetService<T>();

        IEnumerable<T> GetServices<T>();
    }
}
<file_sep>/samples/Broker.Samples/Program.cs
using System;
using System.Threading.Tasks;
using Broker.Samples.Messages;
using Broker.Samples.Registars;

namespace Broker.Samples
{
    class Program
    {
        static async Task Main()
        {
            var registars = new IRegistar[]
            {
                new ServiceCollectionRegistar(),
                new AutofacRegistar()
            };

            var greetingMessage = new GreetingMessage { Name = "World", User = "User" };

            foreach (var registar in registars)
            {
                Console.WriteLine($"Running {registar.GetType().Name}");
                Console.WriteLine();

                var broker = registar.Register();

                await broker.SendAsync(greetingMessage);
                await broker.PublishAsync(greetingMessage);
                await broker.SendAsync<IAudit>(greetingMessage);

                var result = await broker.SendAsync<GreetingMessage, string>(greetingMessage);
                Console.WriteLine(result);

                Console.WriteLine();
                Console.WriteLine();
            }

            Console.ReadKey();
        }
    }
}
<file_sep>/samples/Broker.Samples/Registars/IRegistar.cs
namespace Broker.Samples.Registars
{
    internal interface IRegistar
    {
        IBroker Register();
    }
}
<file_sep>/samples/Broker.Samples/Pipelines/GenericQueryPipeline.cs
using System;
using System.Threading.Tasks;

namespace Broker.Samples.Pipelines
{
    public class GenericQueryPipeline<TMessage, TResult> : IPipeline<TMessage, TResult>
    {
        public async Task<TResult> ExecuteAsync(TMessage message, Func<Task<TResult>> next)
        {
            Console.WriteLine("Before generic");
            var result = await next();
            Console.WriteLine("After generic");
            return result;
        }
    }
}
<file_sep>/samples/Broker.Samples/Registars/ServiceCollectionRegistar.cs
using Broker.Extensions.Microsoft.DependencyInjection;
using Broker.Samples.Messages;
using Broker.Samples.Pipelines;
using Microsoft.Extensions.DependencyInjection;

namespace Broker.Samples.Registars
{
    internal class ServiceCollectionRegistar : IRegistar
    {
        public IBroker Register()
        {
            var services = new ServiceCollection();
            services.AddBroker();
            services.AddTransient(typeof(IPipeline<>), typeof(GenericPipeline<>));
            services.AddTransient(typeof(IPipeline<GreetingMessage>), typeof(GreetingPipeline));
            services.AddTransient(typeof(IPipeline<,>), typeof(GenericQueryPipeline<,>));
            services.AddTransient(typeof(IPipeline<GreetingMessage, string>), typeof(GreetingQueryPipeline));

            var provider = services.BuildServiceProvider();
            var broker = provider.GetService<IBroker>();

            return broker;
        }
    }
}
<file_sep>/src/Broker/IBroker.cs
using System.Threading.Tasks;

namespace Broker
{
    public interface IBroker
    {
        Task SendAsync<TMessage>(TMessage message);

        Task<TResult> SendAsync<TMessage, TResult>(TMessage message);

        Task PublishAsync<TMessage>(TMessage message);
    }
}
<file_sep>/samples/Broker.Samples/Handlers/SecondHandler.cs
using System;
using System.Threading.Tasks;
using Broker.Samples.Messages;

namespace Broker.Samples.Handlers
{
    public class SecondHandler : IHandle<GreetingMessage>
    {
        public Task HandleAsync(GreetingMessage message)
        {
            Console.WriteLine("Second");
            return Task.CompletedTask;
        }
    }
}
<file_sep>/src/Broker/IHandle.cs
using System.Threading.Tasks;

namespace Broker
{
    public interface IHandle<in TMessage>
    {
        Task HandleAsync(TMessage message);
    }

    public interface IHandle<in TMessage, TResult>
    {
        Task<TResult> HandleAsync(TMessage message);
    }
}
<file_sep>/samples/Broker.Samples/Handlers/AuditHandler.cs
using System;
using System.Threading.Tasks;
using Broker.Samples.Messages;

namespace Broker.Samples.Handlers
{
    public class AuditHandler : IHandle<IAudit>
    {
        public Task HandleAsync(IAudit message)
        {
            Console.WriteLine($"User: {message.User}");
            return Task.CompletedTask;
        }
    }
}
<file_sep>/samples/Broker.Samples/Pipelines/GreetingPipeline.cs
using System;
using System.Threading.Tasks;
using Broker.Samples.Messages;

namespace Broker.Samples.Pipelines
{
    public class GreetingPipeline : IPipeline<GreetingMessage>
    {
        public async Task ExecuteAsync(GreetingMessage message, Func<Task> next)
        {
            Console.WriteLine("Before greeting");
            await next().ConfigureAwait(false);
            Console.WriteLine("After greeting");
        }
    }
}
<file_sep>/samples/Broker.Samples/Handlers/GreetingQueryHandler.cs
using System.Threading.Tasks;
using Broker.Samples.Messages;

namespace Broker.Samples.Handlers
{
    public class GreetingHandleHandler : IHandle<GreetingMessage, string>
    {
        public Task<string> HandleAsync(GreetingMessage message)
        {
            return Task.FromResult($"Hello, {message.Name}");
        }
    }
}
<file_sep>/samples/Broker.Samples/Pipelines/GreetingQueryPipeline.cs
using System;
using System.Threading.Tasks;
using Broker.Samples.Messages;

namespace Broker.Samples.Pipelines
{
    public class GreetingQueryPipeline : IPipeline<GreetingMessage, string>
    {
        public async Task<string> ExecuteAsync(GreetingMessage message, Func<Task<string>> next)
        {
            Console.WriteLine("Before typed");
            var result = await next();
            result += "_Added";
            Console.WriteLine(result);
            Console.WriteLine("After typed");
            return result;
        }
    }
}
<file_sep>/samples/Broker.Samples/Messages/IAudit.cs
﻿namespace Broker.Samples.Messages
{
    public interface IAudit
    {
        string User { get; set; }
    }
}
<file_sep>/src/Broker.Extensions.Microsoft.DependencyInjection/ServiceCollectionExtensions.cs
using System;
using System.Collections.Generic;
using System.Linq;
using System.Reflection;
using Microsoft.Extensions.DependencyInjection;

namespace Broker.Extensions.Microsoft.DependencyInjection
{
    public static class ServiceCollectionExtensions
    {
        public static IServiceCollection AddBroker(this IServiceCollection services) =>
            services.AddBroker(AppDomain.CurrentDomain.GetAssemblies().Where(a => !a.IsDynamic));

        public static IServiceCollection AddBroker(this IServiceCollection services, params Assembly[] assemblies) =>
            services.AddBroker(assemblies.AsEnumerable());

        public static IServiceCollection AddBroker(this IServiceCollection services, IEnumerable<Assembly> assemblies)
        {
            RegisterRequiredServices(services);
            RegisterHandlers(services, assemblies);
            return services;
        }

        private static void RegisterRequiredServices(IServiceCollection services)
        {
            services.AddScoped<IServiceFactory>(p => new ServiceFactory(p));
            services.AddScoped<IBroker, Broker>();
        }

        private static void RegisterHandlers(IServiceCollection services, IEnumerable<Assembly> assemblies)
        {
            assemblies = (assemblies as Assembly[] ?? assemblies).Distinct().ToArray();

            var handlerTypes = new[] { typeof(IHandle<>), typeof(IHandle<,>) };

            var descriptors =
                from a in assemblies
                from t in a.DefinedTypes
                where t.IsClass && !t.IsAbstract
                from i in t.ImplementedInterfaces
                where i.IsGenericType && handlerTypes.Contains(i.GetGenericTypeDefinition())
                select new { ServiceType = i, ImplementationType = t };

            foreach (var descriptor in descriptors)
            {
                services.AddTransient(descriptor.ServiceType, descriptor.ImplementationType);
            }
        }
    }
}
<file_sep>/src/Broker.Extensions.Autofac.DependencyInjection/BrokerAutofacModule.cs
using System.Reflection;
using Autofac;
using Module = Autofac.Module;

namespace Broker.Extensions.Autofac.DependencyInjection
{
    public class BrokerAutofacModule : Module
    {
        private readonly Assembly[] _assemblies;

        public BrokerAutofacModule(Assembly[] assemblies)
        {
            _assemblies = assemblies;
        }

        protected override void Load(ContainerBuilder builder)
        {
            RegisterRequiredServices(builder);
            RegisterHandlers(builder, _assemblies);
        }

        private void RegisterRequiredServices(ContainerBuilder builder)
        {
            builder.RegisterType<Broker>().As<IBroker>().InstancePerLifetimeScope();
            builder.RegisterType<ServiceFactory>().As<IServiceFactory>().InstancePerLifetimeScope();
        }

        private void RegisterHandlers(ContainerBuilder builder, Assembly[] assemblies)
        {
            var handlerTypes = new[] { typeof(IHandle<>), typeof(IHandle<,>) };

            foreach (var handlerType in handlerTypes)
            {
                builder.RegisterAssemblyTypes(assemblies)
                    .AsClosedTypesOf(handlerType)
                    .AsImplementedInterfaces();
            }
        }
    }
}
