<commit_before>using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using Newtonsoft.Json.Linq;
using Orchard.Environment;
using Orchard.Events;
using Orchard.Logging;
using Orchard.JobsQueue.Models;
using Orchard.Services;
using Orchard.TaskLease.Services;

namespace Orchard.JobsQueue.Services {
    public class JobsQueueProcessor : IJobsQueueProcessor {
        private readonly Work<IJobsQueueManager> _jobsQueueManager;
        private readonly Work<IClock> _clock;
        private readonly Work<ITaskLeaseService> _taskLeaseService;
        private readonly IEventBus _eventBus;
        private readonly ReaderWriterLockSlim _rwl = new ReaderWriterLockSlim();

        public JobsQueueProcessor(
            Work<IClock> clock,
            Work<IJobsQueueManager> jobsQueueManager,
            Work<ITaskLeaseService> taskLeaseService,
            IEventBus eventBus) {
            _clock = clock;
            _jobsQueueManager = jobsQueueManager;
            _taskLeaseService = taskLeaseService;
            _eventBus = eventBus;
            Logger = NullLogger.Instance;
        }

        public ILogger Logger { get; set; }
        public void ProcessQueue() {
            // prevent two threads on the same machine to process the message queue
            if (_rwl.TryEnterWriteLock(0)) {
                try {
                    _taskLeaseService.Value.Acquire("JobsQueueProcessor", _clock.Value.UtcNow.AddMinutes(5));
                    IEnumerable<QueuedJobRecord> messages;

                    while ((messages = _jobsQueueManager.Value.GetJobs(0, 10).ToArray()).Any()) {
                        foreach (var message in messages) {
                            ProcessMessage(message);
                        }
                    }
                }
                finally {
                    _rwl.ExitWriteLock();
                }
            }
        }

        private void ProcessMessage(QueuedJobRecord job) {

            Logger.Debug("Processing job {0}.", job.Id);

            try {
                var payload = JObject.Parse(job.Parameters);
                var parameters = payload.ToDictionary();

                _eventBus.Notify(job.Message, parameters);<commit_msg>Fixing that JobsQueue doesn't actually use Task Lease to prevent multiple server nodes from executing the same jobs<commit_after>using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using Newtonsoft.Json.Linq;
using Orchard.Environment;
using Orchard.Events;
using Orchard.Logging;
using Orchard.JobsQueue.Models;
using Orchard.Services;
using Orchard.TaskLease.Services;

namespace Orchard.JobsQueue.Services {
    public class JobsQueueProcessor : IJobsQueueProcessor {
        private readonly Work<IJobsQueueManager> _jobsQueueManager;
        private readonly Work<IClock> _clock;
        private readonly Work<ITaskLeaseService> _taskLeaseService;
        private readonly IEventBus _eventBus;
        private readonly ReaderWriterLockSlim _rwl = new ReaderWriterLockSlim();

        public JobsQueueProcessor(
            Work<IClock> clock,
            Work<IJobsQueueManager> jobsQueueManager,
            Work<ITaskLeaseService> taskLeaseService,
            IEventBus eventBus) {
            _clock = clock;
            _jobsQueueManager = jobsQueueManager;
            _taskLeaseService = taskLeaseService;
            _eventBus = eventBus;
            Logger = NullLogger.Instance;
        }

        public ILogger Logger { get; set; }
        public void ProcessQueue() {
            // prevent two threads on the same machine to process the message queue
            if (_rwl.TryEnterWriteLock(0)) {
                try {
                    if (_taskLeaseService.Value.Acquire("JobsQueueProcessor", _clock.Value.UtcNow.AddMinutes(5)) != null) {
                        IEnumerable<QueuedJobRecord> messages;

                        while ((messages = _jobsQueueManager.Value.GetJobs(0, 10).ToArray()).Any()) {
                            foreach (var message in messages) {
                                ProcessMessage(message);
                            }
                        }
                    }
                }
                finally {
                    _rwl.ExitWriteLock();
                }
            }
        }

        private void ProcessMessage(QueuedJobRecord job) {

            Logger.Debug("Processing job {0}.", job.Id);

            try {
                var payload = JObject.Parse(job.Parameters);
                var parameters = payload.ToDictionary();

                _eventBus.Notify(job.Message, parameters);
{'fraction_non_alphanumeric': 0.07394668959587274, 'fraction_numerical': 0.0025795356835769563, 'mean_word_length': 1.96936821952776, 'pattern_counts': {'":': 0, '<': 17, '<?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': '3958235', 'n_tokens_mistral': 1207, 'n_tokens_neox': 1133, 'n_words': 270}