//

// redis-sharp.cs: ECMA CLI Binding to the Redis key-value storage system

//

// Authors:

// Miguel de Icaza (miguel@gnome.org)

//

// Copyright 2010 Novell, Inc.

//

// Licensed under the same terms of Redis: new BSD license.

//

//#define DEBUG

using System;

using System.Globalization;

using System.IO;

using System.Collections.Generic;

using System.Linq;

using System.Net.Security;

using System.Net.Sockets;

using System.Threading;

using ServiceStack.Logging;

using ServiceStack.Redis.Pipeline;

using ServiceStack.Text;

namespace ServiceStack.Redis

{

/// <summary>

/// This class contains all the common operations for the RedisClient.

/// The client contains a 1:1 mapping of c# methods to redis operations of the same name.

///

/// Not threadsafe, use a pooled manager!

/// All redis calls on a single instances write to the same Socket.

/// If used in multiple threads (or async Tasks) at the same time you will find

/// that commands are not executed properly by redis and Servicestack wont be able to (json) serialize

/// the data that comes back.

/// </summary>

public partial class RedisNativeClient

: IRedisNativeClient

{

private static readonly ILog log = LogManager.GetLogger(typeof(RedisNativeClient));

internal const int Success = 1;

internal const int OneGb = 1073741824;

private readonly byte[] endData = new[] { (byte)'\r', (byte)'\n' };

private int clientPort;

private string lastCommand;

private SocketException lastSocketException;

internal long deactivatedAtTicks;

public DateTime? DeactivatedAt

{

get

{

return deactivatedAtTicks != 0

? new DateTime(Interlocked.Read(ref deactivatedAtTicks), DateTimeKind.Utc)

: (DateTime?)null;

}

set

{

var ticksValue = value == null ? 0 : value.Value.Ticks;

Interlocked.Exchange(ref deactivatedAtTicks, ticksValue);

}

}

public bool HadExceptions { get { return deactivatedAtTicks > 0; } }

protected Socket socket;

protected BufferedStream Bstream;

protected SslStream sslStream;

private IRedisTransactionBase transaction;

private IRedisPipelineShared pipeline;

const int YES = 1;

const int NO = 0;

/// <summary>

/// Used to manage connection pooling

/// </summary>

private int active;

internal bool Active

{

get

{

return Interlocked.CompareExchange(ref active, 0, 0) == YES;

}

set

{

Interlocked.Exchange(ref active, value ? YES : NO);

}

}

internal IHandleClientDispose ClientManager { get; set; }

internal long LastConnectedAtTimestamp;

public long Id { get; set; }

public string Host { get; private set; }

public int Port { get; private set; }

public bool Ssl { get; private set; }

/// <summary>

/// Gets or sets object key prefix.

/// </summary>

public string NamespacePrefix { get; set; }

public int ConnectTimeout { get; set; }

private TimeSpan retryTimeout;

public int RetryTimeout

{

get { return (int)retryTimeout.TotalMilliseconds; }

set { retryTimeout = TimeSpan.FromMilliseconds(value); }

}

public int RetryCount { get; set; }

public int SendTimeout { get; set; }

public int ReceiveTimeout { get; set; }

public string Password { get; set; }

public string Client { get; set; }

public int IdleTimeOutSecs { get; set; }

public Action<IRedisNativeClient> ConnectionFilter { get; set; }

internal IRedisTransactionBase Transaction

{

get

{

return transaction;

}

set

{

if (value != null)

AssertConnectedSocket();

transaction = value;

}

}

internal IRedisPipelineShared Pipeline

{

get

{

return pipeline;

}

set

{

if (value != null)

AssertConnectedSocket();

pipeline = value;

}

}

internal void EndPipeline()

{

ResetSendBuffer();

if (Pipeline != null)

{

Pipeline = null;

Interlocked.Increment(ref __requestsPerHour);

}

}

public RedisNativeClient(string connectionString)

: this(connectionString.ToRedisEndpoint()) { }

public RedisNativeClient(RedisEndpoint config)

{

Init(config);

}

public RedisNativeClient(string host, int port)

: this(host, port, null) { }

public RedisNativeClient(string host, int port, string password = null, long db = RedisConfig.DefaultDb)

{

if (host == null)

throw new ArgumentNullException("host");

Init(new RedisEndpoint(host, port, password, db));

}

private void Init(RedisEndpoint config)

{

Host = config.Host;

Port = config.Port;

ConnectTimeout = config.ConnectTimeout;

SendTimeout = config.SendTimeout;

ReceiveTimeout = config.ReceiveTimeout;

RetryTimeout = config.RetryTimeout;

Password = config.Password;

NamespacePrefix = config.NamespacePrefix;

Client = config.Client;

Db = config.Db;

Ssl = config.Ssl;

IdleTimeOutSecs = config.IdleTimeOutSecs;

ServerVersionNumber = RedisConfig.AssumeServerVersion.GetValueOrDefault();

JsConfig.InitStatics();

}

public RedisNativeClient()

: this(RedisConfig.DefaultHost, RedisConfig.DefaultPort) { }

#region Common Operations

long db;

public long Db

{

get

{

return db;

}

set

{

db = value;

if (HasConnected)

{

ChangeDb(db);

}

}

}

public void ChangeDb(long db)

{

this.db = db;

SendExpectSuccess(Commands.Select, db.ToUtf8Bytes());

}

public long DbSize

{

get

{

return SendExpectLong(Commands.DbSize);

}

}

public DateTime LastSave

{

get

{

var t = SendExpectLong(Commands.LastSave);

return t.FromUnixTime();

}

}

public Dictionary<string, string> Info

{

get

{

var lines = SendExpectString(Commands.Info);

var info = new Dictionary<string, string>();

foreach (var line in lines

.Split(new[] { "\r\n" }, StringSplitOptions.RemoveEmptyEntries))

{

var p = line.IndexOf(':');

if (p == -1) continue;

info.Add(line.Substring(0, p), line.Substring(p + 1));

}

return info;

}

}

public string ServerVersion

{

get

{

string version;

this.Info.TryGetValue("redis_version", out version);

return version;

}

}

public RedisData RawCommand(params object[] cmdWithArgs)

{

var byteArgs = new List<byte[]>();

foreach (var arg in cmdWithArgs)

{

if (arg == null)

{

byteArgs.Add(TypeConstants.EmptyByteArray);

continue;

}

var bytes = arg as byte[];

if (bytes != null)

{

byteArgs.Add(bytes);

}

else if (arg.GetType().IsUserType())

{

var json = arg.ToJson();

byteArgs.Add(json.ToUtf8Bytes());

}

else

{

var str = arg.ToString();

byteArgs.Add(str.ToUtf8Bytes());

}

}

var data = SendExpectComplexResponse(byteArgs.ToArray());

return data;

}

public RedisData RawCommand(params byte[][] cmdWithBinaryArgs)

{

return SendExpectComplexResponse(cmdWithBinaryArgs);

}

public bool Ping()

{

return SendExpectCode(Commands.Ping) == "PONG";

}

public string Echo(string text)

{

return SendExpectData(Commands.Echo, text.ToUtf8Bytes()).FromUtf8Bytes();

}

public void SlaveOf(string hostname, int port)

{

SendExpectSuccess(Commands.SlaveOf, hostname.ToUtf8Bytes(), port.ToUtf8Bytes());

}

public void SlaveOfNoOne()

{

SendExpectSuccess(Commands.SlaveOf, Commands.No, Commands.One);

}

public byte[][] ConfigGet(string pattern)

{

return SendExpectMultiData(Commands.Config, Commands.Get, pattern.ToUtf8Bytes());

}

public void ConfigSet(string item, byte[] value)

{

SendExpectSuccess(Commands.Config, Commands.Set, item.ToUtf8Bytes(), value);

}

public void ConfigResetStat()

{

SendExpectSuccess(Commands.Config, Commands.ResetStat);

}

public void ConfigRewrite()

{

SendExpectSuccess(Commands.Config, Commands.Rewrite);

}

public byte[][] Time()

{

return SendExpectMultiData(Commands.Time);

}

public void DebugSegfault()

{

SendExpectSuccess(Commands.Debug, Commands.Segfault);

}

public void DebugSleep(double durationSecs)

{

SendExpectSuccess(Commands.Debug, Commands.Sleep, durationSecs.ToUtf8Bytes());

}

public byte[] Dump(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectData(Commands.Dump, key.ToUtf8Bytes());

}

public byte[] Restore(string key, long expireMs, byte[] dumpValue)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectData(Commands.Restore, key.ToUtf8Bytes(), expireMs.ToUtf8Bytes(), dumpValue);

}

public void Migrate(string host, int port, string key, int destinationDb, long timeoutMs)

{

if (key == null)

throw new ArgumentNullException("key");

SendExpectSuccess(Commands.Migrate, host.ToUtf8Bytes(), port.ToUtf8Bytes(), key.ToUtf8Bytes(), destinationDb.ToUtf8Bytes(), timeoutMs.ToUtf8Bytes());

}

public bool Move(string key, int db)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.Move, key.ToUtf8Bytes(), db.ToUtf8Bytes()) == Success;

}

public long ObjectIdleTime(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.Object, Commands.IdleTime, key.ToUtf8Bytes());

}

public string Type(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectCode(Commands.Type, key.ToUtf8Bytes());

}

public RedisKeyType GetEntryType(string key)

{

switch (Type(key))

{

case "none":

return RedisKeyType.None;

case "string":

return RedisKeyType.String;

case "set":

return RedisKeyType.Set;

case "list":

return RedisKeyType.List;

case "zset":

return RedisKeyType.SortedSet;

case "hash":

return RedisKeyType.Hash;

}

throw CreateResponseError("Invalid value");

}

public long StrLen(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.StrLen, key.ToUtf8Bytes());

}

public void Set(string key, byte[] value)

{

if (key == null)

throw new ArgumentNullException("key");

value = value ?? TypeConstants.EmptyByteArray;

if (value.Length > OneGb)

throw new ArgumentException("value exceeds 1G", "value");

SendExpectSuccess(Commands.Set, key.ToUtf8Bytes(), value);

}

public void Set(string key, byte[] value, int expirySeconds, long expiryMs = 0)

{

Set(key.ToUtf8Bytes(), value, expirySeconds, expiryMs);

}

public void Set(byte[] key, byte[] value, int expirySeconds, long expiryMs = 0)

{

if (key == null)

throw new ArgumentNullException("key");

value = value ?? TypeConstants.EmptyByteArray;

if (value.Length > OneGb)

throw new ArgumentException("value exceeds 1G", "value");

if (expirySeconds > 0)

SendExpectSuccess(Commands.Set, key, value, Commands.Ex, expirySeconds.ToUtf8Bytes());

else if (expiryMs > 0)

SendExpectSuccess(Commands.Set, key, value, Commands.Px, expiryMs.ToUtf8Bytes());

else

SendExpectSuccess(Commands.Set, key, value);

}

public bool Set(string key, byte[] value, bool exists, int expirySeconds = 0, long expiryMs = 0)

{

var entryExists = exists ? Commands.Xx : Commands.Nx;

if (expirySeconds > 0)

return SendExpectString(Commands.Set, key.ToUtf8Bytes(), value, Commands.Ex, expirySeconds.ToUtf8Bytes(), entryExists) == OK;

if (expiryMs > 0)

return SendExpectString(Commands.Set, key.ToUtf8Bytes(), value, Commands.Px, expiryMs.ToUtf8Bytes(), entryExists) == OK;

return SendExpectString(Commands.Set, key.ToUtf8Bytes(), value, entryExists) == OK;

}

public void SetEx(string key, int expireInSeconds, byte[] value)

{

SetEx(key.ToUtf8Bytes(), expireInSeconds, value);

}

public void SetEx(byte[] key, int expireInSeconds, byte[] value)

{

if (key == null)

throw new ArgumentNullException("key");

value = value ?? TypeConstants.EmptyByteArray;

if (value.Length > OneGb)

throw new ArgumentException("value exceeds 1G", "value");

SendExpectSuccess(Commands.SetEx, key, expireInSeconds.ToUtf8Bytes(), value);

}

public bool Persist(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.Persist, key.ToUtf8Bytes()) == Success;

}

public void PSetEx(string key, long expireInMs, byte[] value)

{

if (key == null)

throw new ArgumentNullException("key");

SendExpectSuccess(Commands.PSetEx, key.ToUtf8Bytes(), expireInMs.ToUtf8Bytes(), value);

}

public long SetNX(string key, byte[] value)

{

if (key == null)

throw new ArgumentNullException("key");

value = value ?? TypeConstants.EmptyByteArray;

if (value.Length > OneGb)

throw new ArgumentException("value exceeds 1G", "value");

return SendExpectLong(Commands.SetNx, key.ToUtf8Bytes(), value);

}

public void MSet(byte[][] keys, byte[][] values)

{

var keysAndValues = MergeCommandWithKeysAndValues(Commands.MSet, keys, values);

SendExpectSuccess(keysAndValues);

}

public void MSet(string[] keys, byte[][] values)

{

MSet(keys.ToMultiByteArray(), values);

}

public bool MSetNx(byte[][] keys, byte[][] values)

{

var keysAndValues = MergeCommandWithKeysAndValues(Commands.MSet, keys, values);

return SendExpectLong(keysAndValues) == Success;

}

public bool MSetNx(string[] keys, byte[][] values)

{

return MSetNx(keys.ToMultiByteArray(), values);

}

public byte[] Get(string key)

{

return GetBytes(key);

}

public byte[] Get(byte[] key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectData(Commands.Get, key);

}

public object[] Slowlog(int? top)

{

if (top.HasValue)

return SendExpectDeeplyNestedMultiData(Commands.Slowlog, Commands.Get, top.Value.ToUtf8Bytes());

else

return SendExpectDeeplyNestedMultiData(Commands.Slowlog, Commands.Get);

}

public void SlowlogReset()

{

SendExpectSuccess(Commands.Slowlog, "RESET".ToUtf8Bytes());

}

public byte[] GetBytes(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectData(Commands.Get, key.ToUtf8Bytes());

}

public byte[] GetSet(string key, byte[] value)

{

if (key == null)

throw new ArgumentNullException("key");

value = value ?? TypeConstants.EmptyByteArray;

if (value.Length > OneGb)

throw new ArgumentException("value exceeds 1G", "value");

return SendExpectData(Commands.GetSet, key.ToUtf8Bytes(), value);

}

public long Exists(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.Exists, key.ToUtf8Bytes());

}

public long Del(string key)

{

return Del(key.ToUtf8Bytes());

}

public long Del(byte[] key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.Del, key);

}

public long Del(params string[] keys)

{

if (keys == null)

throw new ArgumentNullException("keys");

var cmdWithArgs = MergeCommandWithArgs(Commands.Del, keys);

return SendExpectLong(cmdWithArgs);

}

public long Incr(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.Incr, key.ToUtf8Bytes());

}

public long IncrBy(string key, int count)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.IncrBy, key.ToUtf8Bytes(), count.ToUtf8Bytes());

}

public long IncrBy(string key, long count)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.IncrBy, key.ToUtf8Bytes(), count.ToUtf8Bytes());

}

public double IncrByFloat(string key, double incrBy)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectDouble(Commands.IncrByFloat, key.ToUtf8Bytes(), incrBy.ToUtf8Bytes());

}

public long Decr(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.Decr, key.ToUtf8Bytes());

}

public long DecrBy(string key, int count)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.DecrBy, key.ToUtf8Bytes(), count.ToUtf8Bytes());

}

public long Append(string key, byte[] value)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.Append, key.ToUtf8Bytes(), value);

}

public byte[] GetRange(string key, int fromIndex, int toIndex)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectData(Commands.GetRange, key.ToUtf8Bytes(), fromIndex.ToUtf8Bytes(), toIndex.ToUtf8Bytes());

}

public long SetRange(string key, int offset, byte[] value)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.SetRange, key.ToUtf8Bytes(), offset.ToUtf8Bytes(), value);

}

public long GetBit(string key, int offset)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.GetBit, key.ToUtf8Bytes(), offset.ToUtf8Bytes());

}

public long SetBit(string key, int offset, int value)

{

if (key == null)

throw new ArgumentNullException("key");

if (value > 1 || value < 0)

throw new ArgumentException("value is out of range");

return SendExpectLong(Commands.SetBit, key.ToUtf8Bytes(), offset.ToUtf8Bytes(), value.ToUtf8Bytes());

}

public long BitCount(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.BitCount, key.ToUtf8Bytes());

}

public string RandomKey()

{

return SendExpectData(Commands.RandomKey).FromUtf8Bytes();

}

public void Rename(string oldKeyname, string newKeyname)

{

if (oldKeyname == null)

throw new ArgumentNullException("oldKeyname");

if (newKeyname == null)

throw new ArgumentNullException("newKeyname");

SendExpectSuccess(Commands.Rename, oldKeyname.ToUtf8Bytes(), newKeyname.ToUtf8Bytes());

}

public bool RenameNx(string oldKeyname, string newKeyname)

{

if (oldKeyname == null)

throw new ArgumentNullException("oldKeyname");

if (newKeyname == null)

throw new ArgumentNullException("newKeyname");

return SendExpectLong(Commands.RenameNx, oldKeyname.ToUtf8Bytes(), newKeyname.ToUtf8Bytes()) == Success;

}

public bool Expire(string key, int seconds)

{

return Expire(key.ToUtf8Bytes(), seconds);

}

public bool Expire(byte[] key, int seconds)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.Expire, key, seconds.ToUtf8Bytes()) == Success;

}

public bool PExpire(string key, long ttlMs)

{

return PExpire(key.ToUtf8Bytes(), ttlMs);

}

public bool PExpire(byte[] key, long ttlMs)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.PExpire, key, ttlMs.ToUtf8Bytes()) == Success;

}

public bool ExpireAt(string key, long unixTime)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.ExpireAt, key.ToUtf8Bytes(), unixTime.ToUtf8Bytes()) == Success;

}

public bool PExpireAt(string key, long unixTimeMs)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.PExpireAt, key.ToUtf8Bytes(), unixTimeMs.ToUtf8Bytes()) == Success;

}

public long Ttl(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.Ttl, key.ToUtf8Bytes());

}

public long PTtl(string key)

{

if (key == null)

throw new ArgumentNullException("key");

return SendExpectLong(Commands.PTtl, key.ToUtf8Bytes());

}

public void Save()

{

SendExpectSuccess(Commands.Save);

}

public void SaveAsync()

{

BgSave();

}

public void BgSave()

{

SendExpectSuccess(Commands.BgSave);

}

public void Shutdown()

{

SendWithoutRead(Commands.Shutdown);

}

public void ShutdownNoSave()

{

SendWithoutRead(Commands.Shutdown, Commands.NoSave);

}

public void BgRewriteAof()

{

SendExpectSuccess(Commands.BgRewriteAof);

}

public void Quit()

{

SendWithoutRead(Commands.Quit);

}

public void FlushDb()

{

SendExpectSuccess(Commands.FlushDb);

}

public void FlushAll()

{

SendExpectSuccess(Commands.FlushAll);

}

public RedisText Role()

{

return SendExpectComplexResponse(Commands.Role).ToRedisText();

}

public string ClientGetName()

{

return SendExpectString(Commands.Client, Commands.GetName);

}

public void ClientSetName(string name)

{

if (string.IsNullOrEmpty(name))

throw new ArgumentException("Name cannot be null or empty");

if (name.Contains(" "))

throw new ArgumentException("Name cannot contain spaces");

SendExpectSuccess(Commands.Client, Commands.SetName, name.ToUtf8Bytes());

}

public void ClientPause(int timeOutMs)

{

SendExpectSuccess(Commands.Client, Commands.Pause, timeOutMs.ToUtf8Bytes());

}

public byte[] ClientList()

{

return SendExpectData(Commands.Client, Commands.List);

}

public void ClientKill(string clientAddr)

{

SendExpectSuccess(Commands.Client, Commands.Kill, clientAddr.ToUtf8Bytes());

}

public long ClientKill(string addr = null, string id = null, string type = null, string skipMe = null)

{

var cmdWithArgs = new List<byte[]>

{

Commands.Client, Commands.Kill,

};

if (addr != null)

{

cmdWithArgs.Add(Commands.Addr);

cmdWithArgs.Add(addr.ToUtf8Bytes());

}

if (id != null)

{

cmdWithArgs.Add(Commands.Id);

cmdWithArgs.Add(id.ToUtf8Bytes());

}

if (type != null)

{

cmdWithArgs.Add(Commands.Type);

cmdWithArgs.Add(type.ToUtf8Bytes());

}

if (skipMe != null)

{

cmdWithArgs.Add(Commands.SkipMe);

cmdWithArgs.Add(skipMe.ToUtf8Bytes());

}

return SendExpectLong(cmdWithArgs.ToArray());

}

public byte[][] Keys(string pattern)

{

if (pattern == null)

throw new ArgumentNullException("pattern");

return SendExpectMultiData(Commands.Keys, pattern.ToUtf8Bytes());

}

public byte[][] MGet(params byte[][] keys)

{

if (keys == null)

throw new ArgumentNullException("keys");

if (keys.Length == 0)

throw new ArgumentException("keys");

var cmdWithArgs = MergeCommandWithArgs(Commands.MGet, keys);

return SendExpectMultiData(cmdWithArgs);

}

public byte[][] MGet(params string[] keys)

{

if (keys == null)

throw new ArgumentNullException("keys");

if (keys.Length == 0)

throw new ArgumentException("keys");

var cmdWithArgs = MergeCommandWithArgs(Commands.MGet, keys);

return SendExpectMultiData(cmdWithArgs);

}

public void Watch(params string[] keys)

{

if (keys == null)

throw new ArgumentNullException("keys");

if (keys.Length == 0)

throw new ArgumentException("keys");

var cmdWithArgs = MergeCommandWithArgs(Commands.Watch, keys);

SendExpectCode(cmdWithArgs);

}

public void UnWatch()

{

SendExpectCode(Commands.UnWatch);

}

internal void Multi()

{

//make sure socket is connected. Otherwise, fetch of server info will interfere

//with pipeline

AssertConnectedSocket();

SendWithoutRead(Commands.Multi);

}

/// <summary>

/// Requires custom result parsing

/// </summary>

/// <returns>Number of results</returns>

internal void Exec()

{

SendWithoutRead(Commands.Exec);

}