Merge branch 'master' into rabbitmq-server-1873-binding-recovery

Author: michaelklishin

docs/rabbitmq-service.8

@@ -115,7 +115,7 @@ Defaults to
 for 64-bit environments).
 This is the installation location of the Erlang service manager.
 .It Ev RABBITMQ_CONSOLE_LOG
-Set this varable to
+Set this variable to
 .Sy new or
 .Sy reuse
 to have the console output from the server redirected to a file named

docs/rabbitmq.conf.example

@@ -804,7 +804,7 @@
 ## See http://rabbitmq.com/logging.html and https://github.com/erlang-lager/lager for details.
 ##
 
-## Log direcrory, taken from the RABBITMQ_LOG_BASE env variable by default.
+## Log directory, taken from the RABBITMQ_LOG_BASE env variable by default.
 ##
 # log.dir = /var/log/rabbitmq
 

docs/rabbitmq.config.example

@@ -339,7 +339,7 @@
 
    %% Enables flow control between queue mirrors.
    %% Disabling this can be dangerous and is not recommended.
-   %% When flow control is disablied, queue masters can outpace mirrors and not allow mirrors to catch up.
+   %% When flow control is disabled, queue masters can outpace mirrors and not allow mirrors to catch up.
    %% Mirrors will end up using increasingly more RAM, eventually triggering a memory alarm.
    %%
    %% {mirroring_flow_control, true},
@@ -909,7 +909,7 @@
    %%   {rabbit_channel_lager_event, [{handlers, [
    %%                                   {lager_forwarder_backend,
    %%                                    [lager_event, info]}]}]},
-   %%   {rabbit_conection_lager_event, [{handlers, [
+   %%   {rabbit_connection_lager_event, [{handlers, [
    %%                                     {lager_forwarder_backend,
    %%                                      [lager_event, info]}]}]},
    %%   {rabbit_mirroring_lager_event, [{handlers, [

docs/rabbitmqctl.8

@@ -156,7 +156,7 @@ To use this precompiled files, you should set
 .Ev RABBITMQ_SERVER_CODE_PATH
 environment variable to directory specified in
 .Cm hipe_compile
-invokation.
+invocation.
 .Pp
 For example, to HiPE-compile modules and store them to
 .Pa /tmp/rabbit-hipe/ebin
@@ -255,7 +255,7 @@ For example, to instruct the RabbitMQ node to terminate:
 .Dl rabbitmqctl stop
 .\" ------------------------------------
 .It Cm stop_app
-Stops the RabbitMQ application, leaving the runtme (Erlang VM) running.
+Stops the RabbitMQ application, leaving the runtime (Erlang VM) running.
 .Pp
 This command is typically run prior to performing other management
 actions that require the RabbitMQ application to be stopped, e.g.\&

erlang.mk

@@ -596,7 +596,7 @@ pkg_check_node_commit = master
 
 PACKAGES += chronos
 pkg_chronos_name = chronos
-pkg_chronos_description = Timer module for Erlang that makes it easy to abstact time out of the tests.
+pkg_chronos_description = Timer module for Erlang that makes it easy to abstract time out of the tests.
 pkg_chronos_homepage = https://github.com/lehoff/chronos
 pkg_chronos_fetch = git
 pkg_chronos_repo = https://github.com/lehoff/chronos
@@ -1340,7 +1340,7 @@ pkg_erlang_cep_commit = master
 
 PACKAGES += erlang_js
 pkg_erlang_js_name = erlang_js
-pkg_erlang_js_description = A linked-in driver for Erlang to Mozilla's Spidermonkey Javascript runtime.
+pkg_erlang_js_description = A linked-in driver for Erlang to Mozilla's Spidermonkey JavaScript runtime.
 pkg_erlang_js_homepage = https://github.com/basho/erlang_js
 pkg_erlang_js_fetch = git
 pkg_erlang_js_repo = https://github.com/basho/erlang_js
@@ -4132,7 +4132,7 @@ pkg_yaws_commit = master
 
 PACKAGES += zab_engine
 pkg_zab_engine_name = zab_engine
-pkg_zab_engine_description = zab propotocol implement by erlang
+pkg_zab_engine_description = zab protocol implement by erlang
 pkg_zab_engine_homepage = https://github.com/xinmingyao/zab_engine
 pkg_zab_engine_fetch = git
 pkg_zab_engine_repo = https://github.com/xinmingyao/zab_engine

priv/schema/rabbit.schema

@@ -903,7 +903,7 @@ fun(Conf) ->
     end
 end}.
 
-%% Classic config-driven peer discuvery backend.
+%% Classic config-driven peer discovery backend.
 %%
 %% Make clustering happen *automatically* at startup - only applied
 %% to nodes that have just been reset or started for the first time.
@@ -1530,7 +1530,7 @@ fun(Size) when is_integer(Size) ->
     Size > 0 andalso Size < 536870912
 end}.
 
-{validator, "less_than_1", "Flooat is not beetween 0 and 1",
+{validator, "less_than_1", "Float is not between 0 and 1",
 fun(Float) when is_float(Float) ->
     Float > 0 andalso Float < 1
 end}.

rabbitmq-components.mk

@@ -226,7 +226,7 @@ export base_rmq_ref
 # If cloning from this computed location fails, we fallback to RabbitMQ
 # upstream which is GitHub.
 
-# Maccro to transform eg. "rabbit_common" to "rabbitmq-common".
+# Macro to transform eg. "rabbit_common" to "rabbitmq-common".
 rmq_cmp_repo_name = $(word 2,$(dep_$(1)))
 
 # Upstream URL for the current project.

scripts/rabbitmq-server

@@ -219,7 +219,7 @@ fi
 # The defaults are meant to reduce RabbitMQ's memory usage and help
 # it reclaim memory at the cost of a slight decrease in performance
 # (due to an increase in memory operations). These defaults can be
-# overriden using the RABBITMQ_SERVER_ERL_ARGS variable.
+# overridden using the RABBITMQ_SERVER_ERL_ARGS variable.
 RABBITMQ_DEFAULT_ALLOC_ARGS="+MBas ageffcbf +MHas ageffcbf +MBlmbcs 512 +MHlmbcs 512 +MMmcs 30"
 
 ${ERL_DIR}erl ${RABBITMQ_DEFAULT_ALLOC_ARGS} \
@@ -374,7 +374,7 @@ else
     #     This is considered an abnormal process termination. Normally, we
     #     don't need to specify this exit code because the shell propagates it.
     #     Unfortunately, the signal handler doesn't work as expected in Dash,
-    #     thus we need to explicitely restate the exit code.
+    #     thus we need to explicitly restate the exit code.
     trap "stop_rabbitmq_server; exit 0" HUP TERM TSTP
     trap "stop_rabbitmq_server; exit 130" INT
 

scripts/rabbitmq-server.bat

@@ -167,7 +167,7 @@ rem
 rem The defaults are meant to reduce RabbitMQ's memory usage and help
 rem it reclaim memory at the cost of a slight decrease in performance
 rem (due to an increase in memory operations). These defaults can be
-rem overriden using the RABBITMQ_SERVER_ERL_ARGS variable.
+rem overridden using the RABBITMQ_SERVER_ERL_ARGS variable.
 
 set RABBITMQ_DEFAULT_ALLOC_ARGS=+MBas ageffcbf +MHas ageffcbf +MBlmbcs 512 +MHlmbcs 512 +MMmcs 30
 

scripts/rabbitmq-service.bat

@@ -254,7 +254,7 @@ rem
 rem The defaults are meant to reduce RabbitMQ's memory usage and help
 rem it reclaim memory at the cost of a slight decrease in performance
 rem (due to an increase in memory operations). These defaults can be
-rem overriden using the RABBITMQ_SERVER_ERL_ARGS variable.
+rem overridden using the RABBITMQ_SERVER_ERL_ARGS variable.
 
 set RABBITMQ_DEFAULT_ALLOC_ARGS=+MBas ageffcbf +MHas ageffcbf +MBlmbcs 512 +MHlmbcs 512 +MMmcs 30
 

src/amqqueue.erl

@@ -107,7 +107,7 @@
                                                                 %% implicit
                                                                 %% update
                                                                 %% as above
-          gm_pids = [] :: [pid()] | none | '_', %% transient
+          gm_pids = [] :: [{pid(), pid()} | pid()] | none | '_', %% transient
           decorators :: [atom()] | none | undefined | '_', %% transient,
                                                           %% recalculated
                                                           %% as above
@@ -359,14 +359,14 @@ get_exclusive_owner(#amqqueue{exclusive_owner = Owner}) ->
 get_exclusive_owner(Queue) ->
     amqqueue_v1:get_exclusive_owner(Queue).
 
--spec get_gm_pids(amqqueue()) -> [pid()] | none.
+-spec get_gm_pids(amqqueue()) -> [{pid(), pid()} | pid()] | none.
 
 get_gm_pids(#amqqueue{gm_pids = GMPids}) ->
     GMPids;
 get_gm_pids(Queue) ->
     amqqueue_v1:get_gm_pids(Queue).
 
--spec set_gm_pids(amqqueue(), [pid()] | none) -> amqqueue().
+-spec set_gm_pids(amqqueue(), [{pid(), pid()} | pid()] | none) -> amqqueue().
 
 set_gm_pids(#amqqueue{} = Queue, GMPids) ->
     Queue#amqqueue{gm_pids = GMPids};

src/gm.erl

@@ -1271,7 +1271,7 @@ neighbour_cast(N, Msg) -> ?INSTR_MOD:cast(get_pid(N), Msg).
 neighbour_call(N, Msg) -> ?INSTR_MOD:call(get_pid(N), Msg, infinity).
 
 %% ---------------------------------------------------------------------------
-%% View monitoring and maintanence
+%% View monitoring and maintenance
 %% ---------------------------------------------------------------------------
 
 ensure_neighbour(_Ver, Self, {Self, undefined}, Self) ->

src/lager_exchange_backend.erl

@@ -165,8 +165,8 @@ handle_log_event({log, Message},
                                  headers      = Headers},
             Body = rabbit_data_coercion:to_binary(Formatter:format(Message, FormatConfig)),
             case rabbit_basic:publish(LogExch, RoutingKey, AmqpMsg, Body) of
-                {ok, _DeliveredQPids} -> ok;
-                {error, not_found}    -> ok
+                ok                 -> ok;
+                {error, not_found} -> ok
             end,
             {ok, State};
         false ->

src/lqueue.erl

@@ -52,17 +52,17 @@ is_empty(_)       -> false.
 
 in(V, {L, Q}) -> {L+1, ?QUEUE:in(V, Q)}.
 
--spec in_r(value(), ?MODULE()) -> ?MODULE().
+-spec in_r(value(), ?MODULE(T)) -> ?MODULE(T).
 
 in_r(V, {L, Q}) -> {L+1, ?QUEUE:in_r(V, Q)}.
 
--spec out(?MODULE(T)) -> {result(T), ?MODULE()}.
+-spec out(?MODULE(T)) -> {result(T), ?MODULE(T)}.
 
 out({0, _Q} = Q) -> {empty, Q};
 out({L,  Q})     -> {Result, Q1} = ?QUEUE:out(Q),
                     {Result, {L-1, Q1}}.
 
--spec out_r(?MODULE(T)) -> {result(T), ?MODULE()}.
+-spec out_r(?MODULE(T)) -> {result(T), ?MODULE(T)}.
 
 out_r({0, _Q} = Q) -> {empty, Q};
 out_r({L,  Q})     -> {Result, Q1} = ?QUEUE:out_r(Q),

src/rabbit_amqqueue.erl

@@ -77,7 +77,7 @@
 -type qpids() :: [pid()].
 -type qlen() :: rabbit_types:ok(non_neg_integer()).
 -type qfun(A) :: fun ((amqqueue:amqqueue()) -> A | no_return()).
--type qmsg() :: {name(), pid(), msg_id(), boolean(), rabbit_types:message()}.
+-type qmsg() :: {name(), pid() | {atom(), pid()}, msg_id(), boolean(), rabbit_types:message()}.
 -type msg_id() :: non_neg_integer().
 -type ok_or_errors() ::
         'ok' | {'error', [{'error' | 'exit' | 'throw', any()}]}.
@@ -244,7 +244,9 @@ recover_durable_queues(QueuesAndRecoveryTerms) ->
               rabbit_framing:amqp_table(),
               rabbit_types:maybe(pid()),
               rabbit_types:username()) ->
-    {'new' | 'existing' | 'absent' | 'owner_died', amqqueue:amqqueue()} |
+    {'new' | 'existing' | 'owner_died', amqqueue:amqqueue()} |
+    {'new', amqqueue:amqqueue(), rabbit_fifo_client:state()} |
+    {'absent', amqqueue:amqqueue(), absent_reason()} |
     rabbit_types:channel_exit().
 
 declare(QueueName, Durable, AutoDelete, Args, Owner, ActingUser) ->
@@ -1038,6 +1040,7 @@ notify_policy_changed(Q) when ?amqqueue_is_quorum(Q) ->
 
 -spec consumers(amqqueue:amqqueue()) ->
           [{pid(), rabbit_types:ctag(), boolean(), non_neg_integer(),
+            boolean(), atom(),
             rabbit_framing:amqp_table(), rabbit_types:username()}].
 
 consumers(Q) when ?amqqueue_is_classic(Q) ->

src/rabbit_amqqueue_process.erl

@@ -1195,7 +1195,7 @@ prioritise_cast(Msg, _Len, State) ->
 %% will be rate limited by how fast consumers receive messages -
 %% i.e. by notify_sent. We prioritise ack and resume to discourage
 %% starvation caused by prioritising notify_sent. We don't vary their
-%% prioritiy since acks should stay in order (some parts of the queue
+%% priority since acks should stay in order (some parts of the queue
 %% stack are optimised for that) and to make things easier to reason
 %% about. Finally, we prioritise ack over resume since it should
 %% always reduce memory use.
@@ -1592,10 +1592,10 @@ handle_cast({force_event_refresh, Ref},
             [emit_consumer_created(
                Ch, CTag, false, AckRequired, QName, Prefetch,
                Args, Ref, ActingUser) ||
-                {Ch, CTag, AckRequired, Prefetch, Args, ActingUser}
+                {Ch, CTag, AckRequired, Prefetch, _, _, Args, ActingUser}
                     <- AllConsumers];
         {Ch, CTag} ->
-            [{Ch, CTag, AckRequired, Prefetch, Args, ActingUser}] = AllConsumers,
+            [{Ch, CTag, AckRequired, Prefetch, _, _, Args, ActingUser}] = AllConsumers,
             emit_consumer_created(
               Ch, CTag, true, AckRequired, QName, Prefetch, Args, Ref, ActingUser)
     end,

src/rabbit_backing_queue.erl

@@ -158,7 +158,7 @@
 
 %% Drop messages from the head of the queue while the supplied
 %% predicate on message properties returns true. Returns the first
-%% message properties for which the predictate returned false, or
+%% message properties for which the predicate returned false, or
 %% 'undefined' if the whole backing queue was traversed w/o the
 %% predicate ever returning false.
 -callback dropwhile(msg_pred(), state())
@@ -259,7 +259,7 @@
 
 -callback set_queue_mode(queue_mode(), state()) -> state().
 
--callback zip_msgs_and_acks(delivered_publish(),
+-callback zip_msgs_and_acks([delivered_publish()],
                             [ack()], Acc, state())
                            -> Acc.
 

src/rabbit_basic.erl

@@ -30,7 +30,7 @@
 -type properties_input() ::
         rabbit_framing:amqp_property_record() | [{atom(), any()}].
 -type publish_result() ::
-        {ok, [pid()]} | rabbit_types:error('not_found').
+        ok | rabbit_types:error('not_found').
 -type header() :: any().
 -type headers() :: rabbit_framing:amqp_table() | 'undefined'.
 
@@ -76,8 +76,7 @@ publish(Delivery = #delivery{
 
 publish(X, Delivery) ->
     Qs = rabbit_amqqueue:lookup(rabbit_exchange:route(X, Delivery)),
-    DeliveredQPids = rabbit_amqqueue:deliver(Qs, Delivery),
-    {ok, DeliveredQPids}.
+    rabbit_amqqueue:deliver(Qs, Delivery).
 
 -spec delivery
         (boolean(), boolean(), rabbit_types:message(), undefined | integer()) ->

src/rabbit_channel.erl

@@ -869,9 +869,7 @@ send(Command, #ch{writer_pid = WriterPid}) ->
     ok = rabbit_writer:send_command(WriterPid, Command).
 
 format_soft_error(#amqp_error{name = N, explanation = E, method = M}) ->
-    io_lib:format("operation ~s caused a channel exception ~s: ~ts", [M, N, E]);
-format_soft_error(Reason) ->
-    Reason.
+    io_lib:format("operation ~s caused a channel exception ~s: ~ts", [M, N, E]).
 
 handle_exception(Reason, State = #ch{protocol     = Protocol,
                                      channel      = Channel,
@@ -1455,7 +1453,7 @@ handle_method(#'basic.qos'{prefetch_size = Size}, _, _State) when Size /= 0 ->
 handle_method(#'basic.qos'{global         = false,
                            prefetch_count = PrefetchCount},
               _, State = #ch{limiter = Limiter}) ->
-    %% Ensures that if default was set, it's overriden
+    %% Ensures that if default was set, it's overridden
     Limiter1 = rabbit_limiter:unlimit_prefetch(Limiter),
     {reply, #'basic.qos_ok'{}, State#ch{consumer_prefetch = PrefetchCount,
                                         limiter = Limiter1}};
@@ -2145,19 +2143,14 @@ send_confirms_and_nacks(State) ->
 
 send_nacks([], _, State) ->
     State;
-send_nacks(_Rs, _, State = #ch{state = closing,
-                                tx    = none}) -> %% optimisation
+send_nacks(_Rs, _, State = #ch{state = closing}) -> %% optimisation
     State;
-send_nacks(Rs, Cs, State = #ch{tx = none}) ->
+send_nacks(Rs, Cs, State) ->
     coalesce_and_send(Rs, Cs,
                       fun(MsgSeqNo, Multiple) ->
                               #'basic.nack'{delivery_tag = MsgSeqNo,
                                             multiple     = Multiple}
-                      end, State);
-send_nacks(_MXs, _, State = #ch{state = closing}) -> %% optimisation
-    State#ch{tx = failed};
-send_nacks(_, _, State) ->
-    maybe_complete_tx(State#ch{tx = failed}).
+                      end, State).
 
 send_confirms([], _, State) ->
     State;

src/rabbit_feature_flags.erl

@@ -1411,7 +1411,7 @@ does_node_support(Node, FeatureNames, Timeout) ->
             %% If rabbit_feature_flags:is_supported_locally/1 is undefined
             %% on the remote node, we consider it to be a 3.7.x node.
             %%
-            %% Theoritically, it could be an older version (3.6.x and
+            %% Theoretically, it could be an older version (3.6.x and
             %% older). But the RabbitMQ version consistency check
             %% (rabbit_misc:version_minor_equivalent/2) called from
             %% rabbit_mnesia:check_rabbit_consistency/2 already blocked

src/rabbit_ff_registry.erl

@@ -148,7 +148,7 @@ is_registry_written_to_disk() ->
 always_return_true() ->
     %% This function is here to trick Dialyzer. We want some functions
     %% in this initial on-disk registry to always return `true` or
-    %% `false`. However the generated regsitry will return actual
+    %% `false`. However the generated registry will return actual
     %% booleans. The `-spec()` correctly advertises a return type of
     %% `boolean()`. But in the meantime, Dialyzer only knows about this
     %% copy which, without the trick below, would always return either
@@ -161,7 +161,7 @@ always_return_true() ->
     %% That's why this function makes a call which we know the result,
     %% but not Dialyzer, to "create" that hard-coded `true` return
     %% value.
-    rand:uniform(1) > 0.
+    erlang:get({?MODULE, always_undefined}) =:= undefined.
 
 always_return_false() ->
     not always_return_true().