Support PreserveAsyncOrder in connection string (#792)

Looks fine and valid to me, thanks! Makes perfect sense.

StackExchange.Redis.Tests/Issues/Issue791.cs

+using Xunit;
+using Xunit.Abstractions;
+
+namespace StackExchange.Redis.Tests.Issues
+{
+    public class Issue791 : TestBase
+    {
+        public Issue791(ITestOutputHelper output) : base(output) { }
+
+        
+        [Fact]
+        public void PreserveAsyncOrderImplicitValue_ParsedFromConnectionString()
+        {
+            var options = ConfigurationOptions.Parse("preserveAsyncOrder=true");
+            Assert.True(options.PreserveAsyncOrder);
+            Assert.Equal("preserveAsyncOrder=True", options.ToString());
+
+            options = ConfigurationOptions.Parse("preserveAsyncOrder=false");
+            Assert.False(options.PreserveAsyncOrder);
+            Assert.Equal("preserveAsyncOrder=False", options.ToString());
+        }
+
+
+        [Fact]
+        public void DefaultValue_IsTrue()
+        {
+            var options = ConfigurationOptions.Parse("ssl=true");
+            Assert.True(options.PreserveAsyncOrder);            
+        }
+
+        
+        [Fact]
+        public void PreserveAsyncOrder_SetConnectionMultiplexerProperty()
+        {
+            var multiplexer = ConnectionMultiplexer.Connect(TestConfig.Current.MasterServer + ":6500,preserveAsyncOrder=false");
+            Assert.False(multiplexer.PreserveAsyncOrder);
+        }
+    }
+}

StackExchange.Redis/StackExchange/Redis/ConfigurationOptions.cs

@@ -91,6 +91,7 @@ internal static void Unknown(string key)
                         ChannelPrefix = "channelPrefix", Proxy = "proxy", ConnectRetry = "connectRetry",
                         ConfigCheckSeconds = "configCheckSeconds", ResponseTimeout = "responseTimeout", DefaultDatabase = "defaultDatabase";
             internal const string SslProtocols = "sslProtocols";
+            internal const string PreserveAsyncOrder = "preserveAsyncOrder";
 
             private static readonly Dictionary<string, string> normalizedOptions = new[]
             {
@@ -101,7 +102,7 @@ internal static void Unknown(string key)
                 ConfigChannel, AbortOnConnectFail, ResolveDns,
                 ChannelPrefix, Proxy, ConnectRetry,
                 ConfigCheckSeconds, DefaultDatabase,
-                SslProtocols,
+                SslProtocols, PreserveAsyncOrder
             }.ToDictionary(x => x, StringComparer.OrdinalIgnoreCase);
 
             public static string TryNormalize(string value)
@@ -118,7 +119,7 @@ public static string TryNormalize(string value)
 
         private readonly EndPointCollection endpoints = new EndPointCollection();
 
-        private bool? allowAdmin, abortOnConnectFail, highPrioritySocketThreads, resolveDns, ssl;
+        private bool? allowAdmin, abortOnConnectFail, highPrioritySocketThreads, resolveDns, ssl, preserveAsyncOrder;
 
         private string clientName, serviceName, password, tieBreaker, sslHost, configChannel;
 
@@ -312,6 +313,11 @@ public CommandMap CommandMap
         /// </summary>
         public int? DefaultDatabase { get { return defaultDatabase; } set { defaultDatabase = value; } }
         
+        /// <summary>
+        /// Specifies whether asynchronous operations should be invoked in a way that guarantees their original delivery order
+        /// </summary>
+        public bool PreserveAsyncOrder { get { return preserveAsyncOrder.GetValueOrDefault(true); } set { preserveAsyncOrder = value; } }
+
         internal LocalCertificateSelectionCallback CertificateSelectionCallback { get { return CertificateSelection; } private set { CertificateSelection = value; } }
 
         // these just rip out the underlying handlers, bypassing the event accessors - needed when creating the SSL stream
@@ -379,6 +385,7 @@ public ConfigurationOptions Clone()
                 responseTimeout = responseTimeout,
                 defaultDatabase = defaultDatabase,
                 ReconnectRetryPolicy = reconnectRetryPolicy,
+                preserveAsyncOrder = preserveAsyncOrder,
 #if !CORE_CLR
                 SslProtocols = SslProtocols,
 #endif
@@ -440,6 +447,7 @@ public string ToString(bool includePassword)
             Append(sb, OptionKeys.ConfigCheckSeconds, configCheckSeconds);
             Append(sb, OptionKeys.ResponseTimeout, responseTimeout);
             Append(sb, OptionKeys.DefaultDatabase, defaultDatabase);
+            Append(sb, OptionKeys.PreserveAsyncOrder, preserveAsyncOrder);
             commandMap?.AppendDeltas(sb);
             return sb.ToString();
         }
@@ -527,7 +535,7 @@ void Clear()
         {
             clientName = serviceName = password = tieBreaker = sslHost = configChannel = null;
             keepAlive = syncTimeout = connectTimeout = writeBuffer = connectRetry = configCheckSeconds = defaultDatabase = null;
-            allowAdmin = abortOnConnectFail = highPrioritySocketThreads = resolveDns = ssl = null;
+            allowAdmin = abortOnConnectFail = highPrioritySocketThreads = resolveDns = ssl = preserveAsyncOrder = null;
             defaultVersion = null;
             endpoints.Clear();
             commandMap = null;
@@ -640,6 +648,9 @@ private void DoParse(string configuration, bool ignoreUnknown)
                         case OptionKeys.DefaultDatabase:
                             defaultDatabase = OptionKeys.ParseInt32(key, value);
                             break;
+                        case OptionKeys.PreserveAsyncOrder:
+                            PreserveAsyncOrder = OptionKeys.ParseBoolean(key, value);
+                            break;
 #if !CORE_CLR
                         case OptionKeys.SslProtocols:
                             SslProtocols = OptionKeys.ParseSslProtocols(key, value);

StackExchange.Redis/StackExchange/Redis/ConnectionMultiplexer.cs

@@ -936,7 +936,7 @@ private ConnectionMultiplexer(ConfigurationOptions configuration)
                 map.AssertAvailable(RedisCommand.EXISTS);
             }
 
-            PreserveAsyncOrder = true; // safest default
+            PreserveAsyncOrder = configuration.PreserveAsyncOrder;
             timeoutMilliseconds = configuration.SyncTimeout;
 
             OnCreateReaderWriter(configuration);