public void WriteEofResult(
     Guid subscriptionId, string partition, string resultBody, CheckpointTag causedBy, Guid causedByGuid,
     string correlationId)
 {
     if (resultBody != null)
         WriteResult(partition, resultBody, causedBy, causedByGuid, correlationId);
 }

        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 DataReportBase(Guid correlationId, Guid projectionId, string partition, CheckpointTag position)
     : base(projectionId)
 {
     _correlationId = correlationId;
     _partition = partition;
     _position = position;
 }

 public EmittedLinkTo(
     string streamId, Guid eventId,
     string targetStreamId, CheckpointTag causedByTag, CheckpointTag expectedTag, Action<int> onCommitted = null)
     : base(streamId, eventId, "$>", causedByTag, expectedTag, onCommitted)
 {
     _targetStreamId = targetStreamId;
 }

        public bool ProcessEvent(
            string partition, CheckpointTag eventPosition, string category, ResolvedEvent data, out string newState,
            out string newSharedState, out EmittedEventEnvelope[] emittedEvents)
        {
            newSharedState = null;
            if (!data.EventStreamId.StartsWith(UserStreamPrefix))
                throw new InvalidOperationException(
                    string.Format(
                        "Invalid stream name: '{0}' The IndexUsersProjectionHandler cannot handle events from other streams than named after the '$user-' pattern",
                        data.EventStreamId));

            var loginName = data.EventStreamId.Substring(UserStreamPrefix.Length);

            var userData = data.Data.ParseJson<UserData>();
            if (userData.LoginName != loginName)
                throw new InvalidOperationException(
                    string.Format(
                        "Invalid $UserCreated event found.  '{0}' login name expected, but '{1}' found", loginName,
                        userData.LoginName));

            emittedEvents = new[]
            {
                new EmittedEventEnvelope(
                    new EmittedDataEvent(
                        UsersStream, Guid.NewGuid(), UserEventType, false, loginName, null, eventPosition, null))
            };
            newState = "";
            return true;
        }

        public bool ProcessEvent(
            string partition, CheckpointTag eventPosition, string category1, ResolvedEvent data,
            out string newState, out EmittedEventEnvelope[] emittedEvents)
        {
            emittedEvents = null;
            newState = null;
            if (data.PositionStreamId.StartsWith("$"))
                return false;
            var lastSlashPos = data.PositionStreamId.LastIndexOf(_separator);
            if (lastSlashPos < 0)
                return true; // handled but not interesting to us

            var category = data.PositionStreamId.Substring(0, lastSlashPos);

            string linkTarget;
            if (data.EventType == SystemEventTypes.LinkTo) 
                linkTarget = data.Data;
            else 
                linkTarget = data.EventSequenceNumber + "@" + data.EventStreamId;

            emittedEvents = new[]
            {
                new EmittedEventEnvelope(
                    new EmittedLinkToWithRecategorization(
                        _categoryStreamPrefix + category, Guid.NewGuid(), linkTarget, eventPosition, expectedTag: null,
                        originalStreamId: data.PositionStreamId))
            };

            return true;
        }

 public override void Initialize()
 {
     base.Initialize();
     _lastOrderCheckpointTag = null;
     if (_orderStream != null) _orderStream.Dispose();
     _orderStream = null;
 }

 public void EnqueueTask(StagedTask workItem, CheckpointTag workItemCheckpointTag)
 {
     if (_queueState == QueueState.Stopped)
         throw new InvalidOperationException("Queue is Stopped");
     ValidateQueueingOrder(workItemCheckpointTag);
     _queuePendingEvents.Enqueue(workItem);
 }

 public ProjectionCheckpoint(
     RequestResponseDispatcher
         <ClientMessage.ReadStreamEventsBackward, ClientMessage.ReadStreamEventsBackwardCompleted> readDispatcher,
     RequestResponseDispatcher<ClientMessage.WriteEvents, ClientMessage.WriteEventsCompleted> writeDispatcher,
     ProjectionVersion projectionVersion, IPrincipal runAs, IProjectionCheckpointManager readyHandler,
     CheckpointTag from, PositionTagger positionTagger, CheckpointTag zero, int maxWriteBatchLength,
     ILogger logger = null)
 {
     if (readDispatcher == null) throw new ArgumentNullException("readDispatcher");
     if (writeDispatcher == null) throw new ArgumentNullException("writeDispatcher");
     if (readyHandler == null) throw new ArgumentNullException("readyHandler");
     if (positionTagger == null) throw new ArgumentNullException("positionTagger");
     if (zero == null) throw new ArgumentNullException("zero");
     if (from.CommitPosition <= from.PreparePosition) throw new ArgumentException("from");
     //NOTE: fromCommit can be equal fromPrepare on 0 position.  Is it possible anytime later? Ignoring for now.
     _readDispatcher = readDispatcher;
     _writeDispatcher = writeDispatcher;
     _projectionVersion = projectionVersion;
     _runAs = runAs;
     _readyHandler = readyHandler;
     _positionTagger = positionTagger;
     _zero = zero;
     _from = _last = from;
     _maxWriteBatchLength = maxWriteBatchLength;
     _logger = logger;
 }

 public SubscriptionStarted(
     Guid subscriptionId, CheckpointTag checkpointTag, long startingLastCommitPosition,
     long subscriptionMessageSequenceNumber, object source = null)
     : base(subscriptionId, checkpointTag, 0f, subscriptionMessageSequenceNumber, source)
 {
     _startingLastCommitPosition = startingLastCommitPosition;
 }

 public CheckpointLoaded(Guid projectionId, CheckpointTag checkpointTag, string checkpointData)
     : base(projectionId)
 {
     if (checkpointTag == null) throw new ArgumentNullException("checkpointTag");
     _checkpointTag = checkpointTag;
     _checkpointData = checkpointData;
 }

 public void UpdateToZero()
 {
     var zero = new EventPosition(0, -1);
     if (_lastEventPosition != zero || _lastTag != null)
         throw new InvalidOperationException("Posistion tagger has be already updated");
     _lastTag = _positionTagger.MakeZeroCheckpointTag();
 }

 public void UpdateByCheckpointTagForward(CheckpointTag newTag)
 {
     if (newTag <= _lastTag)
         throw new InvalidOperationException(
             string.Format("Event at checkpoint tag {0} has been already processed", newTag));
     _lastTag = newTag;
 }

 private void WriteResult(
     string partition, string resultBody, CheckpointTag causedBy, Guid causedByGuid, string correlationId)
 {
     var resultEvents = ResultUpdated(partition, resultBody, causedBy);
     if (resultEvents != null)
         _coreProjectionCheckpointManager.EventsEmitted(resultEvents, causedByGuid, correlationId);
 }

 public override bool IsMessageAfterCheckpointTag(
     CheckpointTag previous, ReaderSubscriptionMessage.CommittedEventDistributed committedEvent)
 {
     if (previous.Mode_ != CheckpointTag.Mode.Position)
         throw new ArgumentException("Mode.Position expected", "previous");
     return committedEvent.Data.Position > previous.Position;
 }

 public override void Start(CheckpointTag checkpointTag)
 {
     base.Start(checkpointTag);
     _lastOrderCheckpointTag = checkpointTag;
     _orderStream = CreateOrderStream();
     _orderStream.Start();
 }

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

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

 public void when()
 {
     _cache = new PartitionStateCache(CheckpointTag.FromPosition(0, -1));
     _cachedAtCheckpointTag = CheckpointTag.FromPosition(1000, 900);
     _cache.CachePartitionState(
         "partition", new PartitionState("data", null, _cachedAtCheckpointTag));
 }

        public bool ProcessEvent(
            string partition, CheckpointTag eventPosition, string category1, ResolvedEvent data,
            out string newState, out string newSharedState, out EmittedEventEnvelope[] emittedEvents)
        {
            newSharedState = null;
            emittedEvents = null;
            newState = null;
            if (data.EventStreamId != data.PositionStreamId)
                return false;
            var indexedEventType = data.EventType;
            if (indexedEventType == "$>")
                return false;

            string positionStreamId;
            var isStreamDeletedEvent = StreamDeletedHelper.IsStreamDeletedEvent(
                data.PositionStreamId, data.EventType, data.Data, out positionStreamId);
            if (isStreamDeletedEvent)
                indexedEventType = "$deleted";

            emittedEvents = new[]
            {
                new EmittedEventEnvelope(
                    new EmittedDataEvent(
                        _indexStreamPrefix + indexedEventType, Guid.NewGuid(), "$>", false,
                        data.EventSequenceNumber + "@" + positionStreamId,
                        isStreamDeletedEvent
                            ? new ExtraMetaData(new Dictionary<string, JRaw> {{"$deleted", new JRaw(-1)}})
                            : null, eventPosition, expectedTag: null))
            };

            return true;
        }