public SubscriptionClient(IServiceBus bus, SubscriptionRouter router, Uri subscriptionServiceUri,
		                          TimeSpan startTimeout)
		{
			_bus = bus;
			_router = router;
			_subscriptionServiceUri = subscriptionServiceUri;
			_startTimeout = startTimeout;
			_network = router.Network;

			if (_log.IsDebugEnabled)
				_log.DebugFormat("Starting SubscriptionClient using {0}", subscriptionServiceUri);

			VerifyClientAndServiceNotOnSameEndpoint(bus);

			_ready.Reset();

			var consumerInstance = new SubscriptionMessageConsumer(_router, _network);

			_unsubscribeAction = _bus.ControlBus.SubscribeInstance(consumerInstance);
		    _unsubscribeAction += _bus.ControlBus.SubscribeContextHandler<SubscriptionRefresh>(Consume);

		    _subscriptionEndpoint = _bus.GetEndpoint(subscriptionServiceUri);
		    _producer = new SubscriptionServiceMessageProducer(router, _subscriptionEndpoint);

			WaitForSubscriptionServiceResponse();
		}

		public ThreadPoolConsumerPool(IServiceBus bus, IObjectBuilder objectBuilder, Pipe eventAggregator, TimeSpan receiveTimeout)
		{
			_objectBuilder = objectBuilder;
			_receiveTimeout = receiveTimeout;
			_eventAggregator = eventAggregator;
			_bus = bus;
		}

        public void Initialize(IServiceBus bus)
        {
            if (ServiceBusConfiguration.ServiceBusSection == null || ServiceBusConfiguration.ServiceBusSection.ForwardingRoutes == null)
            {
                return;
            }

            var factory = new MessageRouteSpecificationFactory();

            foreach (MessageRouteElement mapElement in ServiceBusConfiguration.ServiceBusSection.ForwardingRoutes)
            {
                var map = messageRoutes.Find(mapElement.Uri);

                if (map == null)
                {
                    map = new MessageRoute(bus.Configuration.QueueManager.GetQueue(mapElement.Uri));

                    messageRoutes.Add(map);
                }

                foreach (SpecificationElement specificationElement in mapElement)
                {
                    map.AddSpecification(factory.Create(specificationElement.Name, specificationElement.Value));
                }
            }
        }

		public SubscriptionRouterService(IServiceBus bus, string network)
		{
			_peerUri = bus.ControlBus.Endpoint.Address.Uri;

			_network = network;

			_peerId = CombGuid.Generate();

			_observers = new List<SubscriptionObserver>();
			_listeners = new List<BusSubscriptionEventListener>();

			_unregister = () => true;

			_peerUri = bus.ControlBus.Endpoint.Address.Uri;

			var connector = new BusSubscriptionConnector(bus);

			_peerCache = ActorFactory.Create<PeerCache>(x =>
				{
					x.ConstructedBy((fiber, scheduler, inbox) =>
					                new PeerCache(fiber, scheduler, connector, _peerId, _peerUri));
					x.UseSharedScheduler();
					x.HandleOnPoolFiber();
				})
				.GetActor();
		}

		public MulticastSubscriptionClient(IServiceBus subscriptionBus, Uri uri, string networkKey)
		{
			_subscriptionBus = subscriptionBus;
			_uri = uri;

			_networkKey = networkKey;
		}

 public TaskMonitoringSource(HealthMonitoringSettings settings, ILogger logger, IServiceBus serviceBus, ISubscriptionRepository repository)
 {
     _settings = settings;
     _logger = logger;
     _serviceBus = serviceBus;
     _repository = repository;
 }

        public void SetUp()
        {
            // Need to do something about this.  Little ridiculous
            FubuTransport.SetupForInMemoryTesting();
            TestMessageRecorder.Clear();
            MessageHistory.ClearAll();
            InMemoryQueueManager.ClearAll();

            runtime = FubuTransport.For<DelayedRegistry>().StructureMap(new Container())
                                       .Bootstrap();

            theServiceBus = runtime.Factory.Get<IServiceBus>();

            theClock = runtime.Factory.Get<ISystemTime>().As<SettableClock>();

            message1 = new OneMessage();
            message2 = new OneMessage();
            message3 = new OneMessage();
            message4 = new OneMessage();

            theServiceBus.DelaySend(message1, theClock.UtcNow().AddHours(1));
            theServiceBus.DelaySend(message2, theClock.UtcNow().AddHours(1));
            theServiceBus.DelaySend(message3, theClock.UtcNow().AddHours(2));
            theServiceBus.DelaySend(message4, theClock.UtcNow().AddHours(2));
        }

 public LoginController(IUserRepository userRepository, IServiceBus bus, ITranslationService translationService, IPasswordPolicy passwordPolicy)
 {
     _userRepository = userRepository;
     _bus = bus;
     _translationService = translationService;
     _passwordPolicy = passwordPolicy;
 }

 public CreateCustomerViewModel(IServiceBus bus, IEventAggregator eventAggregator)
 {
     _Bus = bus;
     _EventAggregator = eventAggregator;
     Command = new CreateNewCustomer(CombGuid.Generate(),
         "unknown", "unknown", "unknown", "unknown", "unknown", "305533333");
 }

        public SubscriptionRouterService(IServiceBus bus, SubscriptionRepository repository, string network)
        {
            _peerUri = bus.ControlBus.Endpoint.Address.Uri;

            _repository = repository;
            _network = network;

            _peerId = NewId.NextGuid();

            _observers = new List<SubscriptionObserver>();
            _listeners = new List<BusSubscriptionEventListener>();

            _unregister = () => true;

            _peerUri = bus.ControlBus.Endpoint.Address.Uri;

            var connector = new BusSubscriptionConnector(bus);

            _peerCache = ActorFactory.Create<PeerCache>(x =>
                {
                    x.ConstructedBy((fiber, scheduler, inbox) =>
                                    new PeerCache(connector, _peerId, _peerUri, repository));
                    x.UseSharedScheduler();
                    x.HandleOnPoolFiber();
                })
                .GetActor();

            // at this point, existing subscriptions need to be loaded...

            _repository.Load(this);
        }

        public SendMessagePipeline(IServiceBus bus)
            : base(bus)
        {
            RegisterStage("Send")
                .WithEvent<OnPrepareMessage>()
                .WithEvent<OnAfterPrepareMessage>()
                .WithEvent<OnFindRouteForMessage>()
                .WithEvent<OnAfterFindRouteForMessage>()
                .WithEvent<OnSerializeMessage>()
                .WithEvent<OnAfterSerializeMessage>()
                .WithEvent<OnEncryptMessage>()
                .WithEvent<OnAfterEncryptMessage>()
                .WithEvent<OnCompressMessage>()
                .WithEvent<OnAfterCompressMessage>()
                .WithEvent<OnSerializeTransportMessage>()
                .WithEvent<OnAfterSerializeTransportMessage>()
                .WithEvent<OnSendMessage>()
                .WithEvent<OnAfterSendMessage>();

            RegisterObserver(new PrepareMessageObserver());
            RegisterObserver(new FindMessageRouteObserver());
            RegisterObserver(new SerializeMessageObserver());
            RegisterObserver(new SerializeTransportMessageObserver());
            RegisterObserver(new CompressMessageObserver());
            RegisterObserver(new EncryptMessageObserver());
            RegisterObserver(new SendMessageObserver());
        }

        public TodoModule(IAppSettings appSettings, ITodoService todoService, IServiceBus bus)
        {
            _todoService = todoService;
            _bus = bus;

            Post["/todo"] = _ =>
            {
                var slashCommand = this.Bind<SlashCommand>();
                if (slashCommand == null ||
                    slashCommand.command.Missing())
                {
                    Log.Info("Rejected an incoming slash command (unable to parse request body).");
                    return HttpStatusCode.BadRequest.WithReason("Unable to parse slash command.");
                }
                if (!appSettings.Get("todo:slackSlashCommandToken").Equals(slashCommand.token))
                {
                    Log.Info("Blocked an unauthorized slash command.");
                    return HttpStatusCode.Unauthorized.WithReason("Missing or invalid token.");
                }
                if (!slashCommand.command.Equals("/todo", StringComparison.InvariantCultureIgnoreCase))
                {
                    Log.Info("Rejected an incoming slash command ({0} is not handled by this module).", slashCommand.command);
                    return HttpStatusCode.BadRequest.WithReason("Unsupported slash command.");
                }

                var responseText = HandleTodo(slashCommand);
                if (responseText.Missing())
                {
                    return HttpStatusCode.OK;
                }
                return responseText;
            };
        }

        public DistributedHashTable(
            string database,
            Uri url,
            IEndpointRouter endpointRouter,
            IServiceBus bus,
            Node metadata)
        {
            Url = url;
            this.endpointRouter = endpointRouter;
            this.bus = bus;
            Metadata = metadata;

            if (Metadata != null) // sole node in the network, probably
            {
                Metadata.ExecuteSync(uri =>
                {
                    ServiceUtil.Execute<IDistributedHashTableMetaDataProvider>(uri, srv =>
                    {
                        failOver = srv.GetNodeByUri(url);
                    });
                });
            }
            try
            {
                hashTable = new PersistentHashTable(database);
                hashTable.Initialize();
            }
            catch (Exception)
            {
                hashTable.Dispose();
                throw;
            }
        }

 public WebSocketHandler(IServiceBus bus, WebSocket ws, IEventSerializator serializer, ILogger log)
 {
     _ws = ws;
     _queue = bus;
     _log = log;
     _serializer = serializer;
 }

 public void Promoted(IServiceBus bus)
 {
     if (_scheduler != null && !_scheduler.IsShutdown) _scheduler.Shutdown();
     _scheduler = _schedulerFactory();
     _scheduler.JobFactory = new ServiceBusJobFactory(bus);
     _scheduler.Start();
 }

        public static void RegisterHandlers(IReadOnlyStore readOnlyStore, IServiceBus serviceBus)
        {
            _serviceBus = serviceBus;

            RegisterCommandHandler(new ProductListCommandHandlers());
            RegisterEventHandler(new ProductListEventHandlers(readOnlyStore));
        }

		public override IServiceBus GetDecoratedBus(IServiceBus bus)
		{
			if (_realBus == bus)
				return _bus;

			return base.GetDecoratedBus(bus);
		}

        public void Initialize(IServiceBus serviceBus)
        {
            Guard.AgainstNull(serviceBus, "serviceBus");

            AddAssertion("BeforeDequeueStream", Assertion);
            AddAssertion("AfterDequeueStream", Assertion);
            AddAssertion("AfterMessageDeserialization", Assertion);
            AddAssertion("BeforeEnqueueStream", Assertion);
            AddAssertion("AfterEnqueueStream", Assertion);
            AddAssertion("BeforeHandleMessage", Assertion);
            AddAssertion("AfterHandleMessage", Assertion);
            AddAssertion("BeforeRemoveMessage", Assertion);
            AddAssertion("AfterRemoveMessage", Assertion);

            serviceBus.Events.BeforeDequeueStream += (sender, e) => ThrowException("BeforeDequeueStream");
            serviceBus.Events.AfterDequeueStream += (sender, e) => ThrowException("AfterDequeueStream");
            serviceBus.Events.AfterMessageDeserialization += (sender, e) => ThrowException("AfterMessageDeserialization");
            serviceBus.Events.BeforeEnqueueStream += (sender, e) => ThrowException("BeforeEnqueueStream");
            serviceBus.Events.AfterEnqueueStream += (sender, e) => ThrowException("AfterEnqueueStream");
            serviceBus.Events.BeforeHandleMessage += (sender, e) => ThrowException("BeforeHandleMessage");
            serviceBus.Events.AfterHandleMessage += (sender, e) => ThrowException("AfterHandleMessage");
            serviceBus.Events.BeforeRemoveMessage += (sender, e) => ThrowException("BeforeRemoveMessage");
            serviceBus.Events.AfterRemoveMessage += (sender, e) => ThrowException("AfterRemoveMessage");

            serviceBus.Events.PipelineReleased += PipelineReleased;
        }

 public SimpleEventStore(IStoreSettings<IDbConnection> settings, IServiceBus serviceBus, ISerialize serializer, ISnapshotStore snapshotStore)
 {
     _settings = settings;
     _serviceBus = serviceBus;
     _serializer = serializer;
     _snapshotStore = snapshotStore;
 }

        /// <summary>
        /// 
        /// </summary>
        /// <param name="serviceBus">The service bus on which commands are received</param>
        /// <param name="controlBus">The control bus for the synchronization messages between nodes</param>
        public void Start(IServiceBus serviceBus, IServiceBus controlBus)
        {
            _serviceBus = serviceBus;
            _controlBus = controlBus;

            _nodeState.ChangeTo<FollowerBehavior>();
        }