private IEnumerable<Tuple<Type, Action<IEvent>>> BuildHandlerInvocations(IEventHandler handler)
 {
     return handler.GetType().GetInterfaces()
             .Where(i => i.IsGenericType && i.GetGenericTypeDefinition() == typeof(IEventHandler<>))
             .Select(i => new { HandlerInterface = i, EventType = i.GetGenericArguments()[0] })
             .Select(e => new Tuple<Type, Action<IEvent>>(e.EventType, this.BuildHandlerInvocation(handler, e.HandlerInterface, e.EventType)));
 }

        // Can't really infer the topic from the subscription, since subscriptions of the same 
        // name can exist across different topics (i.e. "all" currently)
        public EventProcessor CreateEventProcessor(string subscription, IEventHandler handler, ITextSerializer serializer)
        {
            if (!this.initialized)
                throw new InvalidOperationException("Service bus configuration has not been initialized.");

            var topicSettings = this.settings.Topics.Find(x => x.IsEventBus);
            if (topicSettings == null)
                throw new ArgumentOutOfRangeException("No topic has been marked with the IsEventBus attribute. Cannot create event processor.");

            var subscriptionSettings = topicSettings.Subscriptions.Find(x => x.Name == subscription);
            if (subscriptionSettings == null)
                throw new ArgumentOutOfRangeException(string.Format(
                    CultureInfo.CurrentCulture,
                    "Subscription '{0}' for topic '{1}' has not been registered in the service bus configuration.",
                    subscription, topicSettings.Path));

            var receiver = subscriptionSettings.RequiresSession ?
                (IMessageReceiver)new SessionSubscriptionReceiver(this.settings, topicSettings.Path, subscription) :
                (IMessageReceiver)new SubscriptionReceiver(this.settings, topicSettings.Path, subscription);

            var processor = new EventProcessor(receiver, serializer);
            processor.Register(handler);

            return processor;
        }

 public static void RegisterEventHandler(IEventHandler handler)
 {
     Monitor.Enter(_lock);
     if (!_handlers.Contains(handler))
         _handlers.Add(handler);
     Monitor.Exit(_lock);
 }

 /// <summary>
 /// Detaches a handler
 /// </summary>
 /// <param name="handler">A event handler/subscriber</param>
 public void DetachHandler(IEventHandler handler)
 {
     if (!_handlers.Remove(handler))
     {
         throw new ArgumentException("Referenced handler is not attached to this emmiter!");
     }
 }

 void IEventHelpers.RegisterSocketRead(IEvent e, BufferedSocket socket, IEventHandler protocol,
                                       ReadFunction callback)
 {
     //Contract.Requires(socket != null);
     Contract.Requires(protocol != null);
     Contract.Requires(callback != null);
 }

 private static void _ProcessEvent(object obj)
 {
     IEvent evnt = (IEvent)obj;
     lock(_rand){
         Thread.CurrentThread.Name = "EventProcessor{" + evnt.Name + "}["+_rand.Next().ToString()+"]";
     }
     Thread.CurrentThread.IsBackground = true;
     IEventHandler[] handlers = new IEventHandler[0];
     Monitor.Enter(_lock);
     handlers = new IEventHandler[_handlers.Count];
     _handlers.CopyTo(handlers);
     Monitor.Exit(_lock);
     foreach (IEventHandler handler in handlers)
     {
         Log.Trace("Checking if " + handler.GetType().FullName + " handles event " + evnt.Name);
         if (handler.HandlesEvent(evnt))
         {
             Thread delThread = null;
             Log.Trace("Handling event " + evnt.Name + " with handler " + handler.GetType().FullName);
             IAsyncResult res = _eventHandle.BeginInvoke(handler, evnt,ref delThread, null, null);
             if (res.AsyncWaitHandle.WaitOne(_EVENT_HANDLING_TIMEOUT))
                 _eventHandle.EndInvoke(ref delThread,res);
             else
             {
                 Log.Error("Handle " + handler.GetType().FullName + " timed out handling event " + evnt.Name);
                 try
                 {
                     delThread.Abort();
                 }
                 catch (Exception e) { }
                 _eventHandle.EndInvoke(ref delThread,res);
             }
         }
     }
 }

        private static void RegisterEventHandler(IEventHandler eventHandler)
        {
            object instance = eventHandler.Model;
            _modelMap.Add(instance.GetType(), instance);

            RegisterMessageHandler("SubscribeEventHandler", eventHandler);
        }

        private static MethodInfo GetMatchingMethod(IEventHandler eventHandler, Type interfaceType, string methodName, IDictionary<string, object> arguments)
        {
            var allMethods = new List<MethodInfo>(interfaceType.GetMethods());
            var candidates = new List<MethodInfo>(allMethods);

            foreach (var method in allMethods) {
                if (string.Equals(method.Name, methodName, StringComparison.OrdinalIgnoreCase)) {
                    ParameterInfo[] parameterInfos = method.GetParameters();
                    foreach (var parameter in parameterInfos) {
                        if (!arguments.ContainsKey(parameter.Name)) {
                            candidates.Remove(method);
                            break;
                        }
                    }
                }
                else {
                    candidates.Remove(method);
                }
            }

            // treating common case separately
            if (candidates.Count == 1) {
                return candidates[0];
            }

            if (candidates.Count != 0) {
                return candidates.OrderBy(x => x.GetParameters().Length).Last();
            }

            return null;
        }

 /// <summary>Constructs a timer stage</summary>
 /// <param name="handler">an event handler</param>
 /// <param name="initialDelay">an initial delay</param>
 /// <param name="period">a period in milli-seconds</param>
 public TimerStage(IEventHandler<PeriodicEvent> handler, long initialDelay, long period)
 {
     _handler = handler;
     _timer = new Timer(period);
     _timer.Elapsed += (sender, e) => OnTimedEvent(sender, e, _handler, _value);
     _timer.Enabled = true;
 }

        /// <summary>
        /// Registers a listener for a specific event
        /// </summary>
        /// <param name="eventType">Event the listener is listening for</param>
        /// <param name="listener">Listener receiving messages for the specified event</param>
        public void RegisterListener(EventType eventType, IEventHandler listener)
        {
            List<IEventHandler> listeners = RegisterEvent(eventType);
            if (listeners.Contains(listener))
                throw new Exception(string.Format("Listener {0} already listen to event {1}", listener, eventType.EventName));

            listeners.Add(listener);
        }

        public virtual void Invoke(IEventHandler handler, IEvent @event)
        {
            Ensure.That(handler, "handler").IsNotNull();
            Ensure.That(@event, "event").IsNotNull();

            var container = ContainerFactory.Invoke(handler.ContainerType);
            handler.Handler.Invoke(container, new object[] { @event });
        }

        private static IEventBus InitializeEventBus(IEventHandler<NewNoteAdded> handler, TextChangedHandler textChangedHandler)
        {
            var bus = new InProcessEventBus();
            bus.RegisterHandler(handler);
            bus.RegisterHandler(textChangedHandler);

            return bus;
        }

 public string RegisterHandler(IEventHandler handler)
 {
     _mut.WaitOne();
     string ret = _rand.NextString(_ID_LENGTH);
     _handlers.Add(new sEventHandler(handler, ret));
     _mut.ReleaseMutex();
     return ret;
 }

 private static bool TryInvoke(IEventHandler eventHandler, string interfaceName, string methodName, IDictionary<string, object> arguments, out IEnumerable returnValue) {
     Type type = eventHandler.GetType();
     foreach (var interfaceType in type.GetInterfaces()) {
         if (String.Equals(interfaceType.Name, interfaceName, StringComparison.OrdinalIgnoreCase)) {
             return TryInvokeMethod(eventHandler, interfaceType, methodName, arguments, out returnValue);
         }
     }
     returnValue = null;
     return false;
 }

        /// <summary>Constructs a timer stage</summary>
        /// <param name="handler">an event handler</param>
        /// <param name="initialDelay">an initial delay in the interval [0,MaxTimeValue]</param>
        /// <param name="period">a period in milli-seconds in the interval [0,MaxTimeValue]</param>
        public TimerStage(IEventHandler<PeriodicEvent> handler, long initialDelay, long period)
        {
            // Core .NET only supports 32 bit timers.
            Validate("initialDelay", initialDelay);
            Validate("period", period);

            _handler = handler;
            _timer = new Timer(
                (object state) => { OnTimedEvent(_handler, _value); }, this, (int)initialDelay, (int)period);
        }

        private bool TryNotifyHandler(IEventHandler eventHandler, string messageName, string interfaceName, string methodName, IDictionary<string, object> eventData, out IEnumerable returnValue) {
            try {
                return TryInvoke(eventHandler, messageName, interfaceName, methodName, eventData, out returnValue);
            } catch (Exception exception) {
                if (!_exceptionPolicy.HandleException(this, exception))
                    throw;

                returnValue = null;
                return false;
            }
        }

 public void addHandler(GameEventType gameEventType, int listeningToEntityId, IEventHandler handler)
 {
     if (!_handlers.ContainsKey(gameEventType))
         _handlers.Add(gameEventType, new Dictionary<int,List<IEventHandler>>());
     if (!_handlers[gameEventType].ContainsKey(listeningToEntityId))
         _handlers[gameEventType].Add(listeningToEntityId, new List<IEventHandler>());
     if (!_handlers[gameEventType][listeningToEntityId].Contains(handler))
     {
         _handlers[gameEventType][listeningToEntityId].Add(handler);
     }
 }

            protected override void Context()
            {
                _handler = MockRepository.GenerateMock<IEventHandler<EventData>>();

                IEventPublisher eventPublisher = new EventPublisher();

                eventPublisher.RegisterHandler(_handler);
                eventPublisher.UnregisterHandler(_handler);

                eventPublisher.Publish(new EventData("My Event Data"));
            }

        /// <summary>
        /// A factory method for creating a consumer from an array of hashes.
        /// </summary>
        /// <param name="user">The user object that's creating this consumer.</param>
        /// <param name="type">The consumer type.</param>
        /// <param name="hashes">An array of stream hashes.</param>
        /// <param name="event_handler">An object that implements IEventHandler which will receive the events.</param>
        /// <returns>An object of a type derived from this class.</returns>
        public static StreamConsumer factory(User user, string type, string[] hashes, IEventHandler event_handler)
        {
            switch (type)
            {
                case "http":
                    return new StreamConsumer_Http(user, hashes, event_handler);

                default:
                    throw new InvalidDataException("Consumer type \"" + type + "\" is unknown");
            }
        }

        public EventRegistration(
            IEventHandler handler,
            Func<IEventHandler, Envelope, object> interceptor)
        {
            this.Handler = handler;
            _interceptor = interceptor;

            _registrationInfo = new Dictionary<string, string>();
            _registrationInfo.SetMessageTopic(handler.Topic);

            _log = LogManager.GetLogger(this.GetType());
        }