FailoverRedisStorage

FailoverRedisStorage.cs

using System;
using System.Collections.Generic;
using System.Collections.ObjectModel;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Hangfire;
using Hangfire.Logging;
using Hangfire.Redis;
using Hangfire.Storage;
using ServiceStack.Redis;

namespace ConsoleSample
{
    public class FailoverRedisStorage : RedisStorage
    {
        private static readonly ILog Logger = LogProvider.GetCurrentClassLogger();

        private readonly object _currentStorageLock = new object();

        private readonly ReadOnlyDictionary<string, HostInfo> _hosts;

        private string _currentHost;
        private RedisStorage _currentStorage;

        private readonly Thread _thread;

        public FailoverRedisStorage(params string[] hosts)
            : base(hosts[0], 0, new RedisStorageOptions())
        {
            _hosts = new ReadOnlyDictionary<string, HostInfo>(hosts.ToDictionary(x => x, x => new HostInfo()));

            _thread = new Thread(CheckConnections) { IsBackground = true };
            _thread.Start();

            ReconfigureStorage();
        }

        public override IMonitoringApi GetMonitoringApi()
        {
            return GetCurrentStorage().GetMonitoringApi();
        }

        public override IStorageConnection GetConnection()
        {
            return GetCurrentStorage().GetConnection();
        }
        
        public override string ToString()
        {
            lock (_currentStorageLock)
            {
                if (_currentStorage == null)
                {
                    return "Master is not available";
                }

                return GetCurrentStorage().ToString();
            }
        }

        private void ReconfigureStorage()
        {
            var available = _hosts.Where(x => x.Value.IsAvailable && x.Value.IsMaster).ToList();

            if (available.Count != 0)
            {
                SetCurrentStorage(available[0].Key);
            }
            else
            {
                _currentStorage = null;
            }
        }

        private JobStorage GetCurrentStorage()
        {
            lock (_currentStorageLock)
            {
                if (_currentStorage == null)
                {
                    throw new InvalidOperationException("Redis master is not temporary available.");
                }

                return _currentStorage;
            }
        }

        private void SetCurrentStorage(string hostAndPort)
        {
            lock (_currentStorageLock)
            {
                _currentHost = hostAndPort;

                if (_currentHost == null)
                {
                    _currentStorage = null;
                }
                else
                {
                    _currentStorage = new RedisStorage(_currentHost);
                    WriteOptionsToLog(Logger);
                }
            }
        }

        private void CheckConnections()
        {
            while (true)
            {
                var tasks = new List<Task>();

                foreach (var host in _hosts)
                {
                    var currentHost = host;

                    tasks.Add(Task.Factory.StartNew(() =>
                    {
                        var hostAndPort = currentHost.Key.Split(':');

                        try
                        {
                            using (var redis = new RedisClient(hostAndPort[0], int.Parse(hostAndPort[1])))
                            {
                                var info = redis.Info;

                                currentHost.Value.IsAvailable = true;
                                currentHost.Value.IsMaster = info["role"].Equals("master",
                                    StringComparison.OrdinalIgnoreCase);
                            }
                        }
                        catch (Exception)
                        {
                            currentHost.Value.IsAvailable = false;
                        }
                    }));
                }

                Task.WaitAll(tasks.ToArray());

                lock (_currentStorageLock)
                {
                    if (_currentHost == null || !_hosts[_currentHost].IsAvailable || !_hosts[_currentHost].IsMaster)
                    {
                        ReconfigureStorage();
                    }
                }

                Thread.Sleep(5000);
            }
        }

        private class HostInfo
        {
            private readonly object _lock = new object();

            private bool _isMaster;
            private bool _isAvailable;

            public bool IsAvailable
            {
                get { lock (_lock) { return _isAvailable; } }
                set { lock (_lock) { _isAvailable = value; } }
            }

            public bool IsMaster
            {
                get { lock (_lock) { return _isMaster; } }
                set { lock (_lock) { _isMaster = value; } }
            }
        }
    }
}