Refactor all underlying SendExpect commands to re-use same SendReceive* API's

Author: mythz

src/ServiceStack.Redis/RedisNativeClient_Utils.cs

@@ -385,6 +385,8 @@ private void PushCurrentBuffer()
             currentBufferIndex = 0;
         }
 
+        public Action OnBeforeFlush { get; set; }
+
         public bool FlushSendBuffer()
         {
             try
@@ -394,6 +396,9 @@ public bool FlushSendBuffer()
 
                 if (cmdBuffer.Count > 0)
                 {
+                    if (OnBeforeFlush != null)
+                        OnBeforeFlush();
+
                     if (!Env.IsMono && sslStream == null)
                     {
                         socket.Send(cmdBuffer); //Optimized for Windows
@@ -462,132 +467,83 @@ private int SafeReadByte()
 	        }
         }
 
-        protected void SendExpectSuccess(params byte[][] cmdWithBinaryArgs)
+        protected T SendReceive<T>(byte[][] cmdWithBinaryArgs, Func<T> fn, Action<Func<T>> completePipelineFn = null)
         {
             if (!SendCommand(cmdWithBinaryArgs))
                 throw CreateConnectionError();
 
             if (Pipeline != null)
             {
-                Pipeline.CompleteVoidQueuedCommand(ExpectSuccess);
-                return;
-            }
-            ExpectSuccess();
-        }
-
-        protected long SendExpectLong(params byte[][] cmdWithBinaryArgs)
-        {
-            if (!SendCommand(cmdWithBinaryArgs))
-                throw CreateConnectionError();
+                if (completePipelineFn == null)
+                    throw new NotSupportedException("Pipeline is not supported.");
 
-            if (Pipeline != null)
-            {
-                Pipeline.CompleteLongQueuedCommand(ReadLong);
-                return default(long);
+                completePipelineFn(fn);
+                return default(T);
             }
-            return ReadLong();
+            return fn();
         }
 
-        protected byte[] SendExpectData(params byte[][] cmdWithBinaryArgs)
+        protected void SendReceiveVoid(byte[][] cmdWithBinaryArgs, Action fn, Action<Action> completePipelineFn)
         {
             if (!SendCommand(cmdWithBinaryArgs))
                 throw CreateConnectionError();
 
             if (Pipeline != null)
             {
-                Pipeline.CompleteBytesQueuedCommand(ReadData);
-                return null;
+                if (completePipelineFn == null)
+                    throw new NotSupportedException("Pipeline is not supported.");
+
+                completePipelineFn(fn);
+                return;
             }
-            return ReadData();
+            fn();
         }
 
-        protected string SendExpectString(params byte[][] cmdWithBinaryArgs)
+        protected void SendExpectSuccess(params byte[][] cmdWithBinaryArgs)
         {
-            var bytes = SendExpectData(cmdWithBinaryArgs);
-            return bytes.FromUtf8Bytes();
+            SendReceiveVoid(cmdWithBinaryArgs, ExpectSuccess, Pipeline != null ? Pipeline.CompleteVoidQueuedCommand : (Action<Action>) null);
         }
 
-        protected double SendExpectDouble(params byte[][] cmdWithBinaryArgs)
+        protected long SendExpectLong(params byte[][] cmdWithBinaryArgs)
         {
-            if (!SendCommand(cmdWithBinaryArgs))
-                throw CreateConnectionError();
-
-            if (Pipeline != null)
-            {
-                Pipeline.CompleteDoubleQueuedCommand(ReadDouble);
-                return Double.NaN;
-            }
-
-            return ReadDouble();
+            return SendReceive(cmdWithBinaryArgs, ReadLong, Pipeline != null ? Pipeline.CompleteLongQueuedCommand : (Action<Func<long>>) null);
         }
 
-        public double ReadDouble()
+        protected byte[] SendExpectData(params byte[][] cmdWithBinaryArgs)
         {
-            var bytes = ReadData();
-            return (bytes == null) ? double.NaN : ParseDouble(bytes);
+            return SendReceive(cmdWithBinaryArgs, ReadData, Pipeline != null ? Pipeline.CompleteBytesQueuedCommand : (Action<Func<byte[]>>)null);
         }
 
-        public static double ParseDouble(byte[] doubleBytes)
+        protected double SendExpectDouble(params byte[][] cmdWithBinaryArgs)
         {
-            var doubleString = Encoding.UTF8.GetString(doubleBytes);
-
-            double d;
-            double.TryParse(doubleString, NumberStyles.Any, CultureInfo.InvariantCulture.NumberFormat, out d);
-
-            return d;
+            return SendReceive(cmdWithBinaryArgs, ReadDouble, Pipeline != null ? Pipeline.CompleteDoubleQueuedCommand : (Action<Func<double>>)null);
         }
 
         protected string SendExpectCode(params byte[][] cmdWithBinaryArgs)
         {
-            if (!SendCommand(cmdWithBinaryArgs))
-                throw CreateConnectionError();
-
-            if (Pipeline != null)
-            {
-                Pipeline.CompleteStringQueuedCommand(ExpectCode);
-                return null;
-            }
-
-            return ExpectCode();
+            return SendReceive(cmdWithBinaryArgs, ExpectCode, Pipeline != null ? Pipeline.CompleteStringQueuedCommand : (Action<Func<string>>)null);
         }
 
         protected byte[][] SendExpectMultiData(params byte[][] cmdWithBinaryArgs)
         {
-            if (!SendCommand(cmdWithBinaryArgs))
-                throw CreateConnectionError();
-
-            if (Pipeline != null)
-            {
-                Pipeline.CompleteMultiBytesQueuedCommand(ReadMultiData);
-                return new byte[0][];
-            }
-            return ReadMultiData();
+            return SendReceive(cmdWithBinaryArgs, ReadMultiData, Pipeline != null ? Pipeline.CompleteMultiBytesQueuedCommand : (Action<Func<byte[][]>>)null)
+                ?? new byte[0][];
         }
 
         protected object[] SendExpectDeeplyNestedMultiData(params byte[][] cmdWithBinaryArgs)
         {
-            if (!SendCommand(cmdWithBinaryArgs))
-                throw CreateConnectionError();
-
-            if (Pipeline != null)
-            {
-                throw new NotSupportedException("Pipeline is not supported.");
-            }
-
-            return ReadDeeplyNestedMultiData();
+            return SendReceive(cmdWithBinaryArgs, ReadDeeplyNestedMultiData);
         }
 
         protected RedisData SendExpectComplexResponse(params byte[][] cmdWithBinaryArgs)
         {
-            if (!SendCommand(cmdWithBinaryArgs))
-                throw CreateConnectionError();
-
-            if (Pipeline != null)
-            {
-                throw new NotSupportedException("Pipeline is not supported.");
-            }
+            return SendReceive(cmdWithBinaryArgs, ReadComplexResponse);
+        }
 
-            return ReadComplexResponse();
+        protected string SendExpectString(params byte[][] cmdWithBinaryArgs)
+        {
+            var bytes = SendExpectData(cmdWithBinaryArgs);
+            return bytes.FromUtf8Bytes();
         }
 
         protected void Log(string fmt, params object[] args)
@@ -706,6 +662,22 @@ public long ReadLong()
             throw CreateResponseError("Unknown reply on integer response: " + c + s);
         }
 
+        public double ReadDouble()
+        {
+            var bytes = ReadData();
+            return (bytes == null) ? double.NaN : ParseDouble(bytes);
+        }
+
+        public static double ParseDouble(byte[] doubleBytes)
+        {
+            var doubleString = Encoding.UTF8.GetString(doubleBytes);
+
+            double d;
+            double.TryParse(doubleString, NumberStyles.Any, CultureInfo.InvariantCulture.NumberFormat, out d);
+
+            return d;
+        }
+
         private byte[] ReadData()
         {
             var r = ReadLine();