Merge branch 'master' into max_msg_size

Author: michaelklishin

priv/schema/rabbit.schema

@@ -258,7 +258,7 @@ end}.
 {translation, "rabbit.ssl_options.ciphers",
 fun(Conf) ->
     Settings = cuttlefish_variable:filter_by_prefix("ssl_options.ciphers", Conf),
-    [V || {_, V} <- Settings]
+    lists:reverse([V || {_, V} <- Settings])
 end}.
 
 %% ===========================================================================

src/rabbit_fifo.erl

@@ -56,10 +56,10 @@ return(Key, Value, #?MODULE{data = Data} = State)
   when is_integer(Key) ->
     State#?MODULE{data = maps:put(Key, Value, Data)}.
 
--spec delete(integer(), state()) -> state().
+-spec delete(Index :: integer(), state()) -> state().
 delete(Smallest, #?MODULE{data = Data0,
-                        largest = Largest,
-                        smallest = Smallest} = State) ->
+                          largest = Largest,
+                          smallest = Smallest} = State) ->
     Data = maps:remove(Smallest, Data0),
     case find_next(Smallest + 1, Largest, Data) of
         undefined ->

src/rabbit_fifo_index.erl

@@ -157,13 +157,15 @@ cancel_consumer_handler(QName, {ConsumerTag, ChPid}) ->
     case Node == node() of
         true -> cancel_consumer(QName, ChPid, ConsumerTag);
         false ->
+            %% this could potentially block for a while if the node is
+            %% in disconnected state or tcp buffers are full
             rpc:cast(Node, rabbit_quorum_queue,
                      cancel_consumer,
                      [QName, ChPid, ConsumerTag])
     end.
 
 cancel_consumer(QName, ChPid, ConsumerTag) ->
-    rabbit_core_metrics:consumer_deleted(ChPid, ConsumerTag, QName),
+    catch rabbit_core_metrics:consumer_deleted(ChPid, ConsumerTag, QName),
     rabbit_event:notify(consumer_deleted,
                         [{consumer_tag, ConsumerTag},
                          {channel,      ChPid},

src/rabbit_quorum_queue.erl

@@ -326,15 +326,15 @@ tcp_listen_options.exit_on_close = false",
         {ssl_options,
             [{cacertfile,"test/config_schema_SUITE_data/certs/cacert.pem"},
              {ciphers, [
-               "DHE-RSA-AES256-GCM-SHA384",
+               "ECDHE-ECDSA-AES256-GCM-SHA384",
+               "ECDHE-RSA-AES256-GCM-SHA384",
+               "ECDHE-ECDSA-AES256-SHA384",
+               "ECDHE-RSA-AES256-SHA384",
                "ECDH-ECDSA-AES256-GCM-SHA384",
-               "ECDH-ECDSA-AES256-SHA384",
                "ECDH-RSA-AES256-GCM-SHA384",
+               "ECDH-ECDSA-AES256-SHA384",
                "ECDH-RSA-AES256-SHA384",
-               "ECDHE-ECDSA-AES256-GCM-SHA384",
-               "ECDHE-ECDSA-AES256-SHA384",
-               "ECDHE-RSA-AES256-GCM-SHA384",
-               "ECDHE-RSA-AES256-SHA384"
+               "DHE-RSA-AES256-GCM-SHA384"
              ]},
              {certfile,"test/config_schema_SUITE_data/certs/cert.pem"},
              {keyfile,"test/config_schema_SUITE_data/certs/key.pem"},

test/config_schema_SUITE_data/rabbit.snippets

@@ -1539,8 +1539,8 @@ purge(Config) ->
     _DeliveryTag = consume(Ch, QQ, false),
     wait_for_messages_ready(Servers, RaName, 1),
     wait_for_messages_pending_ack(Servers, RaName, 1),
-    {'queue.purge_ok', 2} = amqp_channel:call(Ch, #'queue.purge'{queue = QQ}),
-    wait_for_messages_pending_ack(Servers, RaName, 0),
+    {'queue.purge_ok', 1} = amqp_channel:call(Ch, #'queue.purge'{queue = QQ}),
+    wait_for_messages_pending_ack(Servers, RaName, 1),
     wait_for_messages_ready(Servers, RaName, 0).
 
 sync_queue(Config) ->

test/quorum_queue_SUITE.erl

@@ -278,8 +278,16 @@ handle_op(purge, T) ->
 
 do_apply(Cmd, #t{effects = Effs, index = Index, state = S0,
                  log = Log} = T) ->
-    {S, Effects, _} = rabbit_fifo:apply(#{index => Index}, Cmd, [], S0),
-    T#t{state = S,
+    {St, Effects} = case rabbit_fifo:apply(#{index => Index}, Cmd, S0) of
+                       {S, _, E} when is_list(E) ->
+                           {S, E};
+                       {S, _, E} ->
+                           {S, [E]};
+                       {S, _} ->
+                           {S, []}
+                   end,
+
+    T#t{state = St,
         index = Index + 1,
         effects = enq_effs(Effects, Effs),
         log = [Cmd | Log]}.
@@ -338,9 +346,13 @@ prefixes(Source, N, Acc) ->
 
 run_log(InitState, Entries) ->
     lists:foldl(fun ({Idx, E}, {Acc0, Efx0}) ->
-                        case rabbit_fifo:apply(meta(Idx), E, Efx0, Acc0) of
-                            {Acc, Efx, _} ->
-                                {Acc, Efx}
+                        case rabbit_fifo:apply(meta(Idx), E, Acc0) of
+                            {Acc, _, Efx} when is_list(Efx) ->
+                                {Acc, Efx0 ++ Efx};
+                            {Acc, _, Efx}  ->
+                                {Acc, Efx0 ++ [Efx]};
+                            {Acc, _}  ->
+                                {Acc, Efx0}
                         end
                 end, {InitState, []}, Entries).