#920 - add Lease<T> API; currently only {String|Hash}GetLease[Async]...

#920 - add Lease<T> API; currently only {String|Hash}GetLease[Async] implemented; includes convenience methods for decoding or treating as a MemoryStream

src/StackExchange.Redis/ExtensionMethods.cs

 using System;
 using System.Collections.Generic;
+using System.IO;
 using System.Net.Security;
 using System.Security.Authentication;
 using System.Security.Cryptography.X509Certificates;
+using System.Text;
 
 namespace StackExchange.Redis
 {
@@ -162,5 +164,65 @@ private static void AuthenticateAsClientUsingDefaultProtocols(SslStream ssl, str
         {
             ssl.AuthenticateAsClient(host);
         }
+
+        /// <summary>
+        /// Represent a byte-Lease as a Stream
+        /// </summary>
+        /// <param name="bytes">The lease upon which to base the stream</param>
+        /// <param name="ownsLease">If true, disposing the stream also disposes the lease</param>
+        public static MemoryStream AsStream(this Lease<byte> bytes, bool ownsLease = true)
+        {
+            if (bytes == null) return null; // GIGO
+            var segment = bytes.ArraySegment;
+            if (ownsLease) return new LeaseMemoryStream(segment, bytes);
+            return new MemoryStream(segment.Array, segment.Offset, segment.Count, true, true);
+        }
+
+        /// <summary>
+        /// Decode a byte-Lease as a String, optionally specifying the encoding (UTF-8 if omitted)
+        /// </summary>
+        /// <param name="bytes">The bytes to decode</param>
+        /// <param name="encoding">The encoding to use</param>
+        public static string DecodeString(this Lease<byte> bytes, Encoding encoding = null)
+        {
+            if (bytes == null) return null;
+            if (encoding == null) encoding = Encoding.UTF8;
+            if (bytes.Length == 0) return "";
+            var segment = bytes.ArraySegment;
+            return encoding.GetString(segment.Array, segment.Offset, segment.Count);
+        }
+
+        /// <summary>
+        /// Decode a byte-Lease as a String, optionally specifying the encoding (UTF-8 if omitted)
+        /// </summary>
+        /// <param name="bytes">The bytes to decode</param>
+        /// <param name="encoding">The encoding to use</param>
+        public static Lease<char> DecodeLease(this Lease<byte> bytes, Encoding encoding = null)
+        {
+            if (bytes == null) return null;
+            if (encoding == null) encoding = Encoding.UTF8;
+            if (bytes.Length == 0) return Lease<char>.Empty;
+            var bytesSegment = bytes.ArraySegment;
+            var charCount = encoding.GetCharCount(bytesSegment.Array, bytesSegment.Offset, bytesSegment.Count);
+            var chars = Lease<char>.Create(charCount, false);
+            var charsSegment = chars.ArraySegment;
+            encoding.GetChars(bytesSegment.Array, bytesSegment.Offset, bytesSegment.Count,
+                charsSegment.Array, charsSegment.Offset);
+            return chars;
+        }
+
+        private sealed class LeaseMemoryStream : MemoryStream
+        {
+            private readonly IDisposable _parent;
+            public LeaseMemoryStream(ArraySegment<byte> segment, IDisposable parent)
+                : base(segment.Array, segment.Offset, segment.Count, true, true)
+                => _parent = parent;
+
+            protected override void Dispose(bool disposing)
+            {
+                base.Dispose(disposing);
+                if (disposing) _parent.Dispose();
+            }
+        }
     }
 }

src/StackExchange.Redis/Interfaces/IDatabase.cs

@@ -240,6 +240,16 @@ public interface IDatabase : IRedis, IDatabaseAsync
         /// <remarks>https://redis.io/commands/hget</remarks>
         RedisValue HashGet(RedisKey key, RedisValue hashField, CommandFlags flags = CommandFlags.None);
 
+        /// <summary>
+        /// Returns the value associated with field in the hash stored at key.
+        /// </summary>
+        /// <param name="key">The key of the hash.</param>
+        /// <param name="hashField">The field in the hash to get.</param>
+        /// <param name="flags">The flags to use for this operation.</param>
+        /// <returns>The value associated with field, or nil when field is not present in the hash or key does not exist.</returns>
+        /// <remarks>https://redis.io/commands/hget</remarks>
+        Lease<byte> HashGetLease(RedisKey key, RedisValue hashField, CommandFlags flags = CommandFlags.None);
+
         /// <summary>
         /// Returns the values associated with the specified fields in the hash stored at key.
         /// For every field that does not exist in the hash, a nil value is returned.Because a non-existing keys are treated as empty hashes, running HMGET against a non-existing key will return a list of nil values.
@@ -1810,6 +1820,15 @@ public interface IDatabase : IRedis, IDatabaseAsync
         /// <remarks>https://redis.io/commands/mget</remarks>
         RedisValue[] StringGet(RedisKey[] keys, CommandFlags flags = CommandFlags.None);
 
+        /// <summary>
+        /// Get the value of key. If the key does not exist the special value nil is returned. An error is returned if the value stored at key is not a string, because GET only handles string values.
+        /// </summary>
+        /// <param name="key">The key of the string.</param>
+        /// <param name="flags">The flags to use for this operation.</param>
+        /// <returns>The value of key, or nil when key does not exist.</returns>
+        /// <remarks>https://redis.io/commands/get</remarks>
+        Lease<byte> StringGetLease(RedisKey key, CommandFlags flags = CommandFlags.None);
+
         /// <summary>
         /// Returns the bit value at offset in the string value stored at key.
         /// When offset is beyond the string length, the string is assumed to be a contiguous space with 0 bits.

src/StackExchange.Redis/Interfaces/IDatabaseAsync.cs

@@ -228,6 +228,16 @@ public interface IDatabaseAsync : IRedisAsync
         /// <remarks>https://redis.io/commands/hget</remarks>
         Task<RedisValue> HashGetAsync(RedisKey key, RedisValue hashField, CommandFlags flags = CommandFlags.None);
 
+        /// <summary>
+        /// Returns the value associated with field in the hash stored at key.
+        /// </summary>
+        /// <param name="key">The key of the hash.</param>
+        /// <param name="hashField">The field in the hash to get.</param>
+        /// <param name="flags">The flags to use for this operation.</param>
+        /// <returns>The value associated with field, or nil when field is not present in the hash or key does not exist.</returns>
+        /// <remarks>https://redis.io/commands/hget</remarks>
+        Task<Lease<byte>> HashGetLeaseAsync(RedisKey key, RedisValue hashField, CommandFlags flags = CommandFlags.None);
+
         /// <summary>
         /// Returns the values associated with the specified fields in the hash stored at key.
         /// For every field that does not exist in the hash, a nil value is returned.Because a non-existing keys are treated as empty hashes, running HMGET against a non-existing key will return a list of nil values.
@@ -1721,6 +1731,15 @@ public interface IDatabaseAsync : IRedisAsync
         /// <remarks>https://redis.io/commands/mget</remarks>
         Task<RedisValue[]> StringGetAsync(RedisKey[] keys, CommandFlags flags = CommandFlags.None);
 
+        /// <summary>
+        /// Get the value of key. If the key does not exist the special value nil is returned. An error is returned if the value stored at key is not a string, because GET only handles string values.
+        /// </summary>
+        /// <param name="key">The key of the string.</param>
+        /// <param name="flags">The flags to use for this operation.</param>
+        /// <returns>The value of key, or nil when key does not exist.</returns>
+        /// <remarks>https://redis.io/commands/get</remarks>
+        Task<Lease<byte>> StringGetLeaseAsync(RedisKey key, CommandFlags flags = CommandFlags.None);
+
         /// <summary>
         /// Returns the bit value at offset in the string value stored at key.
         /// When offset is beyond the string length, the string is assumed to be a contiguous space with 0 bits.

src/StackExchange.Redis/KeyspaceIsolation/DatabaseWrapper.cs

@@ -122,6 +122,11 @@ public RedisValue HashGet(RedisKey key, RedisValue hashField, CommandFlags flags
             return Inner.HashGet(ToInner(key), hashField, flags);
         }
 
+        public Lease<byte> HashGetLease(RedisKey key, RedisValue hashField, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.HashGetLease(ToInner(key), hashField, flags);
+        }
+
         public double HashIncrement(RedisKey key, RedisValue hashField, double value, CommandFlags flags = CommandFlags.None)
         {
             return Inner.HashIncrement(ToInner(key), hashField, value, flags);
@@ -766,6 +771,11 @@ public RedisValue StringGet(RedisKey key, CommandFlags flags = CommandFlags.None
             return Inner.StringGet(ToInner(key), flags);
         }
 
+        public Lease<byte>StringGetLease(RedisKey key, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.StringGetLease(ToInner(key), flags);
+        }
+
         public bool StringGetBit(RedisKey key, long offset, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StringGetBit(ToInner(key), offset, flags);

src/StackExchange.Redis/KeyspaceIsolation/WrapperBase.cs

@@ -98,6 +98,11 @@ public Task<RedisValue> HashGetAsync(RedisKey key, RedisValue hashField, Command
             return Inner.HashGetAsync(ToInner(key), hashField, flags);
         }
 
+        public Task<Lease<byte>> HashGetLeaseAsync(RedisKey key, RedisValue hashField, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.HashGetLeaseAsync(ToInner(key), hashField, flags);
+        }
+
         public Task<double> HashIncrementAsync(RedisKey key, RedisValue hashField, double value, CommandFlags flags = CommandFlags.None)
         {
             return Inner.HashIncrementAsync(ToInner(key), hashField, value, flags);
@@ -746,6 +751,11 @@ public Task<RedisValue> StringGetAsync(RedisKey key, CommandFlags flags = Comman
             return Inner.StringGetAsync(ToInner(key), flags);
         }
 
+        public Task<Lease<byte>> StringGetLeaseAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
+        {
+            return Inner.StringGetLeaseAsync(ToInner(key), flags);
+        }
+
         public Task<bool> StringGetBitAsync(RedisKey key, long offset, CommandFlags flags = CommandFlags.None)
         {
             return Inner.StringGetBitAsync(ToInner(key), offset, flags);

src/StackExchange.Redis/Lease.cs

+using System;
+using System.Buffers;
+using System.Runtime.CompilerServices;
+using System.Threading;
+
+namespace StackExchange.Redis
+{
+    /// <summary>
+    /// A sized region of contiguous memory backed by a memory pool; disposing the lease returns the memory to the pool
+    /// </summary>
+    /// <typeparam name="T">The type of data being leased</typeparam>
+    public sealed class Lease<T> : IMemoryOwner<T>
+    {
+        /// <summary>
+        /// A lease of length zero
+        /// </summary>
+        public static Lease<T> Empty { get; } = new Lease<T>(System.Array.Empty<T>(), 0);
+
+        private T[] _arr;
+
+        /// <summary>
+        /// The length of the lease
+        /// </summary>
+        public int Length { get; }
+
+        /// <summary>
+        /// Create a new lease
+        /// </summary>
+        /// <param name="length">The size required</param>
+        /// <param name="clear">Whether to erase the memory</param>
+        public static Lease<T> Create(int length, bool clear = true)
+        {
+            if (length == 0) return Empty;
+            var arr = ArrayPool<T>.Shared.Rent(length);
+            if (clear) System.Array.Clear(arr, 0, length);
+            return new Lease<T>(arr, length);
+        }
+
+        private Lease(T[] arr, int length)
+        {
+            _arr = arr;
+            Length = length;
+        }
+
+        /// <summary>
+        /// Release all resources owned by the lease
+        /// </summary>
+        public void Dispose()
+        {
+            if (Length != 0)
+            {
+                var arr = Interlocked.Exchange(ref _arr, null);
+                if (arr != null) ArrayPool<T>.Shared.Return(arr);
+            }
+        }
+        [MethodImpl(MethodImplOptions.NoInlining)]
+        private static T[] ThrowDisposed() => throw new ObjectDisposedException(nameof(Lease<T>));
+        private T[] Array
+        {
+            [MethodImpl(MethodImplOptions.AggressiveInlining)]
+            get => _arr ?? ThrowDisposed();
+        }
+
+        /// <summary>
+        /// The data as a Memory
+        /// </summary>
+        public Memory<T> Memory => new Memory<T>(Array, 0, Length);
+
+        /// <summary>
+        /// The data as a Span
+        /// </summary>
+        public Span<T> Span => new Span<T>(Array, 0, Length);
+
+        /// <summary>
+        /// The data as an ArraySegment
+        /// </summary>
+        public ArraySegment<T> ArraySegment => new ArraySegment<T>(Array, 0, Length);
+    }
+}

src/StackExchange.Redis/RawResult.cs

@@ -180,6 +180,21 @@ internal RedisValue AsRedisValue()
             throw new InvalidCastException("Cannot convert to RedisValue: " + Type);
         }
 
+        internal Lease<byte> AsLease()
+        {
+            if (IsNull) return null;
+            switch (Type)
+            {
+                case ResultType.SimpleString:
+                case ResultType.BulkString:
+                    var payload = Payload;
+                    var lease = Lease<byte>.Create(checked((int)payload.Length), false);
+                    payload.CopyTo(lease.Span);
+                    return lease;
+            }
+            throw new InvalidCastException("Cannot convert to Lease: " + Type);
+        }
+
         internal void Recycle(int limit = -1)
         {
             var arr = _itemsOversized;

src/StackExchange.Redis/RedisDatabase.cs

@@ -309,6 +309,12 @@ public RedisValue HashGet(RedisKey key, RedisValue hashField, CommandFlags flags
             return ExecuteSync(msg, ResultProcessor.RedisValue);
         }
 
+        public Lease<byte> HashGetLease(RedisKey key, RedisValue hashField, CommandFlags flags = CommandFlags.None)
+        {
+            var msg = Message.Create(Database, flags, RedisCommand.HGET, key, hashField);
+            return ExecuteSync(msg, ResultProcessor.Lease);
+        }
+
         public RedisValue[] HashGet(RedisKey key, RedisValue[] hashFields, CommandFlags flags = CommandFlags.None)
         {
             if (hashFields == null) throw new ArgumentNullException(nameof(hashFields));
@@ -335,6 +341,12 @@ public Task<RedisValue> HashGetAsync(RedisKey key, RedisValue hashField, Command
             return ExecuteAsync(msg, ResultProcessor.RedisValue);
         }
 
+        public Task<Lease<byte>> HashGetLeaseAsync(RedisKey key, RedisValue hashField, CommandFlags flags = CommandFlags.None)
+        {
+            var msg = Message.Create(Database, flags, RedisCommand.HGET, key, hashField);
+            return ExecuteAsync(msg, ResultProcessor.Lease);
+        }
+
         public Task<RedisValue[]> HashGetAsync(RedisKey key, RedisValue[] hashFields, CommandFlags flags = CommandFlags.None)
         {
             if (hashFields == null) throw new ArgumentNullException(nameof(hashFields));
@@ -2247,12 +2259,24 @@ public RedisValue[] StringGet(RedisKey[] keys, CommandFlags flags = CommandFlags
             return ExecuteSync(msg, ResultProcessor.RedisValueArray);
         }
 
+        public Lease<byte> StringGetLease(RedisKey key, CommandFlags flags = CommandFlags.None)
+        {
+            var msg = Message.Create(Database, flags, RedisCommand.GET, key);
+            return ExecuteSync(msg, ResultProcessor.Lease);
+        }
+
         public Task<RedisValue> StringGetAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
         {
             var msg = Message.Create(Database, flags, RedisCommand.GET, key);
             return ExecuteAsync(msg, ResultProcessor.RedisValue);
         }
 
+        public Task<Lease<byte>> StringGetLeaseAsync(RedisKey key, CommandFlags flags = CommandFlags.None)
+        {
+            var msg = Message.Create(Database, flags, RedisCommand.GET, key);
+            return ExecuteAsync(msg, ResultProcessor.Lease);
+        }
+
         public Task<RedisValue[]> StringGetAsync(RedisKey[] keys, CommandFlags flags = CommandFlags.None)
         {
             if (keys == null) throw new ArgumentNullException(nameof(keys));

src/StackExchange.Redis/ResultProcessor.cs

@@ -68,6 +68,9 @@ public static readonly MultiStreamProcessor
         public static readonly ResultProcessor<RedisValue>
             RedisValue = new RedisValueProcessor();
 
+        public static readonly ResultProcessor<Lease<byte>>
+            Lease = new LeaseProcessor();
+
         public static readonly ResultProcessor<RedisValue[]>
             RedisValueArray = new RedisValueArrayProcessor();
 
@@ -1303,6 +1306,22 @@ protected override bool SetResultCore(PhysicalConnection connection, Message mes
             }
         }
 
+        private sealed class LeaseProcessor : ResultProcessor<Lease<byte>>
+        {
+            protected override bool SetResultCore(PhysicalConnection connection, Message message, RawResult result)
+            {
+                switch (result.Type)
+                {
+                    case ResultType.Integer:
+                    case ResultType.SimpleString:
+                    case ResultType.BulkString:
+                        SetResult(message, result.AsLease());
+                        return true;
+                }
+                return false;
+            }
+        }
+
         private class ScriptResultProcessor : ResultProcessor<RedisResult>
         {
             public override bool SetResult(PhysicalConnection connection, Message message, RawResult result)

tests/StackExchange.Redis.Tests/Strings.cs

-using System.Linq;
+using System.IO;
+using System.Linq;
 using System.Text;
 using System.Threading.Tasks;
 using Xunit;
@@ -65,6 +66,45 @@ public async Task Set()
             }
         }
 
+        [Fact]
+        public async Task GetLease()
+        {
+            using (var muxer = Create())
+            {
+                var conn = muxer.GetDatabase();
+                var key = Me();
+                conn.KeyDelete(key, CommandFlags.FireAndForget);
+
+                conn.StringSet(key, "abc", flags: CommandFlags.FireAndForget);
+                using (var v1 = await conn.StringGetLeaseAsync(key).ConfigureAwait(false))
+                {
+                    string s = v1.DecodeString();
+                    Assert.Equal("abc", s);
+                }
+            }
+        }
+
+        [Fact]
+        public async Task GetLeaseAsStream()
+        {
+            using (var muxer = Create())
+            {
+                var conn = muxer.GetDatabase();
+                var key = Me();
+                conn.KeyDelete(key, CommandFlags.FireAndForget);
+
+                conn.StringSet(key, "abc", flags: CommandFlags.FireAndForget);
+                using (var v1 = (await conn.StringGetLeaseAsync(key).ConfigureAwait(false)).AsStream())
+                {
+                    using (var sr = new StreamReader(v1))
+                    {
+                        string s = sr.ReadToEnd();
+                        Assert.Equal("abc", s);
+                    }
+                }
+            }
+        }
+
         [Fact]
         public async Task SetNotExists()
         {