<commit_before>﻿using System;
using System.Threading;
using System.Threading.Tasks;
using Foundatio.Extensions;
using Foundatio.Queues;
using Foundatio.Utility;
using Foundatio.Logging;

namespace Foundatio.Jobs {
    public abstract class QueueProcessorJobBase<T> : JobBase, IQueueProcessorJob where T : class {
        protected readonly IQueue<T> _queue;

        public QueueProcessorJobBase(IQueue<T> queue) {
            _queue = queue;
            AutoComplete = true;
        }

        protected bool AutoComplete { get; set; }

        protected override async Task<JobResult> RunInternalAsync(CancellationToken cancellationToken) {
            var linkedCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, TimeSpan.FromSeconds(30).ToCancellationToken());

            QueueEntry<T> queueEntry;
            try {
                queueEntry = await _queue.DequeueAsync(linkedCancellationToken.Token).AnyContext();
            } catch (Exception ex) {
                Logger.Error().Exception(ex).Message("Error trying to dequeue message: {0}", ex.Message).Write();
                return JobResult.FromException(ex);
            }

            if (queueEntry == null)
                return JobResult.Success;
            
            using (var lockValue = await GetQueueItemLockAsync(queueEntry, cancellationToken).AnyContext()) {
                if (lockValue == null)
                    return JobResult.SuccessWithMessage("Unable to acquire queue item lock.");

                Logger.Trace().Message("Processing queue entry '{0}'.", queueEntry.Id).Write();
                try {
                    var result = await ProcessQueueItemAsync(queueEntry, cancellationToken).AnyContext();

                    if (!AutoComplete)
                        return result;

                    if (result.IsSuccess)
                        await queueEntry.CompleteAsync().AnyContext();
                    else
                        await queueEntry.AbandonAsync().AnyContext();
<commit_msg>Added auto abandon support to the queue job if the cancellation token was cancelled.<commit_after>﻿using System;
using System.Threading;
using System.Threading.Tasks;
using Foundatio.Extensions;
using Foundatio.Queues;
using Foundatio.Utility;
using Foundatio.Logging;

namespace Foundatio.Jobs {
    public abstract class QueueProcessorJobBase<T> : JobBase, IQueueProcessorJob where T : class {
        protected readonly IQueue<T> _queue;

        public QueueProcessorJobBase(IQueue<T> queue) {
            _queue = queue;
            AutoComplete = true;
        }

        protected bool AutoComplete { get; set; }

        protected override async Task<JobResult> RunInternalAsync(CancellationToken cancellationToken) {
            var linkedCancellationToken = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, TimeSpan.FromSeconds(30).ToCancellationToken());

            QueueEntry<T> queueEntry;
            try {
                queueEntry = await _queue.DequeueAsync(linkedCancellationToken.Token).AnyContext();
            } catch (Exception ex) {
                Logger.Error().Exception(ex).Message($"Error trying to dequeue message: {ex.Message}").Write();
                return JobResult.FromException(ex);
            }

            if (queueEntry == null)
                return JobResult.Success;

            if (cancellationToken.IsCancellationRequested) {
                Logger.Info().Message($"Job was cancelled. Abandoning queue item: {queueEntry.Id}").Write();
                await queueEntry.AbandonAsync().AnyContext();
                return JobResult.Cancelled;
            }

            using (var lockValue = await GetQueueItemLockAsync(queueEntry, cancellationToken).AnyContext()) {
                if (lockValue == null)
                    return JobResult.SuccessWithMessage("Unable to acquire queue item lock.");

                Logger.Trace().Message("Processing queue entry '{0}'.", queueEntry.Id).Write();
                try {
                    var result = await ProcessQueueItemAsync(queueEntry, cancellationToken).AnyContext();

                    if (!AutoComplete)
                        return result;

                    if (result.IsSuccess)
                        await queueEntry.CompleteAsync().AnyContext();
                    else
                        await queueEntry.AbandonAsync().AnyContext();

{'fraction_non_alphanumeric': 0.08800904977375566, 'fraction_numerical': 0.001583710407239819, 'mean_word_length': 2.262730627306273, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, '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': '4497552', 'n_tokens_mistral': 1051, 'n_tokens_neox': 1026, 'n_words': 249}