public SubscriberCreator(IPublisher publisher)
 {
     if (publisher==null)
     {
         throw new NullReferenceException("Publisher should not be null");
     }
 }

        protected CoreProjectionCheckpointManager(
            IPublisher publisher, Guid projectionCorrelationId, ProjectionConfig projectionConfig, string name,
            PositionTagger positionTagger, ProjectionNamesBuilder namingBuilder, bool usePersistentCheckpoints,
            bool producesRunningResults)
        {
            if (publisher == null) throw new ArgumentNullException("publisher");
            if (projectionConfig == null) throw new ArgumentNullException("projectionConfig");
            if (name == null) throw new ArgumentNullException("name");
            if (positionTagger == null) throw new ArgumentNullException("positionTagger");
            if (namingBuilder == null) throw new ArgumentNullException("namingBuilder");
            if (name == "") throw new ArgumentException("name");

            _lastProcessedEventPosition = new PositionTracker(positionTagger);
            _zeroTag = positionTagger.MakeZeroCheckpointTag();

            _publisher = publisher;
            _projectionCorrelationId = projectionCorrelationId;
            _projectionConfig = projectionConfig;
            _logger = LogManager.GetLoggerFor<CoreProjectionCheckpointManager>();
            _namingBuilder = namingBuilder;
            _usePersistentCheckpoints = usePersistentCheckpoints;
            _producesRunningResults = producesRunningResults;
            _requestedCheckpointState = new PartitionState("", null, _zeroTag);
            _currentProjectionState = new PartitionState("", null, _zeroTag);
        }

 protected CoreProjectionCheckpointManager(
     ICoreProjection coreProjection, IPublisher publisher, Guid projectionCorrelationId,
     RequestResponseDispatcher
         <ClientMessage.ReadStreamEventsBackward, ClientMessage.ReadStreamEventsBackwardCompleted> readDispatcher,
     RequestResponseDispatcher<ClientMessage.WriteEvents, ClientMessage.WriteEventsCompleted> writeDispatcher,
     ProjectionConfig projectionConfig, string name,
     PositionTagger positionTagger)
 {
     if (coreProjection == null) throw new ArgumentNullException("coreProjection");
     if (publisher == null) throw new ArgumentNullException("publisher");
     if (readDispatcher == null) throw new ArgumentNullException("readDispatcher");
     if (writeDispatcher == null) throw new ArgumentNullException("writeDispatcher");
     if (projectionConfig == null) throw new ArgumentNullException("projectionConfig");
     if (name == null) throw new ArgumentNullException("name");
     if (positionTagger == null) throw new ArgumentNullException("positionTagger");
     if (name == "") throw new ArgumentException("name");
     _lastProcessedEventPosition = new PositionTracker(positionTagger);
     _coreProjection = coreProjection;
     _publisher = publisher;
     _projectionCorrelationId = projectionCorrelationId;
     _readDispatcher = readDispatcher;
     _writeDispatcher = writeDispatcher;
     _projectionConfig = projectionConfig;
     _logger = LogManager.GetLoggerFor<CoreProjectionCheckpointManager>();
     _name = name;
     _positionTagger = positionTagger;
 }

 public EventDistributionPoint CreatePausedEventDistributionPoint(Guid distributionPointId, IPublisher publisher, CheckpointTag checkpointTag)
 {
     if (_allStreams)
     {
         var distributionPoint = new TransactionFileReaderEventDistributionPoint(
             publisher, distributionPointId,
             new EventPosition(checkpointTag.CommitPosition.Value, checkpointTag.PreparePosition.Value));
         return distributionPoint;
     }
     else if (_streams != null && _streams.Count == 1)
     {
         var streamName = checkpointTag.Streams.Keys.First();
         //TODO: handle if not the same
         return CreatePausedStreamReaderEventDistributionPoint(distributionPointId, publisher, checkpointTag, streamName,
             resolveLinkTos: true, category: null);
     }
     else if (_categories != null && _categories.Count == 1)
     {
         var streamName = checkpointTag.Streams.Keys.First();
         return CreatePausedStreamReaderEventDistributionPoint(distributionPointId, publisher, checkpointTag, streamName,
             resolveLinkTos: true, category: _categories.First());
     }
     else
         throw new NotSupportedException();
 }

 public ProjectsController(IProjectsService projects, IUsersService users, IImagesService images, IPublisher publisher)
 {
     this.projects = projects;
     this.users = users;
     this.images = images;
     this.publisher = publisher;
 }

        public EventStoreEmbeddedNodeConnection(ConnectionSettings settings, string connectionName, IPublisher publisher, ISubscriber bus, IAuthenticationProvider authenticationProvider)
        {
            Ensure.NotNull(publisher, "publisher");
            Ensure.NotNull(settings, "settings");

            Guid connectionId = Guid.NewGuid();

            _settings = settings;
            _connectionName = connectionName;
            _publisher = publisher;
            _authenticationProvider = authenticationProvider;
            _subscriptionBus = new InMemoryBus("Embedded Client Subscriptions");
            _subscriptions = new EmbeddedSubscriber(_subscriptionBus, _authenticationProvider, _settings.Log, connectionId);
            
            _subscriptionBus.Subscribe<ClientMessage.SubscriptionConfirmation>(_subscriptions);
            _subscriptionBus.Subscribe<ClientMessage.SubscriptionDropped>(_subscriptions);
            _subscriptionBus.Subscribe<ClientMessage.StreamEventAppeared>(_subscriptions);
            _subscriptionBus.Subscribe<ClientMessage.PersistentSubscriptionConfirmation>(_subscriptions);
            _subscriptionBus.Subscribe<ClientMessage.PersistentSubscriptionStreamEventAppeared>(_subscriptions);
            _subscriptionBus.Subscribe(new AdHocHandler<ClientMessage.SubscribeToStream>(_publisher.Publish));
            _subscriptionBus.Subscribe(new AdHocHandler<ClientMessage.UnsubscribeFromStream>(_publisher.Publish));
            _subscriptionBus.Subscribe(new AdHocHandler<ClientMessage.ConnectToPersistentSubscription>(_publisher.Publish));

            bus.Subscribe(new AdHocHandler<SystemMessage.BecomeShutdown>(_ => Disconnected(this, new ClientConnectionEventArgs(this, new IPEndPoint(IPAddress.None, 0)))));
        }

        public ReadIndex(IPublisher bus,
                         int readerCount,
                         Func<ITransactionFileSequentialReader> seqReaderFactory,
                         Func<ITransactionFileReader> readerFactory,
                         ITableIndex tableIndex,
                         IHasher hasher,
                         ILRUCache<string, StreamCacheInfo> streamInfoCache)
        {
            Ensure.NotNull(bus, "bus");
            Ensure.Positive(readerCount, "readerCount");
            Ensure.NotNull(seqReaderFactory, "seqReaderFactory");
            Ensure.NotNull(readerFactory, "readerFactory");
            Ensure.NotNull(tableIndex, "tableIndex");
            Ensure.NotNull(hasher, "hasher");
            Ensure.NotNull(streamInfoCache, "streamInfoCache");

            _bus = bus;
            _tableIndex = tableIndex;
            _hasher = hasher;
            _streamInfoCache = streamInfoCache;

            for (int i = 0; i < readerCount; ++i)
            {
                _seqReaders.Push(seqReaderFactory());
                _readers.Push(readerFactory());
            }
        }

		public IAuthenticationProvider BuildAuthenticationProvider(IPublisher mainQueue, IBus mainBus, IPublisher workersQueue, InMemoryBus[] workerBusses)
		{
			var passwordHashAlgorithm = new Rfc2898PasswordHashAlgorithm();
			var dispatcher = new IODispatcher(mainQueue, new PublishEnvelope(workersQueue, crossThread: true));
			
			foreach (var bus in workerBusses) {
				bus.Subscribe(dispatcher.ForwardReader);
				bus.Subscribe(dispatcher.BackwardReader);
				bus.Subscribe(dispatcher.Writer);
				bus.Subscribe(dispatcher.StreamDeleter);
				bus.Subscribe(dispatcher);
			}

			// USER MANAGEMENT
			var ioDispatcher = new IODispatcher(mainQueue, new PublishEnvelope(mainQueue));
			mainBus.Subscribe(ioDispatcher.BackwardReader);
			mainBus.Subscribe(ioDispatcher.ForwardReader);
			mainBus.Subscribe(ioDispatcher.Writer);
			mainBus.Subscribe(ioDispatcher.StreamDeleter);
			mainBus.Subscribe(ioDispatcher);

			var userManagement = new UserManagementService(mainQueue, ioDispatcher, passwordHashAlgorithm, skipInitializeStandardUsersCheck: false);
			mainBus.Subscribe<UserManagementMessage.Create>(userManagement);
			mainBus.Subscribe<UserManagementMessage.Update>(userManagement);
			mainBus.Subscribe<UserManagementMessage.Enable>(userManagement);
			mainBus.Subscribe<UserManagementMessage.Disable>(userManagement);
			mainBus.Subscribe<UserManagementMessage.Delete>(userManagement);
			mainBus.Subscribe<UserManagementMessage.ResetPassword>(userManagement);
			mainBus.Subscribe<UserManagementMessage.ChangePassword>(userManagement);
			mainBus.Subscribe<UserManagementMessage.Get>(userManagement);
			mainBus.Subscribe<UserManagementMessage.GetAll>(userManagement);
			mainBus.Subscribe<SystemMessage.BecomeMaster>(userManagement);
			
			return new InternalAuthenticationProvider(dispatcher, passwordHashAlgorithm, ESConsts.CachedPrincipalCount);
		}

        public override sealed IProjectionProcessingPhase[] CreateProcessingPhases(
            IPublisher publisher, Guid projectionCorrelationId, PartitionStateCache partitionStateCache,
            Action updateStatistics, CoreProjection coreProjection, ProjectionNamesBuilder namingBuilder,
            ITimeProvider timeProvider, IODispatcher ioDispatcher,
            CoreProjectionCheckpointWriter coreProjectionCheckpointWriter)
        {
            var definesFold = _sourceDefinition.DefinesFold;

            var readerStrategy = CreateReaderStrategy(timeProvider);

            var zeroCheckpointTag = readerStrategy.PositionTagger.MakeZeroCheckpointTag();

            var checkpointManager = CreateCheckpointManager(
                projectionCorrelationId, publisher, ioDispatcher, namingBuilder, coreProjectionCheckpointWriter,
                definesFold, readerStrategy);


            var resultWriter = CreateFirstPhaseResultWriter(
                checkpointManager as IEmittedEventWriter, zeroCheckpointTag, namingBuilder);

            var firstPhase = CreateFirstProcessingPhase(
                publisher, projectionCorrelationId, partitionStateCache, updateStatistics, coreProjection,
                _subscriptionDispatcher, zeroCheckpointTag, checkpointManager, readerStrategy, resultWriter);

            return CreateProjectionProcessingPhases(
                publisher, projectionCorrelationId, namingBuilder, partitionStateCache, coreProjection, ioDispatcher,
                firstPhase);
        }

		public ApiModule(ICompositeApp compositeApp, ICommandRegistry registry, IPublisher publisher)
			: base(BaseRoute)
		{
			_compositeApp = compositeApp;
			_registry = registry;
			_publisher = publisher;

			Get[string.Empty] = _ => GetComposite();

			Get[AgentMetadataRoute] = p => GetAgent(stripExtension((string)p.agentSystemName));

			Get[ReadModelMetadataRoute] = p => GetReadModel((string)p.agentSystemName, stripExtension((string)p.readModelName));

			Get[InputModelMetadataRoute] = p => GetInputModel(stripExtension((string)p.inputModelName));

			Get[PublicationRecordRoute] = p => GetPublicationRecord((Guid)p.identifier);

			Get[InputModelMetadataForCommandRoute] = p => GetInputModelMetadata(stripExtension((string)p.commandName));

			Get[CommandMetadataRoute] = p => GetCommandMetadata(stripExtension((string)p.commandName));

			Get[QueryMetadataRoute] = p => GetQueryMetadata(stripExtension((string)p.queryName));

			Get[CommandIsSupportedRoute] = p => GetCommandIsSupported(stripExtension((string)p.commandName));

			Get[QueryForReadModelRoute] = p => QueryForReadmodel((string)p.queryName, (string)p.methodName);

			Get[QueryMethodRoute] = p => GetQueryMethod((string) p.queryName, (string) p.methodName);

			Post[PublishCommandRoute] = p => PublishCommand(this.Bind<IInputModel>());
		}

        public Subscriber(IPublisher publisher, TextWriter textWriter)
        {
            _textWriter = textWriter;

            if (publisher != null)
                publisher.WriteEvent += HandleWriteEvent;
        }

        public MainWindow(ISubscriber<PlayerCommand> commandChannel, IPublisher<RichYouTubeClip> finishedChannel)
        {
            InitializeComponent();
            this.finishedChannel = finishedChannel;

            finishedTimer = new Timer(TimerExpired);

            progressTimer = new System.Timers.Timer(1000);
            progressTimer.AutoReset = true;
            progressTimer.Elapsed += new System.Timers.ElapsedEventHandler(progressTimer_Elapsed);

            var fiber     = new DispatcherFiber(Dispatcher);
            commandChannel.Subscribe(fiber, OnPlayerCommand);
            fiber.Start();

            WindowStyle = WindowStyle.None;
            WindowState = WindowState.Maximized;
            ShowInTaskbar = false;

            Cursor = Cursors.None;
            try
            {
                tv.Initialize();
            }
            catch (Exception ex)
            {
                MessageBox.Show(ex.Message, "Error initialize remote control", MessageBoxButton.OK, MessageBoxImage.Error);
            }
        }

        public SlaveResultWriter(IPublisher resultsPublisher, Guid masterCoreProjectionId)
        {
            if (resultsPublisher == null) throw new ArgumentNullException("resultsPublisher");

            _resultsPublisher = resultsPublisher;
            _masterCoreProjectionId = masterCoreProjectionId;
        }

 public static void DispatchCommit(IPublisher bus, ICommit commit)
 {
     foreach (var @event in commit.Events)
     {
         bus.Publish(@event.Body);
     }
 }

 public static void Bin4Net(IPublisher publisher)
 {
     publisher
     .SetupMetadata(m => m
                       .FromAssemblyAttributes()
                       .AssociateWithTags("tool", "acme"));
 }

        public WriteQueryResultProjectionProcessingPhaseBase(
            IPublisher publisher,
            int phase,
            string resultStream,
            ICoreProjectionForProcessingPhase coreProjection,
            PartitionStateCache stateCache,
            ICoreProjectionCheckpointManager checkpointManager,
            IEmittedEventWriter emittedEventWriter,
            IEmittedStreamsTracker emittedStreamsTracker)
        {
            if (resultStream == null) throw new ArgumentNullException("resultStream");
            if (coreProjection == null) throw new ArgumentNullException("coreProjection");
            if (stateCache == null) throw new ArgumentNullException("stateCache");
            if (checkpointManager == null) throw new ArgumentNullException("checkpointManager");
            if (emittedEventWriter == null) throw new ArgumentNullException("emittedEventWriter");
            if (emittedStreamsTracker == null) throw new ArgumentNullException("emittedStreamsTracker");
            if (string.IsNullOrEmpty(resultStream)) throw new ArgumentException("resultStream");

            _publisher = publisher;
            _phase = phase;
            _resultStream = resultStream;
            _coreProjection = coreProjection;
            _stateCache = stateCache;
            _checkpointManager = checkpointManager;
            _emittedEventWriter = emittedEventWriter;
            _emittedStreamsTracker = emittedStreamsTracker;
        }

        protected ReaderSubscriptionBase(
            IPublisher publisher,
            Guid subscriptionId,
            CheckpointTag @from,
            IReaderStrategy readerStrategy,
            ITimeProvider timeProvider,
            long? checkpointUnhandledBytesThreshold,
            int? checkpointProcessedEventsThreshold,
            bool stopOnEof,
            int? stopAfterNEvents)
        {
            if (publisher == null) throw new ArgumentNullException("publisher");
            if (readerStrategy == null) throw new ArgumentNullException("readerStrategy");
            if (timeProvider == null) throw new ArgumentNullException("timeProvider");
            if (checkpointProcessedEventsThreshold > 0 && stopAfterNEvents > 0)
                throw new ArgumentException("checkpointProcessedEventsThreshold > 0 && stopAfterNEvents > 0");

            _publisher = publisher;
            _readerStrategy = readerStrategy;
            _timeProvider = timeProvider;
            _checkpointUnhandledBytesThreshold = checkpointUnhandledBytesThreshold;
            _checkpointProcessedEventsThreshold = checkpointProcessedEventsThreshold;
            _stopOnEof = stopOnEof;
            _stopAfterNEvents = stopAfterNEvents;
            _subscriptionId = subscriptionId;
            _lastPassedOrCheckpointedEventPosition = null;

            _eventFilter = readerStrategy.EventFilter;

            _positionTagger = readerStrategy.PositionTagger;
            _positionTracker = new PositionTracker(_positionTagger);
            _positionTracker.UpdateByCheckpointTagInitial(@from);
        }

        protected CoreProjectionCheckpointManager(
            ICoreProjection coreProjection, IPublisher publisher, Guid projectionCorrelationId,
            RequestResponseDispatcher<ClientMessage.ReadStreamEventsBackward, ClientMessage.ReadStreamEventsBackwardCompleted> readDispatcher,
            RequestResponseDispatcher<ClientMessage.WriteEvents, ClientMessage.WriteEventsCompleted> writeDispatcher,
            ProjectionConfig projectionConfig, string name, PositionTagger positionTagger,
            ProjectionNamesBuilder namingBuilder, bool useCheckpoints, bool emitStateUpdated,
            bool emitPartitionCheckpoints)
        {
            if (coreProjection == null) throw new ArgumentNullException("coreProjection");
            if (publisher == null) throw new ArgumentNullException("publisher");
            if (readDispatcher == null) throw new ArgumentNullException("readDispatcher");
            if (writeDispatcher == null) throw new ArgumentNullException("writeDispatcher");
            if (projectionConfig == null) throw new ArgumentNullException("projectionConfig");
            if (name == null) throw new ArgumentNullException("name");
            if (positionTagger == null) throw new ArgumentNullException("positionTagger");
            if (namingBuilder == null) throw new ArgumentNullException("namingBuilder");
            if (name == "") throw new ArgumentException("name");

            if (emitPartitionCheckpoints && emitStateUpdated)
                throw new InvalidOperationException("EmitPartitionCheckpoints && EmitStateUpdated cannot be both set");
            _lastProcessedEventPosition = new PositionTracker(positionTagger);
            _coreProjection = coreProjection;
            _publisher = publisher;
            _projectionCorrelationId = projectionCorrelationId;
            _readDispatcher = readDispatcher;
            _writeDispatcher = writeDispatcher;
            _projectionConfig = projectionConfig;
            _logger = LogManager.GetLoggerFor<CoreProjectionCheckpointManager>();
            _name = name;
            _positionTagger = positionTagger;
            _namingBuilder = namingBuilder;
            _useCheckpoints = useCheckpoints;
            _emitStateUpdated = emitStateUpdated;
            _emitPartitionCheckpoints = emitPartitionCheckpoints;
        }

 public ElectionsInstance(Guid instanceId, IPEndPoint endPoint, IPublisher inputBus, IPublisher outputBus)
 {
     InstanceId = instanceId;
     EndPoint = endPoint;
     InputBus = inputBus;
     OutputBus = outputBus;
 }

        protected override IProjectionProcessingPhase[] CreateProjectionProcessingPhases(
            IPublisher publisher, Guid projectionCorrelationId, ProjectionNamesBuilder namingBuilder,
            PartitionStateCache partitionStateCache, CoreProjection coreProjection, IODispatcher ioDispatcher,
            IProjectionProcessingPhase firstPhase)
        {

            var coreProjectionCheckpointWriter =
                new CoreProjectionCheckpointWriter(
                    namingBuilder.MakeCheckpointStreamName(), ioDispatcher, _projectionVersion, _name);
            var checkpointManager2 = new DefaultCheckpointManager(
                publisher, projectionCorrelationId, _projectionVersion, _projectionConfig.RunAs, ioDispatcher,
                _projectionConfig, _name, new PhasePositionTagger(1), namingBuilder, GetUseCheckpoints(), false,
                _sourceDefinition.DefinesFold, coreProjectionCheckpointWriter);

            IProjectionProcessingPhase writeResultsPhase;
            if (GetProducesRunningResults()
                || !string.IsNullOrEmpty(_sourceDefinition.CatalogStream) && _sourceDefinition.ByStreams)
                writeResultsPhase = new WriteQueryEofProjectionProcessingPhase(
                    1, namingBuilder.GetResultStreamName(), coreProjection, partitionStateCache, checkpointManager2,
                    checkpointManager2);
            else
                writeResultsPhase = new WriteQueryResultProjectionProcessingPhase(
                    1, namingBuilder.GetResultStreamName(), coreProjection, partitionStateCache, checkpointManager2,
                    checkpointManager2);

            return new[] {firstPhase, writeResultsPhase};
        }