Merge pull request #1875 from rabbitmq/fix-more-dialyzer

Fix more dialyzer warnings

Author: michaelklishin

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/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()}]}.
@@ -237,7 +237,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) ->
@@ -1031,6 +1033,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

@@ -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

@@ -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,
@@ -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_ff_registry.erl

@@ -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().

src/rabbit_fifo_client.erl

@@ -407,12 +407,8 @@ purge(Node) ->
 stat(Leader) ->
     %% short timeout as we don't want to spend too long if it is going to
     %% fail anyway
-    case ra:local_query(Leader, fun rabbit_fifo:query_stat/1, 250) of
-        {ok, {_, {R, C}}, _} ->
-            {ok, R, C};
-        Err ->
-            Err
-    end.
+    {ok, {_, {R, C}}, _} = ra:local_query(Leader, fun rabbit_fifo:query_stat/1, 250),
+    {ok, R, C}.
 
 %% @doc returns the cluster name
 -spec cluster_name(state()) -> cluster_name().

src/rabbit_networking.erl

@@ -70,6 +70,11 @@
 -type protocol() :: atom().
 -type label() :: string().
 
+%% @todo Remove once Dialyzer only runs on Erlang/OTP 21.3 or above.
+-dialyzer({nowarn_function, boot/0}).
+-dialyzer({nowarn_function, boot_listeners/3}).
+-dialyzer({nowarn_function, record_distribution_listener/0}).
+
 -spec boot() -> 'ok'.
 
 boot() ->
@@ -285,9 +290,6 @@ tcp_listener_stopped(Protocol, Opts, IPAddress, Port) ->
                      port = Port,
                      opts = Opts}).
 
-%% @todo Remove once Dialyzer only runs on Erlang/OTP 21.3 or above.
--dialyzer({nowarn_function, record_distribution_listener/0}).
-
 record_distribution_listener() ->
     {Name, Host} = rabbit_nodes:parts(node()),
     {port, Port, _Version} = erl_epmd:port_please(Name, Host),

src/rabbit_queue_consumers.erl

@@ -86,8 +86,8 @@ inactive(#state{consumers = Consumers}) ->
     priority_queue:is_empty(Consumers).
 
 -spec all(state()) -> [{ch(), rabbit_types:ctag(), boolean(),
-                        non_neg_integer(), rabbit_framing:amqp_table(),
-                        rabbit_types:username()}].
+                        non_neg_integer(), boolean(), atom(),
+                        rabbit_framing:amqp_table(), rabbit_types:username()}].
 
 all(State) ->
     all(State, none, false).

src/rabbit_quorum_queue.erl

@@ -85,8 +85,8 @@ init_state({Name, _}, QName = #resource{}) ->
                             fun() -> credit_flow:unblock(Name), ok end).
 
 -spec handle_event({'ra_event', amqqueue:ra_server_id(), any()}, rabbit_fifo_client:state()) ->
-                          {'internal', Correlators :: [term()], rabbit_fifo_client:state()} |
-                          {rabbit_fifo:client_msg(), rabbit_fifo_client:state()}.
+                          {internal, Correlators :: [term()], rabbit_fifo_client:actions(), rabbit_fifo_client:state()} |
+                          {rabbit_fifo:client_msg(), rabbit_fifo_client:state()} | eol.
 
 handle_event({ra_event, From, Evt}, QState) ->
     rabbit_fifo_client:handle_ra_event(From, Evt, QState).
@@ -560,12 +560,7 @@ info(Q, Items) ->
 stat(Q) when ?is_amqqueue(Q) ->
     Leader = amqqueue:get_pid(Q),
     try
-        case rabbit_fifo_client:stat(Leader) of
-            {ok, _, _} = Stat ->
-                Stat;
-            _ ->
-                {ok, 0, 0}
-        end
+        {ok, _, _} = rabbit_fifo_client:stat(Leader)
     catch
         _:_ ->
             %% Leader is not available, cluster might be in minority
@@ -921,7 +916,7 @@ format(Q) when ?is_amqqueue(Q) ->
 is_process_alive(Name, Node) ->
     erlang:is_pid(rpc:call(Node, erlang, whereis, [Name], ?RPC_TIMEOUT)).
 
--spec quorum_messages(atom()) -> non_neg_integer().
+-spec quorum_messages(rabbit_amqqueue:name()) -> non_neg_integer().
 
 quorum_messages(QName) ->
     case ets:lookup(queue_coarse_metrics, QName) of

src/rabbit_trace.erl

@@ -106,7 +106,7 @@ trace(#exchange{name = Name}, #basic_message{exchange_name = Name},
     ok;
 trace(X, Msg = #basic_message{content = #content{payload_fragments_rev = PFR}},
       RKPrefix, RKSuffix, Extra) ->
-    {ok, _} = rabbit_basic:publish(
+    ok = rabbit_basic:publish(
                 X, <<RKPrefix/binary, ".", RKSuffix/binary>>,
                 #'P_basic'{headers = msg_to_table(Msg) ++ Extra}, PFR),
     ok.