<commit_before>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using Aggregates.Contracts;
using Aggregates.Exceptions;
using Aggregates.Extensions;
using EventStore.ClientAPI;
using EventStore.ClientAPI.Common;
using EventStore.ClientAPI.Exceptions;
using EventStore.ClientAPI.Projections;
using Newtonsoft.Json;
using NServiceBus.Logging;
using NServiceBus.MessageInterfaces;

namespace Aggregates.Internal
{
    internal class EventStoreConsumer : IEventStoreConsumer, IDisposable
    {
        private static readonly ILog Logger = LogManager.GetLogger("EventStoreConsumer");
        
        private readonly IEventStoreConnection[] _clients;
        private readonly JsonSerializerSettings _settings;
        private readonly int _readSize;
        private readonly bool _extraStats;
        private readonly object _subLock;
        private readonly List<EventStoreCatchUpSubscription> _subscriptions;
        private readonly List<EventStorePersistentSubscriptionBase> _persistentSubs;
        private readonly ConcurrentDictionary<Guid, Tuple<EventStorePersistentSubscriptionBase, Guid>> _outstandingEvents;
        private bool _disposed;

        public EventStoreConsumer(IMessageMapper mapper, IEventStoreConnection[] clients, int readSize, bool extraStats)
        {
            _clients = clients;

            _readSize = readSize;
            _extraStats = extraStats;
            _subLock = new object();
            _subscriptions = new List<EventStoreCatchUpSubscription>();
            _persistentSubs = new List<EventStorePersistentSubscriptionBase>();
            _outstandingEvents = new ConcurrentDictionary<Guid, Tuple<EventStorePersistentSubscriptionBase, Guid>>();
            _settings = new JsonSerializerSettings
            {
                TypeNameHandling = TypeNameHandling.Auto,
                SerializationBinder = new EventSerializationBinder(mapper),
                ContractResolver = new EventContractResolver(mapper),
                Converters = new[] { new Newtonsoft.Json.Converters.StringEnumConverter() }
            };


            if (clients.Any(x => x.Settings.GossipSeeds == null || !x.Settings.GossipSeeds.Any()))
                throw new ArgumentException(
                    "Eventstore connection settings does not contain gossip seeds (even if single host call SetGossipSeedEndPoints and SetClusterGossipPort)");
        }
        public void Dispose()
        {
            if (_disposed) return;
            _disposed = true;
            foreach (var sub in _subscriptions)
                sub.Stop(TimeSpan.FromSeconds(5));
        }



        public Task<bool> SubscribeToStreamStart(string stream, CancellationToken token, Action<string, long, IFullEvent> callback, Func<Task> disconnected)
        {
            var clientsToken = CancellationTokenSource.CreateLinkedTokenSource(token);
            foreach (var client in _clients)
            {
                Logger.Write(LogLevel.Info,
                    () => $"Subscribing to beginning of stream [{stream}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");


                var settings = new CatchUpSubscriptionSettings(1000, 50, Logger.IsDebugEnabled, true);
                var startingNumber = 0L;
                try
                {
                    var subscription = client.SubscribeToStreamFrom(stream,
                        startingNumber,
                        settings,
                        eventAppeared: (sub, e) => EventAppeared(sub, e, clientsToken.Token, callback),
                        subscriptionDropped: (sub, reason, ex) => SubscriptionDropped(sub, reason, ex, disconnected));
                    Logger.Write(LogLevel.Info,
                        () => $"Subscribed to stream [{stream}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");
                    lock (_subLock) _subscriptions.Add(subscription);

                }
                catch (OperationTimedOutException)
                {
                    // If one fails, cancel all the others
                    clientsToken.Cancel();
                }
            }
            return Task.FromResult(!clientsToken.IsCancellationRequested);
        }

        public async Task<bool> SubscribeToStreamEnd(string stream, CancellationToken token, Action<string, long, IFullEvent> callback, Func<Task> disconnected)
        {
            var clientsToken = CancellationTokenSource.CreateLinkedTokenSource(token);
            foreach (var client in _clients)
            {
                try
                {
                    Logger.Write(LogLevel.Info,
                        () => $"Subscribing to end of stream [{stream}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");
                    // Subscribe to the end
                    var lastEvent =
                        await client.ReadStreamEventsBackwardAsync(stream, StreamPosition.End, 1, true).ConfigureAwait(false);

                    var settings = new CatchUpSubscriptionSettings(1000, 50, Logger.IsDebugEnabled, true);

                    var startingNumber = 0L;
                    if (lastEvent.Status == SliceReadStatus.Success)
                        startingNumber = lastEvent.Events[0].OriginalEventNumber;

                    var subscription = client.SubscribeToStreamFrom(stream,
                        startingNumber,
                        settings,
                        eventAppeared: (sub, e) => EventAppeared(sub, e, clientsToken.Token, callback),
                        subscriptionDropped: (sub, reason, ex) => SubscriptionDropped(sub, reason, ex, disconnected));
                    Logger.Write(LogLevel.Info,
                        () => $"Subscribed to stream [{stream}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");
                    lock (_subLock) _subscriptions.Add(subscription);
                }
                catch (OperationTimedOutException)
                {
                    // If one fails, cancel all the others
                    clientsToken.Cancel();
                }
            }
            return !clientsToken.IsCancellationRequested;
        }

        public async Task<bool> ConnectPinnedPersistentSubscription(string stream, string group, CancellationToken token,
            Action<string, long, IFullEvent> callback, Func<Task> disconnected)
        {
            var clientsToken = CancellationTokenSource.CreateLinkedTokenSource(token);
            foreach (var client in _clients)
            {
                Logger.Write(LogLevel.Info,
                    () => $"Connecting to persistent subscription stream [{stream}] group [{group}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");


                var settings = PersistentSubscriptionSettings.Create()
                    .StartFromBeginning()
                    .WithMaxRetriesOf(10)
                    .WithReadBatchOf(_readSize)
                    .WithLiveBufferSizeOf(_readSize * 3)
                    .DontTimeoutMessages()
                    .CheckPointAfter(TimeSpan.FromSeconds(30))
                    .MaximumCheckPointCountOf(_readSize * 3)
                    .ResolveLinkTos()
                    .WithNamedConsumerStrategy(SystemConsumerStrategies.Pinned);
                if (_extraStats)
                    settings.WithExtraStatistics();

                try
                {
                    await client.CreatePersistentSubscriptionAsync(stream, group, settings,
                        client.Settings.DefaultUserCredentials).ConfigureAwait(false);
                    Logger.Info($"Created PINNED persistent subscription stream [{stream}] group [{group}]");
                }
                catch (InvalidOperationException)
                {
                    // Already created
                }

                try
                {
                    var subscription = await client.ConnectToPersistentSubscriptionAsync(stream, group,
                        eventAppeared: (sub, e) => EventAppeared(sub, e, clientsToken.Token, callback),
                        subscriptionDropped: (sub, reason, ex) => SubscriptionDropped(sub, reason, ex, disconnected),
                        // Let us accept large number of unacknowledged events
                        bufferSize: _readSize * 3,
                        autoAck: false).ConfigureAwait(false);

                    lock (_subLock) _persistentSubs.Add(subscription);
                    Logger.Write(LogLevel.Info,
                        () => $"Connected to persistent subscription stream [{stream}] group [{group}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");
                }
                catch (OperationTimedOutException)
                {
                    return false;
                }
            }
            return true;
        }
        public async Task<bool> ConnectRoundRobinPersistentSubscription(string stream, string group, CancellationToken token,
            Action<string, long, IFullEvent> callback, Func<Task> disconnected)
        {
            var clientsToken = CancellationTokenSource.CreateLinkedTokenSource(token);
            foreach (var client in _clients)
            {
                Logger.Write(LogLevel.Info,
                    () => $"Connecting to persistent subscription stream [{stream}] group [{group}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");


                var settings = PersistentSubscriptionSettings.Create()
                    .StartFromBeginning()
                    .WithMaxRetriesOf(10)
                    .WithReadBatchOf(_readSize)
                    .WithLiveBufferSizeOf(_readSize * 3)
                    .DontTimeoutMessages()
                    .CheckPointAfter(TimeSpan.FromSeconds(30))
                    .MaximumCheckPointCountOf(_readSize * 3)
                    .ResolveLinkTos()
                    .WithNamedConsumerStrategy(SystemConsumerStrategies.RoundRobin);
                if (_extraStats)
                    settings.WithExtraStatistics();

                try
                {
                    await client.CreatePersistentSubscriptionAsync(stream, group, settings,
                        client.Settings.DefaultUserCredentials).ConfigureAwait(false);
                    Logger.Info($"Created ROUND ROBIN persistent subscription stream [{stream}] group [{group}]");
                }
                catch (InvalidOperationException)
                {
                    // Already created
                }


                try
                {
                    var subscription = await client.ConnectToPersistentSubscriptionAsync(stream, group,
                        eventAppeared: (sub, e) => EventAppeared(sub, e, clientsToken.Token, callback),
                        subscriptionDropped: (sub, reason, ex) => SubscriptionDropped(sub, reason, ex, disconnected),
                        // Let us accept large number of unacknowledged events
                        bufferSize: _readSize * 3,
                        autoAck: false).ConfigureAwait(false);

                    lock (_subLock) _persistentSubs.Add(subscription);
                    Logger.Write(LogLevel.Info,
                        () => $"Connected to persistent subscription stream [{stream}] group [{group}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");
                }
                catch (OperationTimedOutException)
                {
                    return false;
                }
            }
            return true;<commit_msg>Set subscription buffer size to always be bigger than readsize<commit_after>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Text;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using Aggregates.Contracts;
using Aggregates.Exceptions;
using Aggregates.Extensions;
using EventStore.ClientAPI;
using EventStore.ClientAPI.Common;
using EventStore.ClientAPI.Exceptions;
using EventStore.ClientAPI.Projections;
using Newtonsoft.Json;
using NServiceBus.Logging;
using NServiceBus.MessageInterfaces;

namespace Aggregates.Internal
{
    internal class EventStoreConsumer : IEventStoreConsumer, IDisposable
    {
        private static readonly ILog Logger = LogManager.GetLogger("EventStoreConsumer");
        
        private readonly IEventStoreConnection[] _clients;
        private readonly JsonSerializerSettings _settings;
        private readonly int _readSize;
        private readonly bool _extraStats;
        private readonly object _subLock;
        private readonly List<EventStoreCatchUpSubscription> _subscriptions;
        private readonly List<EventStorePersistentSubscriptionBase> _persistentSubs;
        private readonly ConcurrentDictionary<Guid, Tuple<EventStorePersistentSubscriptionBase, Guid>> _outstandingEvents;
        private bool _disposed;

        public EventStoreConsumer(IMessageMapper mapper, IEventStoreConnection[] clients, int readSize, bool extraStats)
        {
            _clients = clients;

            _readSize = readSize;
            _extraStats = extraStats;
            _subLock = new object();
            _subscriptions = new List<EventStoreCatchUpSubscription>();
            _persistentSubs = new List<EventStorePersistentSubscriptionBase>();
            _outstandingEvents = new ConcurrentDictionary<Guid, Tuple<EventStorePersistentSubscriptionBase, Guid>>();
            _settings = new JsonSerializerSettings
            {
                TypeNameHandling = TypeNameHandling.Auto,
                SerializationBinder = new EventSerializationBinder(mapper),
                ContractResolver = new EventContractResolver(mapper),
                Converters = new[] { new Newtonsoft.Json.Converters.StringEnumConverter() }
            };


            if (clients.Any(x => x.Settings.GossipSeeds == null || !x.Settings.GossipSeeds.Any()))
                throw new ArgumentException(
                    "Eventstore connection settings does not contain gossip seeds (even if single host call SetGossipSeedEndPoints and SetClusterGossipPort)");
        }
        public void Dispose()
        {
            if (_disposed) return;
            _disposed = true;
            foreach (var sub in _subscriptions)
                sub.Stop(TimeSpan.FromSeconds(5));
        }



        public Task<bool> SubscribeToStreamStart(string stream, CancellationToken token, Action<string, long, IFullEvent> callback, Func<Task> disconnected)
        {
            var clientsToken = CancellationTokenSource.CreateLinkedTokenSource(token);
            foreach (var client in _clients)
            {
                Logger.Write(LogLevel.Info,
                    () => $"Subscribing to beginning of stream [{stream}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");


                var settings = new CatchUpSubscriptionSettings(1000, 50, Logger.IsDebugEnabled, true);
                var startingNumber = 0L;
                try
                {
                    var subscription = client.SubscribeToStreamFrom(stream,
                        startingNumber,
                        settings,
                        eventAppeared: (sub, e) => EventAppeared(sub, e, clientsToken.Token, callback),
                        subscriptionDropped: (sub, reason, ex) => SubscriptionDropped(sub, reason, ex, disconnected));
                    Logger.Write(LogLevel.Info,
                        () => $"Subscribed to stream [{stream}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");
                    lock (_subLock) _subscriptions.Add(subscription);

                }
                catch (OperationTimedOutException)
                {
                    // If one fails, cancel all the others
                    clientsToken.Cancel();
                }
            }
            return Task.FromResult(!clientsToken.IsCancellationRequested);
        }

        public async Task<bool> SubscribeToStreamEnd(string stream, CancellationToken token, Action<string, long, IFullEvent> callback, Func<Task> disconnected)
        {
            var clientsToken = CancellationTokenSource.CreateLinkedTokenSource(token);
            foreach (var client in _clients)
            {
                try
                {
                    Logger.Write(LogLevel.Info,
                        () => $"Subscribing to end of stream [{stream}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");
                    // Subscribe to the end
                    var lastEvent =
                        await client.ReadStreamEventsBackwardAsync(stream, StreamPosition.End, 1, true).ConfigureAwait(false);

                    var settings = new CatchUpSubscriptionSettings(1000, 50, Logger.IsDebugEnabled, true);

                    var startingNumber = 0L;
                    if (lastEvent.Status == SliceReadStatus.Success)
                        startingNumber = lastEvent.Events[0].OriginalEventNumber;

                    var subscription = client.SubscribeToStreamFrom(stream,
                        startingNumber,
                        settings,
                        eventAppeared: (sub, e) => EventAppeared(sub, e, clientsToken.Token, callback),
                        subscriptionDropped: (sub, reason, ex) => SubscriptionDropped(sub, reason, ex, disconnected));
                    Logger.Write(LogLevel.Info,
                        () => $"Subscribed to stream [{stream}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");
                    lock (_subLock) _subscriptions.Add(subscription);
                }
                catch (OperationTimedOutException)
                {
                    // If one fails, cancel all the others
                    clientsToken.Cancel();
                }
            }
            return !clientsToken.IsCancellationRequested;
        }

        public async Task<bool> ConnectPinnedPersistentSubscription(string stream, string group, CancellationToken token,
            Action<string, long, IFullEvent> callback, Func<Task> disconnected)
        {
            var clientsToken = CancellationTokenSource.CreateLinkedTokenSource(token);
            foreach (var client in _clients)
            {
                Logger.Write(LogLevel.Info,
                    () => $"Connecting to persistent subscription stream [{stream}] group [{group}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");


                var settings = PersistentSubscriptionSettings.Create()
                    .StartFromBeginning()
                    .WithMaxRetriesOf(10)
                    .WithReadBatchOf(_readSize)
                    .WithBufferSizeOf(_readSize * 3)
                    .WithLiveBufferSizeOf(_readSize)
                    .DontTimeoutMessages()
                    .CheckPointAfter(TimeSpan.FromSeconds(30))
                    .MaximumCheckPointCountOf(_readSize * 3)
                    .ResolveLinkTos()
                    .WithNamedConsumerStrategy(SystemConsumerStrategies.Pinned);
                if (_extraStats)
                    settings.WithExtraStatistics();

                try
                {
                    await client.CreatePersistentSubscriptionAsync(stream, group, settings,
                        client.Settings.DefaultUserCredentials).ConfigureAwait(false);
                    Logger.Info($"Created PINNED persistent subscription stream [{stream}] group [{group}]");
                }
                catch (InvalidOperationException)
                {
                    // Already created
                }

                try
                {
                    var subscription = await client.ConnectToPersistentSubscriptionAsync(stream, group,
                        eventAppeared: (sub, e) => EventAppeared(sub, e, clientsToken.Token, callback),
                        subscriptionDropped: (sub, reason, ex) => SubscriptionDropped(sub, reason, ex, disconnected),
                        // Let us accept large number of unacknowledged events
                        bufferSize: _readSize,
                        autoAck: false).ConfigureAwait(false);

                    lock (_subLock) _persistentSubs.Add(subscription);
                    Logger.Write(LogLevel.Info,
                        () => $"Connected to persistent subscription stream [{stream}] group [{group}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");
                }
                catch (OperationTimedOutException)
                {
                    return false;
                }
            }
            return true;
        }
        public async Task<bool> ConnectRoundRobinPersistentSubscription(string stream, string group, CancellationToken token,
            Action<string, long, IFullEvent> callback, Func<Task> disconnected)
        {
            var clientsToken = CancellationTokenSource.CreateLinkedTokenSource(token);
            foreach (var client in _clients)
            {
                Logger.Write(LogLevel.Info,
                    () => $"Connecting to persistent subscription stream [{stream}] group [{group}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");


                var settings = PersistentSubscriptionSettings.Create()
                    .StartFromBeginning()
                    .WithMaxRetriesOf(10)
                    .WithReadBatchOf(_readSize)
                    .WithBufferSizeOf(_readSize * 3)
                    .WithLiveBufferSizeOf(_readSize)
                    .DontTimeoutMessages()
                    .CheckPointAfter(TimeSpan.FromSeconds(30))
                    .MaximumCheckPointCountOf(_readSize * 3)
                    .ResolveLinkTos()
                    .WithNamedConsumerStrategy(SystemConsumerStrategies.RoundRobin);
                if (_extraStats)
                    settings.WithExtraStatistics();

                try
                {
                    await client.CreatePersistentSubscriptionAsync(stream, group, settings,
                        client.Settings.DefaultUserCredentials).ConfigureAwait(false);
                    Logger.Info($"Created ROUND ROBIN persistent subscription stream [{stream}] group [{group}]");
                }
                catch (InvalidOperationException)
                {
                    // Already created
                }


                try
                {
                    var subscription = await client.ConnectToPersistentSubscriptionAsync(stream, group,
                        eventAppeared: (sub, e) => EventAppeared(sub, e, clientsToken.Token, callback),
                        subscriptionDropped: (sub, reason, ex) => SubscriptionDropped(sub, reason, ex, disconnected),
                        // Let us accept large number of unacknowledged events
                        bufferSize: _readSize,
                        autoAck: false).ConfigureAwait(false);

                    lock (_subLock) _persistentSubs.Add(subscription);
                    Logger.Write(LogLevel.Info,
                        () => $"Connected to persistent subscription stream [{stream}] group [{group}] on client {client.Settings.GossipSeeds[0].EndPoint.Address}");
                }
                catch (OperationTimedOutException)
                {
                    return false;
                }
            }
            return true;
{'fraction_non_alphanumeric': 0.07898627243928194, 'fraction_numerical': 0.0032101372756071804, 'mean_word_length': 1.8484119345524543, 'pattern_counts': {'":': 0, '<': 43, '<?xml version=': 0, '>': 77, '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': '5450838', 'n_tokens_mistral': 5647, 'n_tokens_neox': 5341, 'n_words': 1220}