New Hub Pipeline

HubPipeline.cs

using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;

namespace SignalR.Hubs
{
    public class HubPipeline : IHubPipeline, IHubPipelineInvoker
    {
        private readonly Stack<IHubPipelineModule> _modules;
        private Lazy<ComposedPipeline> _builtPipeline;

        public HubPipeline()
        {
            _modules = new Stack<IHubPipelineModule>();
            _builtPipeline = new Lazy<ComposedPipeline>(() => new ComposedPipeline(_modules));
        }

        public IHubPipeline AddModule(IHubPipelineModule builder)
        {
            _modules.Push(builder);
            _builtPipeline = new Lazy<ComposedPipeline>(() => new ComposedPipeline(_modules));
            return this;
        }

        private ComposedPipeline BuiltPipeline
        {
            get { return _builtPipeline.Value; }
        }

        public Task<object> Invoke(IHubIncomingInvokerContext context)
        {
            return BuiltPipeline.Invoke(context);
        }

        public Task Connect(IHub hub)
        {
            return BuiltPipeline.Connect(hub);
        }

        public Task Reconnect(IHub hub)
        {
            return BuiltPipeline.Reconnect(hub);
        }

        public Task Disconnect(IHub hub)
        {
            return BuiltPipeline.Disconnect(hub);
        }

        public bool Authorized(IHub hub)
        {
            return BuiltPipeline.Authorize(hub);
        }

        public IEnumerable<string> RejoiningGroups(IHub hub, IEnumerable<string> groups)
        {
            return BuiltPipeline.RejoiningGroups(hub, groups);
        }

        public Task Send(IHubOutgoingInvokerContext context)
        {
            return BuiltPipeline.Send(context);
        }

        private class ComposedPipeline
        {

            public Func<IHubIncomingInvokerContext, Task<object>> Invoke;
            public Func<IHub, Task> Connect;
            public Func<IHub, Task> Reconnect;
            public Func<IHub, Task> Disconnect;
            public Func<IHub, bool> Authorize;
            public Func<IHub, IEnumerable<string>, IEnumerable<string>> RejoiningGroups;
            public Func<IHubOutgoingInvokerContext, Task> Send;

            public ComposedPipeline(Stack<IHubPipelineModule> modules)
            {
                // This wouldn't look nearly as gnarly if C# had better type inference, but now we don't need the ComposedModule or PassThroughModule.
                Invoke = Compose<Func<IHubIncomingInvokerContext, Task<object>>>(modules, (m, f) => m.BuildIncoming(f))(HubDispatcher.Incoming);
                Connect = Compose<Func<IHub, Task>>(modules, (m, f) => m.BuildConnect(f))(HubDispatcher.Connect);
                Reconnect = Compose<Func<IHub, Task>>(modules, (m, f) => m.BuildReconnect(f))(HubDispatcher.Reconnect);
                Disconnect = Compose<Func<IHub, Task>>(modules, (m, f) => m.BuildDisconnect(f))(HubDispatcher.Disconnect);
                Authorize = Compose<Func<IHub, bool>>(modules, (m, f) => m.BuildAuthorizeConnect(f))(HubDispatcher.AuthorizedConnection);
                RejoiningGroups = Compose<Func<IHub, IEnumerable<string>, IEnumerable<string>>>(modules, (m, f) => m.BuildRejoiningGroups(f))(HubDispatcher.RejoiningGroups);
                Send = Compose<Func<IHubOutgoingInvokerContext, Task>>(modules, (m, f) => m.BuildOutgoing(f))(HubDispatcher.Outgoing);
            }

            // IHubPipelineModule could be turned into a second generic parameter, but it would make the above invocations even longer than the currently are.
            private Func<T, T> Compose<T>(IEnumerable<IHubPipelineModule> modules, Func<IHubPipelineModule, T, T> method)
            {
                // Notice we are reversing and aggregating in one step. (Function composition is associative) 
                return modules.Aggregate<IHubPipelineModule, Func<T, T>>(x => x, (a, b) => (x => method(b, a(x))));
            }
        }
    }
}