Fix #871 (#872)

* Increase dispatcher queue size

* Allow to customize the size of a dispatcher's queue

Author: Pliner

Source/EasyNetQ/ConnectionConfiguration.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.IO;
 using System.Linq;
@@ -37,6 +38,7 @@ public class ConnectionConfiguration
         public bool UseBackgroundThreads { get; set; }
         public IList<AuthMechanismFactory> AuthMechanisms { get; set; }
         public TimeSpan ConnectIntervalAttempt { get;  set; }
+        public int DispatcherQueueSize { get; set; }
 
         public ConnectionConfiguration()
         {
@@ -51,6 +53,7 @@ public ConnectionConfiguration()
             PersistentMessages = true;
             UseBackgroundThreads = false;
             ConnectIntervalAttempt = TimeSpan.FromSeconds(5);
+            DispatcherQueueSize = 1024;
 
             // prefetchCount determines how many messages will be allowed in the local in-memory queue
             // setting to zero makes this infinite, but risks an out-of-memory exception.

Source/EasyNetQ/ConnectionString/ConnectionStringGrammar.cs

@@ -12,13 +12,14 @@ namespace EasyNetQ.ConnectionString
     public static class ConnectionStringGrammar
     {
         public static Parser<string> Text = Parse.CharExcept(';').Many().Text();
-        public static Parser<ushort> Number = Parse.Number.Select(ushort.Parse);
+        public static Parser<ushort> UShortNumber = Parse.Number.Select(ushort.Parse);
+        public static Parser<int> IntNumber = Parse.Number.Select(int.Parse);
 
         public static Parser<bool> Bool = (Parse.CaseInsensitiveString("true").Or(Parse.CaseInsensitiveString("false"))).Text().Select(x => x.ToLower() == "true");
 
         public static Parser<HostConfiguration> Host =
             from host in Parse.Char(c => c != ':' && c != ';' && c != ',', "host").Many().Text()
-            from port in Parse.Char(':').Then(_ => Number).Or(Parse.Return((ushort)0))
+            from port in Parse.Char(':').Then(_ => UShortNumber).Or(Parse.Return((ushort)0))
             select new HostConfiguration { Host = host, Port = port };
 
         public static Parser<IEnumerable<HostConfiguration>> Hosts = Host.ListDelimitedBy(',');
@@ -31,18 +32,19 @@ from port in Parse.Char(':').Then(_ => Number).Or(Parse.Return((ushort)0))
             // add new connection string parts here
             BuildKeyValueParser("amqp", AMQP, c => c.AMQPConnectionString),
             BuildKeyValueParser("host", Hosts, c => c.Hosts),
-            BuildKeyValueParser("port", Number, c => c.Port),
+            BuildKeyValueParser("port", UShortNumber, c => c.Port),
             BuildKeyValueParser("virtualHost", Text, c => c.VirtualHost),
-            BuildKeyValueParser("requestedHeartbeat", Number, c => c.RequestedHeartbeat),
+            BuildKeyValueParser("requestedHeartbeat", UShortNumber, c => c.RequestedHeartbeat),
             BuildKeyValueParser("username", Text, c => c.UserName),
             BuildKeyValueParser("password", Text, c => c.Password),
-            BuildKeyValueParser("prefetchcount", Number, c => c.PrefetchCount),
-            BuildKeyValueParser("timeout", Number, c => c.Timeout),
+            BuildKeyValueParser("prefetchCount", UShortNumber, c => c.PrefetchCount),
+            BuildKeyValueParser("timeout", UShortNumber, c => c.Timeout),
             BuildKeyValueParser("publisherConfirms", Bool, c => c.PublisherConfirms),
             BuildKeyValueParser("persistentMessages", Bool, c => c.PersistentMessages),
             BuildKeyValueParser("product", Text, c => c.Product),
             BuildKeyValueParser("platform", Text, c => c.Platform),
             BuildKeyValueParser("useBackgroundThreads", Bool, c => c.UseBackgroundThreads),
+            BuildKeyValueParser("dispatcherQueueSize", IntNumber, c => c.DispatcherQueueSize),
             BuildKeyValueParser("name", Text, c => c.Name)
         }.Aggregate((a, b) => a.Or(b));
 

Source/EasyNetQ/Producer/ClientCommandDispatcherSingleton.cs

@@ -9,10 +9,9 @@ namespace EasyNetQ.Producer
 {
     public class ClientCommandDispatcherSingleton : IClientCommandDispatcher
     {
-        private const int queueSize = 1;
         private readonly CancellationTokenSource cancellation = new CancellationTokenSource();
         private readonly IPersistentChannel persistentChannel;
-        private readonly BlockingCollection<Action> queue = new BlockingCollection<Action>(queueSize);
+        private readonly BlockingCollection<Action> queue;
 
         public ClientCommandDispatcherSingleton(
             ConnectionConfiguration configuration,
@@ -23,6 +22,7 @@ public ClientCommandDispatcherSingleton(
             Preconditions.CheckNotNull(connection, "connection");
             Preconditions.CheckNotNull(persistentChannelFactory, "persistentChannelFactory");
 
+            queue = new BlockingCollection<Action>(configuration.DispatcherQueueSize);
             persistentChannel = persistentChannelFactory.CreatePersistentChannel(connection);
 
             StartDispatcherThread(configuration);