<commit_before>﻿// ==========================================================================
//  PollingSubscription.cs
//  Squidex Headless CMS
// ==========================================================================
//  Copyright (c) Squidex Group
//  All rights reserved.
// ==========================================================================

using System;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using Squidex.Infrastructure.Actors;
using Squidex.Infrastructure.Tasks;

namespace Squidex.Infrastructure.CQRS.Events
{
    public sealed class PollingSubscription : Actor, IEventSubscription
    {
        private readonly IEventNotifier notifier;
        private readonly MongoEventStore store;
        private readonly CancellationTokenSource disposeToken = new CancellationTokenSource();
        private readonly Regex streamRegex;
        private readonly string streamFilter;
        private readonly IEventSubscriber subscriber;
        private string position;
        private bool isPolling;
        private IDisposable notification;

        private sealed class Connect
        {
        }

        private sealed class StartPoll
        {
        }

        private sealed class StopPoll
        {
        }

        public PollingSubscription(MongoEventStore store, IEventNotifier notifier, IEventSubscriber subscriber, string streamFilter, string position)
        {
            this.notifier = notifier;
            this.store = store;
            this.streamFilter = streamFilter;
            this.subscriber = subscriber;

            streamRegex = new Regex(streamFilter);

            DispatchAsync(new Connect()).Forget();
        }

        public Task StopAsync()
        {
            return StopAndWaitAsync();
        }

        protected override Task OnStop()
        {
            disposeToken?.Cancel();

            notification?.Dispose();

            return TaskHelper.Done;
        }

        protected override async Task OnError(Exception exception)
        {
            await subscriber.OnErrorAsync(this, exception);

            await StopAsync();
        }

        protected override async Task OnMessage(object message)
        {
            switch (message)
            {
                case Connect connect:
                {
                    notification = notifier.Subscribe(streamName =>
                    {
                        if (streamRegex.IsMatch(streamName))
                        {
                            DispatchAsync(new StartPoll()).Forget();
                        }
                    });

                    break;
                }<commit_msg>Fixed polling subscription.<commit_after>﻿// ==========================================================================
//  PollingSubscription.cs
//  Squidex Headless CMS
// ==========================================================================
//  Copyright (c) Squidex Group
//  All rights reserved.
// ==========================================================================

using System;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using Squidex.Infrastructure.Actors;
using Squidex.Infrastructure.Tasks;

namespace Squidex.Infrastructure.CQRS.Events
{
    public sealed class PollingSubscription : Actor, IEventSubscription
    {
        private readonly IEventNotifier notifier;
        private readonly MongoEventStore store;
        private readonly CancellationTokenSource disposeToken = new CancellationTokenSource();
        private readonly Regex streamRegex;
        private readonly string streamFilter;
        private readonly IEventSubscriber subscriber;
        private string position;
        private bool isPolling;
        private IDisposable notification;

        private sealed class Connect
        {
        }

        private sealed class StartPoll
        {
        }

        private sealed class StopPoll
        {
        }

        public PollingSubscription(MongoEventStore store, IEventNotifier notifier, IEventSubscriber subscriber, string streamFilter, string position)
        {
            this.notifier = notifier;
            this.store = store;
            this.streamFilter = streamFilter;
            this.subscriber = subscriber;

            streamRegex = new Regex(streamFilter);

            DispatchAsync(new Connect()).Forget();
        }

        public Task StopAsync()
        {
            return StopAndWaitAsync();
        }

        protected override Task OnStop()
        {
            disposeToken?.Cancel();

            notification?.Dispose();

            return TaskHelper.Done;
        }

        protected override async Task OnError(Exception exception)
        {
            await subscriber.OnErrorAsync(this, exception);

            await StopAsync();
        }

        protected override async Task OnMessage(object message)
        {
            switch (message)
            {
                case Connect connect:
                {
                    notification = notifier.Subscribe(streamName =>
                    {
                        if (streamRegex.IsMatch(streamName))
                        {
                            DispatchAsync(new StartPoll()).Forget();
                        }
                    });

                    DispatchAsync(new StartPoll()).Forget();

                    break;
                }
{'fraction_non_alphanumeric': 0.1427262733602052, 'fraction_numerical': 0.0, 'mean_word_length': 1.9668478260869566, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '3873467', 'n_tokens_mistral': 1236, 'n_tokens_neox': 1133, 'n_words': 294}