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

    public class SqlWatcher : IDisposable
    {
        private readonly SqlConnection _connection;
        private readonly string _connectionString;
        private readonly IEventStore _eventStore;
        private readonly InterlockedBoolean _fetchingEvents = new InterlockedBoolean();
        private SqlCommand _command;
        private string _checkpoint;
        private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();

        public SqlWatcher(string connectionString, IEventStore eventStore)
        {
            _connectionString = connectionString;
            _eventStore = eventStore;
            _connection = new SqlConnection(_connectionString);
            SqlDependency.Start(connectionString);
        }

        public void Dispose()
        {
            _cancellationTokenSource.Cancel();
            _command.Dispose();
            SqlConnection.ClearPool(_connection);
            _connection.Dispose();
            SqlDependency.Stop(_connectionString);
        }

        public async Task Start()
        {
            var allEventsPage = await _eventStore.ReadAll(_eventStore.EndCheckpoint, 1, ReadDirection.Backward);
            _checkpoint = allEventsPage.NextCheckpoint;

            await _connection.OpenAsync();

            Subscribe();

            FetchEvents();
        }

        private void Subscribe()
        {
            _command = new SqlCommand("SELECT Id FROM dbo.Events", _connection)
            {
                Notification = null
            };
            var sqlDependency = new SqlDependency(_command);
            sqlDependency.OnChange += SqlDependencyOnOnChange;

            _command.ExecuteNonQuery();
        }

        private void SqlDependencyOnOnChange(object sender, SqlNotificationEventArgs sqlNotificationEventArgs)
        {
            var dependency = (SqlDependency) sender;
            dependency.OnChange -= SqlDependencyOnOnChange;

            Subscribe();

            FetchEvents();
        }


        /// <summary>
        /// Fetches any newly saved events and reises them to subscribers accordingly.
        /// This is method is designed to run a *single* background process at any given time.
        /// </summary>
        private void FetchEvents()
        {
            if(_fetchingEvents.EnsureCalledOnce())
            {
                return;
            }
            Task.Run(async () =>
            {
                bool isEnd = false;
                while(!isEnd)
                {
                    if(_cancellationTokenSource.IsCancellationRequested)
                    {
                        return;
                    }

                    var allEventsPage = await _eventStore
                        .ReadAll(_checkpoint, 10, cancellationToken: _cancellationTokenSource.Token)
                        .NotOnCapturedContext();

                    isEnd = allEventsPage.IsEnd;

                    _checkpoint = allEventsPage.NextCheckpoint;

                    Console.WriteLine(allEventsPage.NextCheckpoint);
                }
<commit_msg>Typos<commit_after>﻿namespace Cedar.EventStore.Subscriptions
{
    using System;
    using System.Data.SqlClient;
    using System.Threading;
    using System.Threading.Tasks;
    using Cedar.EventStore.Infrastructure;
    using Cedar.EventStore.Streams;

    public class SqlWatcher : IDisposable
    {
        private readonly SqlConnection _connection;
        private readonly string _connectionString;
        private readonly IEventStore _eventStore;
        private readonly InterlockedBoolean _fetchingEvents = new InterlockedBoolean();
        private SqlCommand _command;
        private string _checkpoint;
        private readonly CancellationTokenSource _cancellationTokenSource = new CancellationTokenSource();

        public SqlWatcher(string connectionString, IEventStore eventStore)
        {
            _connectionString = connectionString;
            _eventStore = eventStore;
            _connection = new SqlConnection(_connectionString);
            SqlDependency.Start(connectionString);
        }

        public void Dispose()
        {
            _cancellationTokenSource.Cancel();
            _command.Dispose();
            SqlConnection.ClearPool(_connection);
            _connection.Dispose();
            SqlDependency.Stop(_connectionString);
        }

        public async Task Start()
        {
            var allEventsPage = await _eventStore.ReadAll(_eventStore.EndCheckpoint, 1, ReadDirection.Backward);
            _checkpoint = allEventsPage.NextCheckpoint;

            await _connection.OpenAsync();

            Subscribe();

            FetchEvents();
        }

        private void Subscribe()
        {
            _command = new SqlCommand("SELECT Id FROM dbo.Events", _connection)
            {
                Notification = null
            };
            var sqlDependency = new SqlDependency(_command);
            sqlDependency.OnChange += SqlDependencyOnOnChange;

            _command.ExecuteNonQuery();
        }

        private void SqlDependencyOnOnChange(object sender, SqlNotificationEventArgs sqlNotificationEventArgs)
        {
            var dependency = (SqlDependency) sender;
            dependency.OnChange -= SqlDependencyOnOnChange;

            Subscribe();

            FetchEvents();
        }


        /// <summary>
        /// Fetches any newly saved events and raises them to subscribers accordingly.
        /// This is method is designed to run a *single* background process at any given time.
        /// </summary>
        private void FetchEvents()
        {
            if(_fetchingEvents.EnsureCalledOnce())
            {
                return;
            }
            Task.Run(async () =>
            {
                bool isEnd = false;
                while(!isEnd)
                {
                    if(_cancellationTokenSource.IsCancellationRequested)
                    {
                        return;
                    }

                    var allEventsPage = await _eventStore
                        .ReadAll(_checkpoint, 10, cancellationToken: _cancellationTokenSource.Token)
                        .NotOnCapturedContext();

                    isEnd = allEventsPage.IsEnd;

                    _checkpoint = allEventsPage.NextCheckpoint;

                    //TODO Push events from allEventsPage to subscribers.
                }

{'fraction_non_alphanumeric': 0.06685569318012237, 'fraction_numerical': 0.0008953887479480674, 'mean_word_length': 1.7832225913621262, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, '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': '4227642', 'n_tokens_mistral': 1554, 'n_tokens_neox': 1488, 'n_words': 350}