<commit_before>﻿namespace Cedar.EventStore.Subscriptions
{
    using System;
    using System.Threading;
    using System.Threading.Tasks;
    using Cedar.EventStore.Infrastructure;

    public abstract class SubscriptionBase : IDisposable
    {
        private int _pageSize = 50;
        private IDisposable _eventStoreAppendedSubscription;
        private readonly InterlockedBoolean _shouldFetch = new InterlockedBoolean();
        private readonly InterlockedBoolean _isFetching = new InterlockedBoolean();
        private readonly CancellationTokenSource _isDisposed = new CancellationTokenSource();

        protected SubscriptionBase(
            IReadOnlyEventStore readOnlyEventStore,
            IObservable<Unit> eventStoreAppendedNotification,
            StreamEventReceived streamEventReceived,
            SubscriptionDropped subscriptionDropped = null,
            string name = null)
        {
            ReadOnlyEventStore = readOnlyEventStore;
            EventStoreAppendedNotification = eventStoreAppendedNotification;
            StreamEventReceived = streamEventReceived;
            Name = string.IsNullOrWhiteSpace(name) ? Guid.NewGuid().ToString() : name;
            SubscriptionDropped = subscriptionDropped ?? ((_, __) => { });
        }

        public string Name { get; }

        public int PageSize
        {
            get { return _pageSize; }
            set { _pageSize = (value <= 0) ? 1 : value; }
        }

        protected IObservable<Unit> EventStoreAppendedNotification { get; }

        protected CancellationToken IsDisposed => _isDisposed.Token;

        protected IReadOnlyEventStore ReadOnlyEventStore { get; }

        protected StreamEventReceived StreamEventReceived { get; }

        protected SubscriptionDropped SubscriptionDropped { get; }

        public virtual Task Start(CancellationToken cancellationToken)
        {
            _eventStoreAppendedSubscription = EventStoreAppendedNotification.Subscribe(_ =>
            {
                _shouldFetch.Set(true);
                Fetch();
            });
            Fetch();
            return Task.FromResult(0);
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        ~SubscriptionBase()
        {
            Dispose(false);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (disposing)
            {
                _eventStoreAppendedSubscription?.Dispose();
                _isDisposed.Cancel();
            }
        }

        protected void Fetch()
        {
            if (_isFetching.CompareExchange(true, false))
            {
                return;
            }
            Task.Run(async () =>
            {
                bool isEnd = false;
                while (!isEnd || _shouldFetch.CompareExchange(false, true))
                {
                    isEnd = await DoFetch();
                }
                _isFetching.Set(false);
            }, IsDisposed);
        }

        protected abstract Task<bool> DoFetch();
    }
}<commit_msg>Workaround for test failure via Bart. Need to properly deal with dropped subscriptions.<commit_after>﻿namespace Cedar.EventStore.Subscriptions
{
    using System;
    using System.Threading;
    using System.Threading.Tasks;
    using Cedar.EventStore.Infrastructure;

    public abstract class SubscriptionBase : IDisposable
    {
        private int _pageSize = 50;
        private IDisposable _eventStoreAppendedSubscription;
        private readonly InterlockedBoolean _shouldFetch = new InterlockedBoolean();
        private readonly InterlockedBoolean _isFetching = new InterlockedBoolean();
        private readonly CancellationTokenSource _isDisposed = new CancellationTokenSource();

        protected SubscriptionBase(
            IReadOnlyEventStore readOnlyEventStore,
            IObservable<Unit> eventStoreAppendedNotification,
            StreamEventReceived streamEventReceived,
            SubscriptionDropped subscriptionDropped = null,
            string name = null)
        {
            ReadOnlyEventStore = readOnlyEventStore;
            EventStoreAppendedNotification = eventStoreAppendedNotification;
            StreamEventReceived = streamEventReceived;
            Name = string.IsNullOrWhiteSpace(name) ? Guid.NewGuid().ToString() : name;
            SubscriptionDropped = subscriptionDropped ?? ((_, __) => { });
        }

        public string Name { get; }

        public int PageSize
        {
            get { return _pageSize; }
            set { _pageSize = (value <= 0) ? 1 : value; }
        }

        protected IObservable<Unit> EventStoreAppendedNotification { get; }

        protected CancellationToken IsDisposed => _isDisposed.Token;

        protected IReadOnlyEventStore ReadOnlyEventStore { get; }

        protected StreamEventReceived StreamEventReceived { get; }

        protected SubscriptionDropped SubscriptionDropped { get; }

        public virtual Task Start(CancellationToken cancellationToken)
        {
            _eventStoreAppendedSubscription = EventStoreAppendedNotification.Subscribe(_ =>
            {
                _shouldFetch.Set(true);
                Fetch();
            });
            Fetch();
            return Task.FromResult(0);
        }

        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }

        ~SubscriptionBase()
        {
            Dispose(false);
        }

        protected virtual void Dispose(bool disposing)
        {
            if (disposing)
            {
                _eventStoreAppendedSubscription?.Dispose();
                _isDisposed.Cancel();
            }
        }
        private void Fetch()
        {
            if (_isFetching.CompareExchange(true, false))
            {
                return;
            }
            Task.Run(async () =>
            {
                try
                {
                    bool isEnd = false;
                    while(_shouldFetch.CompareExchange(false, true) || !isEnd)
                    {
                        isEnd = await DoFetch();
                    }
                }
                catch(Exception ex)
                {
                    // Drop subscription
                }
                finally
                {
                    _isFetching.Set(false);
                }
            }, IsDisposed);
        }

        protected abstract Task<bool> DoFetch();
    }
}
{'fraction_non_alphanumeric': 0.07170216025739237, 'fraction_numerical': 0.001532097441397273, 'mean_word_length': 1.7405541561712847, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5624724', 'n_tokens_mistral': 1570, 'n_tokens_neox': 1524, 'n_words': 325}