SkipIfExistsAttribute.cs

SkipIfExistsAttribute.cs

// Zero-Clause BSD (more permissive than MIT, doesn't require copyright notice)
// 
// Permission to use, copy, modify, and/or distribute this software for any purpose
// with or without fee is hereby granted.
// 
// THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
// WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
// AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
// INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS
// OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER
// TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF
// THIS SOFTWARE.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Hangfire;
using Hangfire.Client;
using Hangfire.Common;
using Hangfire.MemoryStorage;
using Hangfire.States;
using Hangfire.Storage;

namespace ConsoleApplication54
{
    public class SkipIfExistsAttribute : JobFilterAttribute, IElectStateFilter
    {
        private readonly string _keyFormat;

        public SkipIfExistsAttribute(string keyFormat)
        {
            _keyFormat = keyFormat;
        }

        public void OnStateElection(ElectStateContext context)
        {
            var connection = context.Connection as JobStorageConnection;
            if (connection == null) throw new NotSupportedException("Storage does not support extended command set");

            if (context.CandidateState is ScheduledState)
            {
                using (context.Connection.AcquireDistributedLock(GetLockKey(context.BackgroundJob.Job), TimeSpan.FromSeconds(15)))
                {
                    var currentId = connection.GetValueFromHash(GetKey(context.BackgroundJob.Job), "id");
                    if (!String.IsNullOrEmpty(currentId) && currentId != context.BackgroundJob.Id)
                    {
                        // Other job is currently executed, we should retry or delete
                        context.CandidateState = new DeletedState();
                        return;
                    }

                    if (currentId != context.BackgroundJob.Id)
                    {
                        // Slot is free, let's own it
                        using (var transaction = connection.CreateWriteTransaction())
                        {
                            transaction.SetRangeInHash(GetKey(context.BackgroundJob.Job), new []
                            {
                                new KeyValuePair<string, string>("id", context.BackgroundJob.Id), 
                            });

                            transaction.Commit();
                        }
                    }
                }
            }
            else if (context.CandidateState.IsFinal)
            {
                using (context.Connection.AcquireDistributedLock(GetLockKey(context.BackgroundJob.Job), TimeSpan.FromMinutes(1)))
                {
                    var currentId = connection.GetValueFromHash(GetKey(context.BackgroundJob.Job), "id");

                    if (currentId == context.BackgroundJob.Id)
                    {
                        // Cleaning up, but only if we own that slot
                        using (var transaction = connection.CreateWriteTransaction())
                        {
                            transaction.RemoveHash(GetKey(context.BackgroundJob.Job));
                            transaction.Commit();
                        }
                    }
                }
            }
        }

        private string GetKey(Job job)
        {
            return String.Format("skipifexists:" + _keyFormat, job.Args.ToArray());
        }

        private string GetLockKey(Job job)
        {
            return String.Format("skipifexists:lock:" + _keyFormat, job.Args.ToArray());
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            GlobalConfiguration.Configuration.UseMemoryStorage();

            BackgroundJob.Schedule(() => MyMethod(123), TimeSpan.FromSeconds(2));
            BackgroundJob.Schedule(() => MyMethod(456), TimeSpan.FromSeconds(2));
            BackgroundJob.Schedule(() => MyMethod(123), TimeSpan.FromSeconds(2));
            BackgroundJob.Schedule(() => MyMethod(123), TimeSpan.FromSeconds(2));
            BackgroundJob.Schedule(() => MyMethod(123), TimeSpan.FromSeconds(2));
            BackgroundJob.Schedule(() => MyMethod(123), TimeSpan.FromSeconds(2));

            using (new BackgroundJobServer(new BackgroundJobServerOptions { SchedulePollingInterval = TimeSpan.FromSeconds(1) }))
            {
                Thread.Sleep(TimeSpan.FromSeconds(5));

                BackgroundJob.Schedule(() => MyMethod(123), TimeSpan.FromSeconds(2));
                BackgroundJob.Schedule(() => MyMethod(123), TimeSpan.FromSeconds(2));
                BackgroundJob.Schedule(() => MyMethod(123), TimeSpan.FromSeconds(2));

                Console.ReadLine();
            }
        }

        [SkipIfExists("id:{0}")]
        public static void MyMethod(long id)
        {
            Console.WriteLine($"{DateTime.UtcNow}: {id}");
        }
    }
}