public async Task<StreamSequenceToken> DeliverBatch(GuidId subscriptionId, Immutable<IBatchContainer> batch, StreamSequenceToken prevToken)
        {
            foreach (var each in batch.Value.GetEvents<object>())
                await handler(each.Item1);

            return null;
        }

        public override int CompareTo(StreamSequenceToken other)
        {
            if (other == null)
                return 1;

            var token = other as TimeSequenceToken;
            if (token == null)
                throw new ArgumentOutOfRangeException(nameof(other));

            var difference = Timestamp.CompareTo(token.Timestamp);
            return difference != 0 ? difference : EventIndex.CompareTo(token.EventIndex);
        }

 public override int CompareTo(StreamSequenceToken other)
 {
     if (other == null)
         return 1;
     
     var token = other as EventSequenceToken;
     if (token == null)
         throw new ArgumentOutOfRangeException("other");
     
     int difference = sequenceNumber.CompareTo(token.sequenceNumber);
     return difference != 0 ? difference : eventIndex.CompareTo(token.eventIndex);
 }

 public IQueueCacheCursor GetCacheCursor(Guid streamGuid, string streamNamespace, StreamSequenceToken token)
 {
     // NOTE: We assume the client ALWAYS wants to replay the whole dictionary, if it doesn't then it shouldn't be using this stream type in the first place.
     if (token != null && !(token is DictStreamToken))
     {
         // Null token can come from a stream subscriber that is just interested to start consuming from latest (the most recent event added to the cache).
         throw new ArgumentOutOfRangeException("token", "token must be of type DictStreamToken");
     }
     var dict = GetDict(streamNamespace, streamGuid);
     var dictCursor = dict.GetCursor();
     return new DictQueueCacheCursor(dict, streamNamespace, streamGuid);
 }

 public StreamPosition(IStreamIdentity streamIdentity, StreamSequenceToken sequenceToken)
 {
     if (streamIdentity == null)
     {
         throw new ArgumentNullException("streamIdentity");
     }
     if (sequenceToken == null)
     {
         throw new ArgumentNullException("sequenceToken");
     }
     StreamIdentity = streamIdentity;
     SequenceToken = sequenceToken;
 }

        public Task OnNextAsync(int item, StreamSequenceToken token = null)
        {
            logger.Info("OnNextAsync(item={0}, token={1})", item, token != null ? token.ToString() : "null");
            if (failPeriodTimer == null)
            {
                eventsConsumedCount++;
            }
            else if(failPeriodTimer.Elapsed >= failPeriod)
            {
                failPeriodTimer = null;
                eventsConsumedCount++;
            }
            else
            {
                eventsFailedCount++;
                throw new AggregateException("GO WAY!");
            }

            return TaskDone.Done;
        }

        private void Add(IBatchContainer batch, StreamSequenceToken sequenceToken)
        {
            if (batch == null) throw new ArgumentNullException(nameof(batch));

            var cacheBucket = GetOrCreateBucket();

            cacheBucket.UpdateNumItems(1);
            // Add message to linked list
            var item = new TimedQueueCacheItem
            {
                Batch = batch,
                SequenceToken = sequenceToken,
                CacheBucket = cacheBucket,
            };

            item.Timestamp = GetTimestampForItem(batch);

            var newNode = new LinkedListNode<TimedQueueCacheItem>(item);

            // If it's the first item, then we also update 
            if (cacheBucket.NumCurrentItems == 1)
            {
                Log(_logger, "TimedQueueCache for QueueId:{0}, Add: The oldest timespan in the cache is {1}", Id.ToString(), item.Timestamp);
                cacheBucket.OldestMemberTimestamp = item.Timestamp;
                cacheBucket.OldestMember = newNode;
            }

            // Setting the newest member
            cacheBucket.NewestMemberTimestamp = item.Timestamp;
            cacheBucket.NewestMember = newNode;

            _cachedMessages.AddFirst(newNode);

            _counterMessagesInCache.Increment(Id.ToString(), 1);
        }

        private LinkedListNode<TimedQueueCacheItem> FindNodeBySequenceToken(StreamSequenceToken sequenceToken)
        {
            // First we find a bucket where the node is in
            var sequenceBucket =
                _cacheCursorHistogram.First(
                    bucket =>
                        !sequenceToken.Newer(bucket.NewestMember.Value.SequenceToken) &&
                        !sequenceToken.Older(bucket.OldestMember.Value.SequenceToken));

            // Now that we have the bucket, we iterate on the members there starting from the newest in the bucket
            LinkedListNode<TimedQueueCacheItem> node = sequenceBucket.NewestMember;
            while (node != null && node.Value.SequenceToken.Newer(sequenceToken))
            {
                // did we get to the end?
                // node is the last message in the cache
                if (node.Next == null)
                    break;

                // if sequenceId is between the two, take the lower
                if (node.Next.Value.SequenceToken.Older(sequenceToken))
                {
                    node = node.Next;
                    break;
                }

                node = node.Next;
            }

            return node;
        }

        //public virtual IQueueCacheCursor GetCacheCursor(Guid streamGuid, string streamNamespace, StreamSequenceToken token)
        public virtual IQueueCacheCursor GetCacheCursor(IStreamIdentity streamIdentity, StreamSequenceToken token)
        {
            if (token != null && !(token is EventSequenceToken))
            {
                // Null token can come from a stream subscriber that is just interested to
                // start consuming from latest (the most recent event added to the cache).
                throw new ArgumentOutOfRangeException(nameof(token), "token must be of type EventSequenceToken");
            }

            var cursor = new TimedQueueCacheCursor(this, streamIdentity.Guid, streamIdentity.Namespace, _logger);
            InitializeCursor(cursor, token);
            return cursor;
        }

        internal void ResetCursor(SimpleQueueCacheCursor cursor, StreamSequenceToken token)
        {
            Log(logger, "ResetCursor: {0} to token {1}", cursor, token);

            if (cursor.IsSet)
            {
                cursor.Element.Value.CacheBucket.UpdateNumCursors(-1);
            }
            cursor.Reset(token);
        }

 private Task OnNextAsync(StreamImmutabilityTestObject myObject, StreamSequenceToken streamSequenceToken)
 {
     _myObject = myObject;
     return TaskDone.Done;
 }

 public StreamConsumerData AddConsumer(GuidId subscriptionId, StreamId streamId, IStreamConsumerExtension streamConsumer, StreamSequenceToken token, IStreamFilterPredicateWrapper filter)
 {
     var consumerData = new StreamConsumerData(subscriptionId, streamId, streamConsumer, filter);
     queueData.Add(subscriptionId, consumerData);
     return consumerData;
 }

 private Task OnNext(int e, StreamSequenceToken token, int countCapture, Counter count)
 {
     logger.Info("Got next event {0} on handle {1}", e, countCapture);
     var contextValue = RequestContext.Get(SampleStreaming_ProducerGrain.RequestContextKey) as string;
     if (!String.Equals(contextValue, SampleStreaming_ProducerGrain.RequestContextValue))
     {
         throw new Exception(String.Format("Got the wrong RequestContext value {0}.", contextValue));
     }
     count.Increment();
     return TaskDone.Done;
 }

 public override bool Equals(StreamSequenceToken other)
 {
     var token = other as EventSequenceToken;
     return token != null && (token.sequenceNumber == sequenceNumber &&
                              token.eventIndex == eventIndex);
 }

 /// <summary>
 /// Sets sequence token by serializing it to property.
 /// </summary>
 /// <param name="token"></param>
 public virtual void SetSequenceToken(StreamSequenceToken token)
 {
     SequenceToken = token != null ? GetTokenBytes(token) : null;
 }

 private static byte[] GetTokenBytes(StreamSequenceToken token)
 {
     var bodyStream = new BinaryTokenStreamWriter();
     SerializationManager.Serialize(token, bodyStream);
     return bodyStream.ToByteArray();
 }

 internal void Reset(StreamSequenceToken token)
 {
     Element = null;
     SequenceToken = token;
 }

        public virtual IQueueCacheCursor GetCacheCursor(Guid streamGuid, string streamNamespace, StreamSequenceToken token)
        {
            if (token != null && !(token is EventSequenceToken))
            {
                // Null token can come from a stream subscriber that is just interested to start consuming from latest (the most recent event added to the cache).
                throw new ArgumentOutOfRangeException("token", "token must be of type EventSequenceToken");
            }

            var cursor = new SimpleQueueCacheCursor(this, streamGuid, streamNamespace, logger);
            InitializeCursor(cursor, token, true);
            return cursor;
        }

        internal void InitializeCursor(SimpleQueueCacheCursor cursor, StreamSequenceToken sequenceToken, bool enforceSequenceToken)
        {
            Log(logger, "InitializeCursor: {0} to sequenceToken {1}", cursor, sequenceToken);
           
            if (cachedMessages.Count == 0) // nothing in cache
            {
                StreamSequenceToken tokenToReset = sequenceToken ?? (lastSequenceTokenAddedToCache != null ? ((EventSequenceToken)lastSequenceTokenAddedToCache).NextSequenceNumber() : null);
                ResetCursor(cursor, tokenToReset);
                return;
            }

            // if offset is not set, iterate from newest (first) message in cache, but not including the irst message itself
            if (sequenceToken == null)
            {
                StreamSequenceToken tokenToReset = lastSequenceTokenAddedToCache != null ? ((EventSequenceToken)lastSequenceTokenAddedToCache).NextSequenceNumber() : null;
                ResetCursor(cursor, tokenToReset);
                return;
            }

            if (sequenceToken.Newer(cachedMessages.First.Value.SequenceToken)) // sequenceId is too new to be in cache
            {
                ResetCursor(cursor, sequenceToken);
                return;
            }

            LinkedListNode<SimpleQueueCacheItem> lastMessage = cachedMessages.Last;
            // Check to see if offset is too old to be in cache
            if (sequenceToken.Older(lastMessage.Value.SequenceToken))
            {
                if (enforceSequenceToken)
                {
                    // throw cache miss exception
                    throw new QueueCacheMissException(sequenceToken, cachedMessages.Last.Value.SequenceToken, cachedMessages.First.Value.SequenceToken);
                }
                sequenceToken = lastMessage.Value.SequenceToken;
            }

            // Now the requested sequenceToken is set and is also within the limits of the cache.

            // Find first message at or below offset
            // Events are ordered from newest to oldest, so iterate from start of list until we hit a node at a previous offset, or the end.
            LinkedListNode<SimpleQueueCacheItem> node = cachedMessages.First;
            while (node != null && node.Value.SequenceToken.Newer(sequenceToken))
            {
                // did we get to the end?
                if (node.Next == null) // node is the last message
                    break;
                
                // if sequenceId is between the two, take the higher
                if (node.Next.Value.SequenceToken.Older(sequenceToken))
                    break;
                
                node = node.Next;
            }

            // return cursor from start.
            SetCursor(cursor, node);
        }

        // Called by rendezvous when new remote subscriber subscribes to this stream.
        private async Task AddSubscriber_Impl(
            GuidId subscriptionId,
            StreamId streamId,
            IStreamConsumerExtension streamConsumer,
            StreamSequenceToken cacheToken,
            IStreamFilterPredicateWrapper filter)
        {
            if (IsShutdown) return;

            StreamConsumerCollection streamDataCollection;
            if (!pubSubCache.TryGetValue(streamId, out streamDataCollection))
            {
                streamDataCollection = new StreamConsumerCollection(DateTime.UtcNow);
                pubSubCache.Add(streamId, streamDataCollection);
            }

            StreamConsumerData data;
            if (!streamDataCollection.TryGetConsumer(subscriptionId, out data))
                data = streamDataCollection.AddConsumer(subscriptionId, streamId, streamConsumer, filter ?? DefaultStreamFilter);

            if (await DoHandshakeWithConsumer(data, cacheToken))
            {
                if (data.State == StreamConsumerDataState.Inactive)
                    RunConsumerCursor(data, data.Filter).Ignore(); // Start delivering events if not actively doing so
            }
        }