/// <summary>
        /// Computes the hash-slot that would be used by the given key
        /// </summary>
        public unsafe int HashSlot(RedisKey key)
        {
            //HASH_SLOT = CRC16(key) mod 16384
            if (key.IsNull) return NoSlot;
            unchecked
            {
                var blob = (byte[])key;
                fixed (byte* ptr = blob)
                {
                    int offset = 0, count = blob.Length, start, end;
                    if ((start = IndexOf(ptr, (byte)'{', 0, count - 1)) >= 0
                        && (end = IndexOf(ptr, (byte)'}', start + 1, count)) >= 0
                        && --end != start)
                    {
                        offset = start + 1;
                        count = end - start; // note we already subtracted one via --end
                    }

                    uint crc = 0;
                    for (int i = 0; i < count; i++)
                        crc = ((crc << 8) ^ crc16tab[((crc >> 8) ^ ptr[offset++]) & 0x00FF]) & 0x0000FFFF;
                    return (int)(crc % RedisClusterSlotCount);
                }
            }
        }

        public object Eval(string script, string[] keyArgs, object[] valueArgs)
        {
            RedisKey[] redisKeyArgs = new RedisKey[keyArgs.Length];
            RedisValue[] redisValueArgs = new RedisValue[valueArgs.Length];

            int i = 0;
            foreach (string key in keyArgs)
            {
                redisKeyArgs[i] = key;
                i++;
            }

            i = 0;
            foreach (object val in valueArgs)
            {
                if (val.GetType() == typeof(byte[]))
                {
                    // User data is always in bytes
                    redisValueArgs[i] = (byte[])val;
                }
                else
                {
                    // Internal data like session timeout and indexes are stored as strings
                    redisValueArgs[i] = val.ToString();
                }
                i++;
            }
            return RetryLogic(() => _connection.ScriptEvaluate(script, redisKeyArgs, redisValueArgs));
        }

        /// <summary>
        ///     Creates a new <see cref="IDatabase"/> instance that provides an isolated key space
        ///     of the specified underyling database instance.
        /// </summary>
        /// <param name="database">
        ///     The underlying database instance that the returned instance shall use.
        /// </param>
        /// <param name="keyPrefix">
        ///     The prefix that defines a key space isolation for the returned database instance.
        /// </param>
        /// <returns>
        ///     A new <see cref="IDatabase"/> instance that invokes the specified underlying
        ///     <paramref name="database"/> but prepends the specified <paramref name="keyPrefix"/>
        ///     to all key paramters and thus forms a logical key space isolation.
        /// </returns>
        /// <remarks>
        /// <para>
        ///     The following methods are not supported in a key space isolated database and
        ///     will throw an <see cref="NotSupportedException"/> when invoked:
        /// </para>    
        /// <list type="bullet">
        ///     <item><see cref="IDatabaseAsync.KeyRandomAsync(CommandFlags)"/></item>
        ///     <item><see cref="IDatabase.KeyRandom(CommandFlags)"/></item>
        /// </list>
        /// <para>
        ///     Please notice that keys passed to a script are prefixed (as normal) but care must
        ///     be taken when a script returns the name of a key as that will (currently) not be
        ///     "unprefixed".
        /// </para>
        /// </remarks>
        public static IDatabase WithKeyPrefix(this IDatabase database, RedisKey keyPrefix)
        {
            if (database == null)
            {
                throw new ArgumentNullException("database");
            }

            if (keyPrefix.IsNull)
            {
                throw new ArgumentNullException("keyPrefix");
            }

            if (keyPrefix.IsEmpty)
            {
                return database; // fine - you can keep using the original, then
            }

            if(database is DatabaseWrapper)
            {
                // combine the key in advance to minimize indirection
                var wrapper = (DatabaseWrapper)database;
                keyPrefix = wrapper.ToInner(keyPrefix);
                database = wrapper.Inner;
            }

            return new DatabaseWrapper(database, keyPrefix.AsPrefix());
        }

        public void CanHandleCommands()
        {
            RedisServiceFactory.Register<IRedisCommandHandler, TestCommandHandler>();
            TestCommandHandler cmdHandler = (TestCommandHandler)RedisServiceFactory.CommandHandlers.First();

            bool onExecutingDone = false;
            bool onExecutedDone = false;
            RedisKey[] testKeys = new RedisKey[] { "test" };
            RedisValue[] testValues = new RedisValue[] { "test value" };
            object testResult = (RedisValue)"hello world";

            cmdHandler.onExecuting = (command, involvedKeys, involvedValues) =>
            {
                Assert.AreEqual(RedisCommand.SET, command);
                Assert.AreEqual(1, testKeys.Intersect(involvedKeys).Count());
                Assert.AreEqual(1, testValues.Intersect(involvedValues).Count());
                onExecutingDone = true;
            };
            cmdHandler.onExecuted = (RedisCommand command, ref object result, RedisKey[] involvedKeys) =>
            {
                Assert.AreEqual(RedisCommand.HMSET, command);
                Assert.AreEqual(1, testKeys.Intersect(involvedKeys).Count());
                Assert.AreEqual(testResult, result);
                onExecutedDone = true;
            };

            RedisServiceFactory.CommandHandlers.ExecuteBeforeHandlers(RedisCommand.SET, new RedisKey[] { "test" }, new RedisValue[] { "test value" });
            RedisServiceFactory.CommandHandlers.ExecuteAfterHandlers(RedisCommand.HMSET, new RedisKey[] { "test" }, ref testResult);

            Assert.IsTrue(onExecutingDone);
            Assert.IsTrue(onExecutedDone);
        }

        public RedisLock([NotNull]IDatabase redis, [NotNull]RedisKey key, [NotNull]RedisValue owner, [NotNull]TimeSpan timeOut)
        {
            _redis = redis;
            _key = key;
            _owner = owner;

            //The comparison below uses timeOut as a max timeSpan in waiting Lock
            int i = 0;
            DateTime lockExpirationTime = DateTime.UtcNow +timeOut;
            while (DateTime.UtcNow < lockExpirationTime)
            {
                if (_redis.LockTake(key, owner, timeOut))
                    return;
                //assumes that a second call made by the same owner means an extension request
                var lockOwner = _redis.LockQuery(key);
                
                if (lockOwner.Equals(owner))
                {
                    //extends the lock only for the remaining time
                    var ttl = redis.KeyTimeToLive(key) ?? TimeSpan.Zero;
                    var extensionTTL = lockExpirationTime - DateTime.UtcNow;
                    if (extensionTTL > ttl)
                        _redis.LockExtend(key, owner, extensionTTL - ttl);
                    isRoot = false;
                    return;
                }
                SleepBackOffMultiplier(i);
                i++;
            }
            throw new TimeoutException(string.Format("Lock on {0} with owner identifier {1} Exceeded timeout of {2}", key, owner.ToString(), timeOut));
        }

                /// <summary>
                /// Returns the value associated with field in the hash stored at key.
                /// </summary>
                /// <returns>
                /// the value associated with field, or nil when field is not present in the hash or key does not exist.
                /// </returns>
                /// <remarks>http://redis.io/commands/hget</remarks>
                public static Task<RedisValue> GetString(RedisKey key, RedisValue field)
                {
                    Task<RedisValue> result = SharedCache.Instance.GetReadConnection(key)
                        .GetDatabase(SharedCache.Instance.Db)
                        .HashGetAsync(key, field);

                    return result;
                }

                /// <summary>
                /// If key already exists and is a string, this command appends the value at the end of the string. If key does
                /// not exist it is created and set as an empty string, so APPEND will be similar to SET in this special case.
                /// </summary>
                /// <param name="key">
                /// The key.
                /// </param>
                /// <param name="value">
                /// The value.
                /// </param>
                /// <returns>
                /// the length of the string after the append operation.
                /// </returns>
                /// <remarks>
                /// http://redis.io/commands/append
                /// </remarks>
                public static Task<long> AppendAsync(RedisKey key, RedisValue value)
                {
                    Task<long> result = SharedCache.Instance.GetWriteConnection(key)
                        .GetDatabase(SharedCache.Instance.Db)
                        .StringAppendAsync(key, value);

                    return result;
                }

 public bool Lock(RedisKey resource, TimeSpan ttl, out Lock lockObject)
 {
     var task = LockAsync(resource, ttl);
     task.Wait();
     var result = task.Result;
     lockObject = result.Item2;
     return result.Item1;
 }

                /// <summary>
                /// Removes the specified key. A key is ignored if it does not exist.
                /// </summary>
                /// <param name="key">
                /// The key.
                /// </param>
                /// <returns>
                /// True if the key was removed.
                /// </returns>
                /// <remarks>
                /// http://redis.io/commands/del
                /// </remarks>
                public static bool Delete(RedisKey key)
                {
                    bool result = SharedCache.Instance.GetWriteConnection(key)
                        .GetDatabase(SharedCache.Instance.Db)
                        .KeyDelete(key);

                    return result;
                }

                /// <summary>
                /// Removes the specified fields from the hash stored at key. Non-existing fields are ignored. Non-existing keys
                /// are treated as empty hashes and this command returns 0.
                /// </summary>
                /// <param name="key">
                /// The key.
                /// </param>
                /// <param name="field">
                /// The field.
                /// </param>
                /// <remarks>
                /// http://redis.io/commands/hdel
                /// </remarks>
                /// <returns>
                /// The number of fields that were removed.
                /// </returns>
                public static bool Delete(RedisKey key, RedisValue field)
                {
                    bool result = SharedCache.Instance.GetWriteConnection(key)
                        .GetDatabase(SharedCache.Instance.Db)
                        .HashDelete(key, field);

                    return result;
                }

                /// <summary>
                /// Removes the specified key. A key is ignored if it does not exist.
                /// </summary>
                /// <param name="key">
                /// The key.
                /// </param>
                /// <returns>
                /// True if the key was removed.
                /// </returns>
                /// <remarks>
                /// http://redis.io/commands/del
                /// </remarks>
                public static Task<bool> DeleteAsync(RedisKey key)
                {
                    Task<bool> result = SharedCache.Instance.GetWriteConnection(key)
                        .GetDatabase(SharedCache.Instance.Db)
                        .KeyDeleteAsync(key);

                    return result;
                }

        public void CommandStart(RedisSettings usedSettings, string command, RedisKey key)
        {
            if (TimerStrategy == null) return;
            this.start = TimerStrategy.Start();

            this.usedSettings = usedSettings;
            this.command = command;
            this.key = key;
        }

        /// <summary>
        /// Executes all Redis command handlers, running behaviors that must run after some given command has been already executed.
        /// </summary>
        /// <typeparam name="TResult">The type of Redis command result</typeparam>
        /// <param name="handlers">The sequence of Redis command handlers</param>
        /// <param name="command">The Redis command</param>
        /// <param name="involvedKeys">An array of involved Redis keys in the command</param>
        /// <param name="result">The result of the Redis command execution</param>
        /// <returns>True if handlers could be executed. Otherwise, false.</returns>
        public static bool ExecuteAfterHandlers(this IEnumerable<IRedisCommandHandler> handlers, RedisCommand command, RedisKey[] involvedKeys, ref object result)
        {
            bool canExecute = CanExecuteHandlers(handlers, command);

            if (canExecute)
                foreach (IRedisCommandHandler handler in handlers)
                    handler.OnExecuted(command, ref result, involvedKeys);

            return canExecute;
        }

        /// <summary>
        /// Executes all Redis command handlers, running behaviors that must run before some given command is about to get executed.
        /// </summary>
        /// <param name="handlers">The sequence of Redis command handlers</param>
        /// <param name="command">The Redis command</param>
        /// <param name="involvedKeys">An array of involved Redis keys in the command</param>
        /// <param name="involvedValues">An array of involved Redis values in the command</param>
        /// <returns>True if handlers could be executed. Otherwise, false.</returns>
        public static bool ExecuteBeforeHandlers(this IEnumerable<IRedisCommandHandler> handlers, RedisCommand command, RedisKey[] involvedKeys = null, RedisValue[] involvedValues = null)
        {
            bool canExecute = CanExecuteHandlers(handlers, command);

            if (canExecute)
                foreach (IRedisCommandHandler handler in handlers)
                    handler.OnExecuting(command, involvedKeys, involvedValues);

            return canExecute;
        }

        internal static void SetUniqueVal(RedisKey hashKey, RedisValue hashField, RedisValue value, RedisKey indexKey)
        {
            var result = Store.Database.ScriptEvaluate(Script, new {hashKey, hashField, finalVal = value, indexKey});
            if ((bool)result)
            {
                return;
            }

            throw new UniqueConstraintViolatedException();
        }

 public async Task<IEnumerable<DeviceToken>> GetUserDeviceTokensAsync(IEnumerable<string> userIds)
 {
     var keys = new RedisKey[userIds.Count()];
     for (int i = 0; i < userIds.Count(); i++)
     {
         keys[i] = userIds.ElementAt(i);
     }
     var dtJsons = await pubsubRedis.GetDatabase().StringGetAsync(keys.ToArray(), CommandFlags.PreferSlave);
     return from item in dtJsons select JsonToDeviceToken(item);
 }

 public RedisTimelineMessage(RedisSettings usedSettings, string command, RedisKey key, object sentObject, long sentSize, object receivedObject, long receivedSize, bool isError)
 {
     UsedSettings = usedSettings;
     Command = command;
     Key = key;
     SentObject = sentObject;
     SentSize = sentSize;
     ReceivedObject = receivedObject;
     ReceivedSize = receivedSize;
     IsError = isError;
 }

 /// <summary>
 /// Removes the specified fields from the hash stored at key. Non-existing fields are ignored. Non-existing keys
 /// are treated as empty hashes and this command returns 0.
 /// </summary>
 /// <remarks>http://redis.io/commands/hdel</remarks>
 /// <returns>The number of fields that were removed.</returns>
 public static Task<long> Remove(RedisKey key, RedisValue[] fields)
 {
     var connections = SharedCache.Instance.GetWriteConnections(key);
     Task<long> result = null;
     foreach (var connection in connections)
     {
         var task = connection.GetDatabase(SharedCache.Instance.Db)
             .HashDeleteAsync(key, fields);
         if (null == result)
             result = task;
     }
     return result;
 }

 /// <summary>
 /// Returns if field is an existing field in the hash stored at key.
 /// </summary>
 /// <returns>
 /// 1 if the hash contains field. 0 if the hash does not contain field, or key does not exist.
 /// </returns>
 /// <remarks>http://redis.io/commands/hexists</remarks>
 public static Task<bool> Exists(RedisKey key, RedisValue field)
 {
     var connections = SharedCache.Instance.GetWriteConnections(key);
     Task<bool> result = null;
     foreach (var connection in connections)
     {
         var task = connection.GetDatabase(SharedCache.Instance.Db)
             .HashExistsAsync(key, field);
         if (null == result)
             result = task;
     }
     return result;
 }

        public Task ScriptEvaluateAsync(int database, string script, string key, byte[] messageArguments)
        {
            if (_connection == null)
            {
                throw new InvalidOperationException(Resources.Error_RedisConnectionNotStarted);
            }

            var keys = new RedisKey[] { key };

            var arguments = new RedisValue[] { messageArguments };

          return _connection.GetDatabase(database).ScriptEvaluateAsync(script, keys, arguments);
        }