<commit_before>﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;

namespace Cap.Consistency.Job
{
    public class CronJobProcessor : IProcessor
    {
        private ILogger _logger;
        private IServiceProvider _provider;

        public CronJobProcessor(
            ILogger<CronJobProcessor> logger,
            IServiceProvider provider) {
            _logger = logger;
            _provider = provider;
        }

        public override string ToString() => nameof(CronJobProcessor);

        public Task ProcessAsync(ProcessingContext context) {
            if (context == null) throw new ArgumentNullException(nameof(context));
            return ProcessCoreAsync(context);
        }

        private async Task ProcessCoreAsync(ProcessingContext context) {
            //var storage = context.Storage;
            //var jobs = await GetJobsAsync(storage);
            var jobs = GetJobsAsync();
            if (!jobs.Any()) {
                _logger.CronJobsNotFound();

                // This will cancel this processor.
                throw new OperationCanceledException();
            }
            _logger.CronJobsScheduling(jobs);

            context.ThrowIfStopping();

            var computedJobs = Compute(jobs, context.CronJobRegistry.Build());
            if (context.IsStopping) {
                return;
            }

            await Task.WhenAll(computedJobs.Select(j => RunAsync(j, context)));
        }

        private async Task RunAsync(ComputedCronJob computedJob, ProcessingContext context) {
            //var storage = context.Storage;
            var retryBehavior = computedJob.RetryBehavior;

            while (!context.IsStopping) {
                var now = DateTime.UtcNow;

                var due = ComputeDue(computedJob, now);
                var timeSpan = due - now;

                if (timeSpan.TotalSeconds > 0) {
                    await context.WaitAsync(timeSpan);
                }

                context.ThrowIfStopping();

                using (var scopedContext = context.CreateScope()) {
                    var provider = scopedContext.Provider;

                    var job = provider.GetService<IJob>();
                    var success = true;

                    try {
                        var sw = Stopwatch.StartNew();
                        await job.ExecuteAsync();
                        sw.Stop();
                        computedJob.Retries = 0;
                        _logger.CronJobExecuted(computedJob.Job.Name, sw.Elapsed.TotalSeconds);
                    }
                    catch (Exception ex) {
                        success = false;
                        if (computedJob.Retries == 0) {
                            computedJob.FirstTry = DateTime.UtcNow;
                        }
                        computedJob.Retries++;
                        _logger.CronJobFailed(computedJob.Job.Name, ex);
                    }

                    if (success) {
                        //var connection = provider.GetRequiredService<IStorageConnection>();
                        //await connection.AttachCronJobAsync(computedJob.Job);

                        //computedJob.Update(DateTime.UtcNow);

                        //await connection.UpdateCronJobAsync(computedJob.Job);
                    }
                }
            }
        }

        private DateTime ComputeDue(ComputedCronJob computedJob, DateTime now) {
            computedJob.UpdateNext(now);

            var retryBehavior = computedJob.RetryBehavior ?? RetryBehavior.DefaultRetry;
            var retries = computedJob.Retries;

            if (retries == 0) {
                return computedJob.Next;
            }

            var realNext = computedJob.Schedule.GetNextOccurrence(now);

            if (!retryBehavior.Retry) {
                // No retry. If job failed before, we don't care, just schedule it next as usual.
                return realNext;
            }

            if (retries >= retryBehavior.RetryCount) {
                // Max retries. Just schedule it for the next occurance.
                return realNext;
            }

            // Delay a bit.
            return computedJob.FirstTry.AddSeconds(retryBehavior.RetryIn(retries));
        }

        //private async Task<CronJob[]> GetJobsAsync(IStorage storage) {
        //    using (var scope = _provider.CreateScope()) {
        //        var provider = scope.ServiceProvider;
        //        var connection = provider.GetRequiredService<IStorageConnection>();

        //        return await connection.GetCronJobsAsync();
        //    }
        //}

        private CronJob[] GetJobsAsync() {
            return new CronJob[] {
                new CronJob {
                     Id= Guid.NewGuid().ToString(),
                     Cron= "* * * * *",
                     Name="Cap.Messages",
                     TypeName = null
                }
            };
        }

        private ComputedCronJob[] Compute(IEnumerable<CronJob> jobs, CronJobRegistry.Entry[] entries)
            => jobs.Select(j => CreateComputedCronJob(j, entries)).ToArray();

        private ComputedCronJob CreateComputedCronJob(CronJob job, CronJobRegistry.Entry[] entries) {
            var entry = entries.First(e => e.Name == job.Name);
            return new ComputedCronJob(job, entry);
        }
    }
}<commit_msg>modify GetJobs from DI.<commit_after>﻿using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;

namespace Cap.Consistency.Job
{
    public class CronJobProcessor : IJobProcessor
    {
        private ILogger _logger;
        private IServiceProvider _provider;
        private DefaultCronJobRegistry _jobRegistry;

        public CronJobProcessor(
            DefaultCronJobRegistry jobRegistry,
            ILogger<CronJobProcessor> logger,
            IServiceProvider provider) {

            _jobRegistry = jobRegistry;
            _logger = logger;
            _provider = provider;
        }

        public override string ToString() => nameof(CronJobProcessor);

        public Task ProcessAsync(ProcessingContext context) {
            if (context == null) throw new ArgumentNullException(nameof(context));
            return ProcessCoreAsync(context);
        }

        private async Task ProcessCoreAsync(ProcessingContext context) {
            //var storage = context.Storage;
            //var jobs = await GetJobsAsync(storage);

            var jobs = GetJobs();
            if (!jobs.Any()) {
                _logger.CronJobsNotFound();

                // This will cancel this processor.
                throw new OperationCanceledException();
            }
            _logger.CronJobsScheduling(jobs);

            context.ThrowIfStopping();

            var computedJobs = Compute(jobs, context.CronJobRegistry.Build());
            if (context.IsStopping) {
                return;
            }

            await Task.WhenAll(computedJobs.Select(j => RunAsync(j, context)));
        }

        private async Task RunAsync(ComputedCronJob computedJob, ProcessingContext context) {
            //var storage = context.Storage;
            var retryBehavior = computedJob.RetryBehavior;

            while (!context.IsStopping) {
                var now = DateTime.UtcNow;

                var due = ComputeDue(computedJob, now);
                var timeSpan = due - now;

                if (timeSpan.TotalSeconds > 0) {
                    await context.WaitAsync(timeSpan);
                }

                context.ThrowIfStopping();

                using (var scopedContext = context.CreateScope()) {
                    var provider = scopedContext.Provider;

                    var job = provider.GetService<IJob>();
                    var success = true;

                    try {
                        var sw = Stopwatch.StartNew();
                        await job.ExecuteAsync();
                        sw.Stop();
                        computedJob.Retries = 0;
                        _logger.CronJobExecuted(computedJob.Job.Name, sw.Elapsed.TotalSeconds);
                    }
                    catch (Exception ex) {
                        success = false;
                        if (computedJob.Retries == 0) {
                            computedJob.FirstTry = DateTime.UtcNow;
                        }
                        computedJob.Retries++;
                        _logger.CronJobFailed(computedJob.Job.Name, ex);
                    }

                    if (success) {
                        //var connection = provider.GetRequiredService<IStorageConnection>();
                        //await connection.AttachCronJobAsync(computedJob.Job);

                        computedJob.Update(DateTime.UtcNow);

                        //await connection.UpdateCronJobAsync(computedJob.Job);
                    }
                }
            }
        }

        private DateTime ComputeDue(ComputedCronJob computedJob, DateTime now) {
            computedJob.UpdateNext(now);

            var retryBehavior = computedJob.RetryBehavior ?? RetryBehavior.DefaultRetry;
            var retries = computedJob.Retries;

            if (retries == 0) {
                return computedJob.Next;
            }

            var realNext = computedJob.Schedule.GetNextOccurrence(now);

            if (!retryBehavior.Retry) {
                // No retry. If job failed before, we don't care, just schedule it next as usual.
                return realNext;
            }

            if (retries >= retryBehavior.RetryCount) {
                // Max retries. Just schedule it for the next occurance.
                return realNext;
            }

            // Delay a bit.
            return computedJob.FirstTry.AddSeconds(retryBehavior.RetryIn(retries));
        }

        private CronJob[] GetJobs() {
            var cronJobs = new List<CronJob>();
            var entries = _jobRegistry.Build() ?? new CronJobRegistry.Entry[0];
            foreach (var entry in entries) {
                cronJobs.Add(new CronJob {
                    Name = entry.Name,
                    TypeName = entry.JobType.AssemblyQualifiedName,
                    Cron = entry.Cron,
                    LastRun = DateTime.MinValue
                });
            }
            return cronJobs.ToArray();
        }

        private ComputedCronJob[] Compute(IEnumerable<CronJob> jobs, CronJobRegistry.Entry[] entries)
            => jobs.Select(j => CreateComputedCronJob(j, entries)).ToArray();

        private ComputedCronJob CreateComputedCronJob(CronJob job, CronJobRegistry.Entry[] entries) {
            var entry = entries.First(e => e.Name == job.Name);
            return new ComputedCronJob(job, entry);
        }
    }
}
{'fraction_non_alphanumeric': 0.08520421963754396, 'fraction_numerical': 0.0008114687584527996, 'mean_word_length': 1.5428702430077945, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 28, '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': '6336403', 'n_tokens_mistral': 2785, 'n_tokens_neox': 2625, 'n_words': 585}