Merge pull request #306 from rabbitmq/rabbitmq-server-143-stable

Backport configurable credit flow to stable

4 files changed, 66 insertions, 2 deletions

diff --git a/ebin/rabbit_app.in b/ebin/rabbit_app.in
index 97e72c0b65..ac60ed6825 100644
--- a/ebin/rabbit_app.in
+++ b/ebin/rabbit_app.in
@@ -92,5 +92,7 @@
          %% writing to the queue index. See the setting
          %% queue_index_embed_msgs_below above.
          {msg_store_credit_disc_bound, {2000, 500}},
-         {msg_store_io_batch_size, 2048}
+         {msg_store_io_batch_size, 2048},
+         %% see rabbitmq-server#143
+         {credit_flow_default_credit, {200, 50}}
         ]}]}.
diff --git a/src/credit_flow.erl b/src/credit_flow.erl
index e7f13a66dc..1271819c5d 100644
--- a/src/credit_flow.erl
+++ b/src/credit_flow.erl
@@ -28,7 +28,19 @@
 %% is itself blocked - thus the only processes that need to check
 %% blocked/0 are ones that read from network sockets.
 
--define(DEFAULT_CREDIT, {200, 50}).
+-define(DEFAULT_INITIAL_CREDIT, 200).
+-define(DEFAULT_MORE_CREDIT_AFTER, 50).
+
+-define(DEFAULT_CREDIT,
+        case get(credit_flow_default_credit) of
+            undefined ->
+                Val = rabbit_misc:get_env(rabbit, credit_flow_default_credit,
+                                           {?DEFAULT_INITIAL_CREDIT,
+                                            ?DEFAULT_MORE_CREDIT_AFTER}),
+                put(credit_flow_default_credit, Val),
+                Val;
+            Val       -> Val
+        end).
 
 -export([send/1, send/2, ack/1, ack/2, handle_bump_msg/1, blocked/0, state/0]).
 -export([peer_down/1]).
diff --git a/test/src/credit_flow_test.erl b/test/src/credit_flow_test.erl
new file mode 100644
index 0000000000..148c9024f1
--- /dev/null
+++ b/test/src/credit_flow_test.erl
@@ -0,0 +1,49 @@
+%% The contents of this file are subject to the Mozilla Public License
+%% Version 1.1 (the "License"); you may not use this file except in
+%% compliance with the License. You may obtain a copy of the License at
+%% http://www.mozilla.org/MPL/
+%%
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+%% License for the specific language governing rights and limitations
+%% under the License.
+%%
+%% The Original Code is RabbitMQ.
+%%
+%% The Initial Developer of the Original Code is GoPivotal, Inc.
+%% Copyright (c) 2011-2014 GoPivotal, Inc.  All rights reserved.
+%%
+
+-module(credit_flow_test).
+
+-export([test_credit_flow_settings/0]).
+
+test_credit_flow_settings() ->
+    %% default values
+    passed = test_proc(200, 50),
+
+    application:set_env(rabbit, credit_flow_default_credit, {100, 20}),
+    passed = test_proc(100, 20),
+
+    application:unset_env(rabbit, credit_flow_default_credit),
+    % back to defaults
+    passed = test_proc(200, 50),
+    passed.
+
+test_proc(InitialCredit, MoreCreditAfter) ->
+    Pid = spawn(fun dummy/0),
+    Pid ! {credit, self()},
+    {InitialCredit, MoreCreditAfter} =
+        receive
+            {credit, Val} -> Val
+        end,
+    passed.
+
+dummy() ->
+    credit_flow:send(self()),
+    receive
+        {credit, From} ->
+            From ! {credit, get(credit_flow_default_credit)};
+        _      ->
+            dummy()
+    end.
diff --git a/test/src/rabbit_tests.erl b/test/src/rabbit_tests.erl
index 39a276f102..6f0f5eb181 100644
--- a/test/src/rabbit_tests.erl
+++ b/test/src/rabbit_tests.erl
@@ -88,6 +88,7 @@ all_tests0() ->
     passed = test_configurable_server_properties(),
     passed = vm_memory_monitor_tests:all_tests(),
     passed = on_disk_store_tunable_parameter_validation_test:test_msg_store_parameter_validation(),
+    passed = credit_flow_test:test_credit_flow_settings(),
     passed.