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

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

            return null;
        }

 public StreamConsumerData(GuidId subscriptionId, StreamId streamId, IStreamConsumerExtension streamConsumer, IStreamFilterPredicateWrapper filter)
 {
     SubscriptionId = subscriptionId;
     StreamId = streamId;
     StreamConsumer = streamConsumer;
     Filter = filter;
 }

 public Task RegisterConsumer(GuidId subscriptionId, StreamId streamId, string streamProvider, IStreamConsumerExtension streamConsumer, IStreamFilterPredicateWrapper filter)
 {
     if (!IsImplicitSubscriber(streamConsumer, streamId))
     {
         throw new ArgumentOutOfRangeException(streamId.ToString(), "Only implicit subscriptions are supported.");
     }
     return TaskDone.Done;
 }

        public bool RemoveConsumer(GuidId subscriptionId, Logger logger)
        {
            StreamConsumerData consumer;
            if (!queueData.TryGetValue(subscriptionId, out consumer)) return false;

            consumer.SafeDisposeCursor(logger);
            return queueData.Remove(subscriptionId);
        }

        public bool RemoveConsumer(GuidId subscriptionId)
        {
            StreamConsumerData consumer;
            if (!queueData.TryGetValue(subscriptionId, out consumer)) return false;

            if (consumer.Cursor != null)
            {
                // kill cursor activity and ensure it does not start again on this consumer data.
                consumer.Cursor.Dispose();
                consumer.Cursor = null; 
            }
            return queueData.Remove(subscriptionId);
        }

 internal bool IsImplicitSubscriber(GuidId subscriptionId, StreamId streamId)
 {
     return SubscriptionMarker.IsImplicitSubscription(subscriptionId.Guid);
 }

        public void RemoveSubscriber_Impl(GuidId subscriptionId, StreamId streamId)
        {
            if (IsShutdown) return;

            StreamConsumerCollection streamData;
            if (!pubSubCache.TryGetValue(streamId, out streamData)) return;

            // remove consumer
            bool removed = streamData.RemoveConsumer(subscriptionId, logger);
            if (removed && logger.IsVerbose) logger.Verbose((int)ErrorCode.PersistentStreamPullingAgent_10, "Removed Consumer: subscription={0}, for stream {1}.", subscriptionId, streamId);
            
            if (streamData.Count == 0)
                pubSubCache.Remove(streamId);
        }

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

 public bool TryGetConsumer(GuidId subscriptionId, out StreamConsumerData data)
 {
     return queueData.TryGetValue(subscriptionId, out data);
 }

 public Task OnSubscriptionFailure(GuidId subscriptionId, string streamProviderName, IStreamIdentity streamIdentity,
     StreamSequenceToken sequenceToken)
 {
     return TaskDone.Done;
 }

 Task IStreamPubSub.UnregisterConsumer(GuidId subscriptionId, StreamId streamId, string streamProvider)
 {
     return registry.UnregisterConsumer(subscriptionId, streamId, streamProvider);
 }

 internal void RemoveRemoteSubscriber(GuidId subscriptionId)
 {
     Tuple<IStreamConsumerExtension, IStreamFilterPredicateWrapper> ignore;
     consumers.TryRemove(subscriptionId, out ignore);
     if (consumers.Count == 0)
     {
         // Unsubscribe from PubSub?
     }
 }

 internal void AddRemoteSubscriber(GuidId subscriptionId, IStreamConsumerExtension streamConsumer, IStreamFilterPredicateWrapper filter)
 {
     consumers.TryAdd(subscriptionId, Tuple.Create(streamConsumer, filter));
 }

 public bool Contains(GuidId subscriptionId)
 {
     return queueData.ContainsKey(subscriptionId);
 }

 public Task<bool> FaultSubscription(StreamId streamId, GuidId subscriptionId)
 {
     return implicitPubSub.IsImplicitSubscriber(subscriptionId, streamId)
         ? implicitPubSub.FaultSubscription(streamId, subscriptionId)
         : explicitPubSub.FaultSubscription(streamId, subscriptionId);
 }

 public Task AddSubscriber(
     GuidId subscriptionId,
     StreamId streamId,
     IStreamConsumerExtension streamConsumer,
     IStreamFilterPredicateWrapper filter)
 {
     if (logger.IsVerbose) logger.Verbose((int)ErrorCode.PersistentStreamPullingAgent_09, "AddSubscriber: Stream={0} Subscriber={1}.", streamId, streamConsumer);
     // cannot await here because explicit consumers trigger this call, so it could cause a deadlock.
     AddSubscriber_Impl(subscriptionId, streamId, streamConsumer, null, filter)
         .LogException(logger, ErrorCode.PersistentStreamPullingAgent_26,
             String.Format("Failed to add subscription for stream {0}." , streamId))
         .Ignore();
     return TaskDone.Done;
 }

 public Task RegisterConsumer(GuidId subscriptionId, StreamId streamId, string streamProvider, IStreamConsumerExtension streamConsumer, IStreamFilterPredicateWrapper filter)
 {
     return implicitPubSub.IsImplicitSubscriber(streamConsumer, streamId)
         ? implicitPubSub.RegisterConsumer(subscriptionId, streamId, streamProvider, streamConsumer, filter)
         : explicitPubSub.RegisterConsumer(subscriptionId, streamId, streamProvider, streamConsumer, filter);
 }

 public Task RemoveSubscriber(GuidId subscriptionId, StreamId streamId)
 {
     RemoveSubscriber_Impl(subscriptionId, streamId);
     return TaskDone.Done;
 }

 public Task UnregisterConsumer(GuidId subscriptionId, StreamId streamId, string streamProvider)
 {
     return implicitPubSub.IsImplicitSubscriber(subscriptionId, streamId)
         ? implicitPubSub.UnregisterConsumer(subscriptionId, streamId, streamProvider)
         : explicitPubSub.UnregisterConsumer(subscriptionId, streamId, streamProvider);
 }