public ParallelQueryProcessingStrategy(
     string name, ProjectionVersion projectionVersion, IProjectionStateHandler stateHandler,
     Func<IProjectionStateHandler> handlerFactory, ProjectionConfig projectionConfig,
     IQueryDefinition sourceDefinition, ProjectionNamesBuilder namesBuilder, ILogger logger,
     SpooledStreamReadingDispatcher spoolProcessingResponseDispatcher,
     ReaderSubscriptionDispatcher subscriptionDispatcher)
     : base(name, projectionVersion, projectionConfig, sourceDefinition, logger, subscriptionDispatcher)
 {
     _stateHandler = stateHandler;
     _projectionConfig = projectionConfig;
     _sourceDefinition = sourceDefinition;
     _namesBuilder = namesBuilder;
     _spoolProcessingResponseDispatcher = spoolProcessingResponseDispatcher;
     if (_sourceDefinition.CatalogStream == SystemStreams.AllStream)
     {
         _catalogStreamName = SystemStreams.AllStream;
     }
     else if (_sourceDefinition.HasCategories())
     {
         _catalogStreamName = _namesBuilder.GetCategoryCatalogStreamName(_sourceDefinition.Categories[0]);
     }
     else
     {
         _catalogStreamName = _sourceDefinition.CatalogStream;
     }
 }

        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};
        }

        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);
        }

        public ProjectionProcessingStrategy CreateProjectionProcessingStrategy(
            string name, ProjectionVersion projectionVersion, ProjectionNamesBuilder namesBuilder,
            IQueryDefinition sourceDefinition, ProjectionConfig projectionConfig,
            Func<IProjectionStateHandler> handlerFactory, IProjectionStateHandler stateHandler)
        {

            if (!sourceDefinition.DisableParallelismOption && projectionConfig.StopOnEof && sourceDefinition.ByStreams
                && sourceDefinition.DefinesFold && !string.IsNullOrEmpty(sourceDefinition.CatalogStream))
            {
                return new ParallelQueryProcessingStrategy(
                    name, projectionVersion, stateHandler, handlerFactory, projectionConfig, sourceDefinition,
                    namesBuilder, _logger, _spoolProcessingResponseDispatcher, _subscriptionDispatcher);
            }

            if (!sourceDefinition.DisableParallelismOption && projectionConfig.StopOnEof && sourceDefinition.ByStreams
                && sourceDefinition.DefinesFold && sourceDefinition.HasCategories())
            {
                return new ParallelQueryProcessingStrategy(
                    name, projectionVersion, stateHandler, handlerFactory, projectionConfig, sourceDefinition,
                    namesBuilder, _logger, _spoolProcessingResponseDispatcher, _subscriptionDispatcher);
            }

            return projectionConfig.StopOnEof
                ? (ProjectionProcessingStrategy)
                    new QueryProcessingStrategy(
                        name, projectionVersion, stateHandler, projectionConfig, sourceDefinition, _logger,
                        _subscriptionDispatcher)
                : new ContinuousProjectionProcessingStrategy(
                    name, projectionVersion, stateHandler, projectionConfig, sourceDefinition, _logger,
                    _subscriptionDispatcher);
        }

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

        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);
        }

 protected override ICoreProjectionCheckpointManager CreateCheckpointManager(
     Guid projectionCorrelationId, IPublisher publisher, IODispatcher ioDispatcher, ProjectionNamesBuilder namingBuilder,
     CoreProjectionCheckpointWriter coreProjectionCheckpointWriter, bool definesFold, IReaderStrategy readerStrategy)
 {
     return new NoopCheckpointManager(
         publisher, projectionCorrelationId, _projectionConfig, _name, readerStrategy.PositionTagger,
         namingBuilder);
 }

 private void Given()
 {
     _projection = "projection";
     _partitionAt = CheckpointTag.FromPosition(100, 50);
     _partition = "partition";
     _namesBuilder = ProjectionNamesBuilder.CreateForTest(_projection);
     _re = new ResultEmitter(_namesBuilder);
 }

 public override IProjectionProcessingPhase[] CreateProcessingPhases(
     IPublisher publisher, Guid projectionCorrelationId, PartitionStateCache partitionStateCache,
     Action updateStatistics, CoreProjection coreProjection, ProjectionNamesBuilder namingBuilder,
     ITimeProvider timeProvider, IODispatcher ioDispatcher,
     CoreProjectionCheckpointWriter coreProjectionCheckpointWriter)
 {
     return new IProjectionProcessingPhase[] {_phase1, _phase2};
 }

 public void setup()
 {
     Given();
     _namingBuilder = ProjectionNamesBuilder.CreateForTest("projection");
     _config = new ProjectionConfig(null, _checkpointHandledThreshold, _checkpointUnhandledBytesThreshold,
         _pendingEventsThreshold, _maxWriteBatchLength, _emitEventEnabled,
         _checkpointsEnabled, _createTempStreams, _stopOnEof, isSlaveProjection: false);
     When();
 }

 public abstract IProjectionProcessingPhase[] CreateProcessingPhases(
     IPublisher publisher,
     IPublisher inputQueue,
     Guid projectionCorrelationId,
     PartitionStateCache partitionStateCache,
     Action updateStatistics,
     CoreProjection coreProjection,
     ProjectionNamesBuilder namingBuilder,
     ITimeProvider timeProvider,
     IODispatcher ioDispatcher,
     CoreProjectionCheckpointWriter coreProjectionCheckpointWriter);

 public MultiStreamMultiOutputCheckpointManager(
     IPublisher publisher, Guid projectionCorrelationId, ProjectionVersion projectionVersion, IPrincipal runAs,
     IODispatcher ioDispatcher, ProjectionConfig projectionConfig, string name, PositionTagger positionTagger,
     ProjectionNamesBuilder namingBuilder, bool useCheckpoints, bool producesRunningResults, bool definesFold,
     CoreProjectionCheckpointWriter coreProjectionCheckpointWriter)
     : base(
         publisher, projectionCorrelationId, projectionVersion, runAs, ioDispatcher, projectionConfig, name,
         positionTagger, namingBuilder, useCheckpoints, producesRunningResults, definesFold,
         coreProjectionCheckpointWriter)
 {
     _positionTagger = positionTagger;
 }

 public ParallelQueryProcessingStrategy(
     string name, ProjectionVersion projectionVersion, Func<IProjectionStateHandler> handlerFactory,
     ProjectionConfig projectionConfig, IQueryDefinition sourceDefinition, ProjectionNamesBuilder namesBuilder,
     ILogger logger, SpooledStreamReadingDispatcher spoolProcessingResponseDispatcher,
     ReaderSubscriptionDispatcher subscriptionDispatcher)
     : base(name, projectionVersion, projectionConfig, sourceDefinition, logger, subscriptionDispatcher)
 {
     _projectionConfig = projectionConfig;
     _sourceDefinition = sourceDefinition;
     _namesBuilder = namesBuilder;
     _spoolProcessingResponseDispatcher = spoolProcessingResponseDispatcher;
 }

 protected override void Given()
 {
     _ioDispatcher = new IODispatcher(_node.Node.MainQueue, new PublishEnvelope(_node.Node.MainQueue));
     _node.Node.MainBus.Subscribe(_ioDispatcher.BackwardReader);
     _node.Node.MainBus.Subscribe(_ioDispatcher.ForwardReader);
     _node.Node.MainBus.Subscribe(_ioDispatcher.Writer);
     _node.Node.MainBus.Subscribe(_ioDispatcher.StreamDeleter);
     _node.Node.MainBus.Subscribe(_ioDispatcher.Awaker);
     _node.Node.MainBus.Subscribe(_ioDispatcher);
     _projectionNamesBuilder = ProjectionNamesBuilder.CreateForTest(_projectionName);
     _emittedStreamsTracker = new EmittedStreamsTracker(_ioDispatcher, new ProjectionConfig(null, 1000, 1000 * 1000, 100, 500, true, true, false, false, false, _trackEmittedStreams), _projectionNamesBuilder);
     _emittedStreamsDeleter = new EmittedStreamsDeleter(_ioDispatcher, _projectionNamesBuilder.GetEmittedStreamsName(), _projectionNamesBuilder.GetEmittedStreamsCheckpointName());
 }

 public DefaultCheckpointManager(
     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 = false)
     : base(
         coreProjection, publisher, projectionCorrelationId, readDispatcher, writeDispatcher, projectionConfig,
         name, positionTagger, namingBuilder, useCheckpoints, emitStateUpdated, emitPartitionCheckpoints)
 {
     _projectionCheckpointStreamId = namingBuilder.MakeCheckpointStreamName();
 }

 public MultiStreamMultiOutputCheckpointManager(
     IPublisher publisher, Guid projectionCorrelationId, ProjectionVersion projectionVersion, IPrincipal runAs,
     RequestResponseDispatcher
         <ClientMessage.ReadStreamEventsBackward, ClientMessage.ReadStreamEventsBackwardCompleted> readDispatcher,
     RequestResponseDispatcher<ClientMessage.WriteEvents, ClientMessage.WriteEventsCompleted> writeDispatcher,
     ProjectionConfig projectionConfig, string name, PositionTagger positionTagger,
     ProjectionNamesBuilder namingBuilder, IResultEmitter resultEmitter, bool useCheckpoints,
     bool emitPartitionCheckpoints = false)
     : base(
         publisher, projectionCorrelationId, projectionVersion, runAs, readDispatcher, writeDispatcher,
         projectionConfig, name, positionTagger, namingBuilder, resultEmitter, useCheckpoints,
         emitPartitionCheckpoints)
 {
     _positionTagger = positionTagger;
 }

        public NoopCheckpointManager(
            IPublisher publisher, Guid projectionCorrelationId, ProjectionConfig projectionConfig, string name,
            PositionTagger positionTagger, ProjectionNamesBuilder namingBuilder)
        {
            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);

            _publisher = publisher;
            _projectionCorrelationId = projectionCorrelationId;
        }

        public CoreProjection Create(
            Guid projectionCorrelationId,
            IPublisher inputQueue,
            Guid workerId,
            IPrincipal runAs,
            IPublisher publisher,
            IODispatcher ioDispatcher,
            ReaderSubscriptionDispatcher subscriptionDispatcher,
            ITimeProvider timeProvider)
        {
            if (inputQueue == null) throw new ArgumentNullException("inputQueue");
            //if (runAs == null) throw new ArgumentNullException("runAs");
            if (publisher == null) throw new ArgumentNullException("publisher");
            if (ioDispatcher == null) throw new ArgumentNullException("ioDispatcher");
            if (timeProvider == null) throw new ArgumentNullException("timeProvider");

            var namingBuilder = new ProjectionNamesBuilder(_name, GetSourceDefinition());

            var coreProjectionCheckpointWriter =
                new CoreProjectionCheckpointWriter(
                    namingBuilder.MakeCheckpointStreamName(),
                    ioDispatcher,
                    _projectionVersion,
                    namingBuilder.EffectiveProjectionName);

            var partitionStateCache = new PartitionStateCache();

            return new CoreProjection(
                this,
                _projectionVersion,
                projectionCorrelationId,
                inputQueue,
                workerId,
                runAs,
                publisher,
                ioDispatcher,
                subscriptionDispatcher,
                _logger,
                namingBuilder,
                coreProjectionCheckpointWriter,
                partitionStateCache,
                namingBuilder.EffectiveProjectionName,
                timeProvider);
        }

 public ProjectionSourceDefinition Build(ProjectionNamesBuilder namingBuilder)
 {
     return new ProjectionSourceDefinition
         {
             AllEvents = _allEvents,
             AllStreams = _allStreams,
             ByStream = _byStream,
             ByCustomPartitions = _byCustomPartitions,
             Categories = (_categories ?? new List<string>()).ToArray(),
             Events = (_events ?? new List<string>()).ToArray(),
             Streams = (_streams ?? new List<string>()).ToArray(),
             DefinesStateTransform = _definesStateTransform,
             Options = _options,
             ResultStreamName = namingBuilder.GetResultStreamName(),
             PartitionResultStreamNamePattern = namingBuilder.GetPartitionResultStreamNamePattern(),
             PartitionResultCatalogStream = namingBuilder.GetPartitionResultCatalogStreamName(),
             PartitionCatalogStream = namingBuilder.GetPartitionCatalogStreamName(),
         };
 }

        protected override IProjectionProcessingPhase[] CreateProjectionProcessingPhases(
            IPublisher publisher,
            IPublisher inputQueue,
            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);

            var writeResultsPhase = new WriteQueryEofProjectionProcessingPhase(
                publisher,
                1,
                namingBuilder.GetResultStreamName(),
                coreProjection,
                partitionStateCache,
                checkpointManager2,
                checkpointManager2,
                firstPhase.EmittedStreamsTracker);

            return new[] {firstPhase, writeResultsPhase};
        }