Specs and bugfixes

Author: dcorbacho

deps/rabbit/src/rabbit_binding.erl

@@ -155,8 +155,6 @@ remove(Binding0, InnerFun, ActingUser) ->
     case
         rabbit_db_binding:delete(Binding, InnerFun)
     of
-        ok ->
-            ok;
         {error, _} = Err ->
             Err;
         Deletions ->
@@ -289,6 +287,15 @@ group_bindings_fold(Fun, Acc, [B = #binding{source = SrcName} | Bs],
                     OnlyDurable) ->
     group_bindings_fold(Fun, SrcName, Acc, Bs, [B], OnlyDurable).
 
+-spec group_bindings_fold(Fun, Name, Deletions, [Binding], [Binding], OnlyDurable)
+                         -> Ret when
+      Fun :: fun((Name, [Binding], Deletions, OnlyDurable) ->
+                        Deletions),
+      Name :: rabbit_exchange:name(),
+      Deletions :: rabbit_binding:deletions(),
+      Binding :: rabbit_types:binding(),
+      OnlyDurable :: boolean(),
+      Ret :: Deletions.
 group_bindings_fold(
   Fun, SrcName, Acc, [B = #binding{source = SrcName} | Bs], Bindings,
   OnlyDurable) ->
@@ -393,7 +400,7 @@ notify_bindings_deletion(Bs, ActingUser) ->
      || B <- Bs],
     ok.
 
--spec process_deletions(deletions()) -> rabbit_misc:thunk('ok').
+-spec process_deletions(deletions()) -> deletions().
 process_deletions(Deletions) ->
     dict:map(fun (_XName, {X, deleted, Bindings}) ->
                      Bs = lists:flatten(Bindings),

deps/rabbit/src/rabbit_classic_queue.erl

@@ -122,7 +122,7 @@ is_recoverable(Q) when ?is_amqqueue(Q) ->
     %% record if it is a mirrored queue (such info is now obtained from
     %% the policy). Thus, we must check if the local pid is alive
     %% - if the record is present - in order to restart.
-    (rabbit_amqqueue:exists(Q)
+    (rabbit_amqqueue:exists(amqqueue:get_name(Q))
      orelse not rabbit_mnesia:is_process_alive(amqqueue:get_pid(Q))).
 
 recover(VHost, Queues) ->

deps/rabbit/src/rabbit_db_binding.erl

@@ -61,8 +61,8 @@ exists_in_mnesia(Binding) ->
 
 -spec create(Binding, ChecksFun) -> Ret when
       Binding :: rabbit_types:binding(),
-      Src :: rabbit_types:r('exchange'),
-      Dst :: rabbit_types:r('exchange') | rabbit_types:r('queue'),
+      Src :: rabbit_types:binding_source(),
+      Dst :: rabbit_types:binding_destination(),
       ChecksFun :: fun((Src, Dst) -> ok | {error, Reason :: any()}),
       Ret :: ok | {error, Reason :: any()}.
 %% @doc Writes a binding if it doesn't exist already and passes the validation in
@@ -83,10 +83,10 @@ create(Binding, ChecksFun) ->
 
 -spec delete(Binding, ChecksFun) -> Ret when
       Binding :: rabbit_types:binding(),
-      Src :: rabbit_types:r('exchange'),
-      Dst :: rabbit_types:r('exchange') | rabbit_types:r('queue'),
+      Src :: rabbit_types:binding_source(),
+      Dst :: rabbit_types:binding_destination(),
       ChecksFun :: fun((Src, Dst) -> ok | {error, Reason :: any()}),
-      Ret :: ok | {error, Reason :: any()}.
+      Ret :: rabbit_binding:deletions() | {error, Reason :: any()}.
 %% @doc Deletes a binding record from the database if it passes the validation in
 %% `ChecksFun`. It also triggers the deletion of auto-delete exchanges if needed.
 %%
@@ -124,7 +124,7 @@ get_all_in_mnesia(VHost) ->
     [B || #route{binding = B} <- rabbit_db:list_in_mnesia(rabbit_route, Match)].
 
 -spec get_all_for_source(Src) -> [Binding] when
-      Src :: rabbit_types:r('exchange'),
+      Src :: rabbit_types:binding_source(),
       Binding :: rabbit_types:binding().
 %% @doc Returns all binding records for a given exchange in the given virtual host.
 %%
@@ -142,7 +142,7 @@ get_all_for_source_in_mnesia(Resource) ->
     [B || #route{binding = B} <- rabbit_db:list_in_mnesia(rabbit_route, Route)].
 
 -spec get_all_for_destination(Dst) -> [Binding] when
-      Dst :: rabbit_types:r('exchange') | rabbit_types:r('queue'),
+      Dst :: rabbit_types:binding_destination(),
       Binding :: rabbit_types:binding().
 %% @doc Returns all binding records for a given exchange or queue destination
 %%  in the given virtual host.
@@ -163,8 +163,8 @@ get_all_for_destination_in_mnesia(Resource) ->
         #reverse_route{reverse_binding = B} <- rabbit_db:list_in_mnesia(rabbit_reverse_route, Route)].
 
 -spec get_all(Src, Dst) -> [Binding] when
-      Src :: rabbit_types:r('exchange'),
-      Dst :: rabbit_types:r('exchange') | rabbit_types:r('queue'),
+      Src :: rabbit_types:binding_source(),
+      Dst :: rabbit_types:binding_destination(),
       Binding :: rabbit_types:binding().
 %% @doc Returns all binding records for a given source and destination
 %% in the given virtual host.
@@ -397,6 +397,11 @@ delete_in_mnesia(Binding, ChecksFun) ->
               end
       end, fun absent_errs_only_in_mnesia/1).
 
+-spec delete_in_mnesia(Src, Dst, Binding) -> Ret when
+      Src :: rabbit_types:exchange() | amqqueue:amqqueue(),
+      Dst :: rabbit_types:exchange() | amqqueue:amqqueue(),
+      Binding :: rabbit_types:binding(),
+      Ret :: fun(() -> rabbit_binding:deletions()).
 delete_in_mnesia(Src, Dst, B) ->
     ok = sync_route(#route{binding = B}, rabbit_binding:binding_type(Src, Dst),
                     should_index_table(Src), fun delete/3),
@@ -596,6 +601,13 @@ sync_index_route(Route, true, Fun) ->
 sync_index_route(_, _, _) ->
     ok.
 
+-spec maybe_auto_delete_exchange_in_mnesia(ExchangeName, [Binding], Deletions, OnlyDurable)
+                                          -> Ret when
+      ExchangeName :: rabbit_exchange:name(),
+      Binding :: rabbit_types:binding(),
+      Deletions :: rabbit_binding:deletions(),
+      OnlyDurable :: boolean(),
+      Ret :: rabbit_binding:deletions().
 maybe_auto_delete_exchange_in_mnesia(XName, Bindings, Deletions, OnlyDurable) ->
     {Entry, Deletions1} =
         case rabbit_db_exchange:maybe_auto_delete_in_mnesia(XName, OnlyDurable) of

deps/rabbit/src/rabbit_db_exchange.erl

@@ -44,8 +44,6 @@
 -define(MNESIA_DURABLE_TABLE, rabbit_durable_exchange).
 -define(MNESIA_SERIAL_TABLE, rabbit_exchange_serial).
 
--type name() :: rabbit_types:r('exchange').
-
 %% -------------------------------------------------------------------
 %% get_all().
 %% -------------------------------------------------------------------
@@ -107,8 +105,8 @@ get_all_durable_in_mnesia() ->
 %% list().
 %% -------------------------------------------------------------------
 
--spec list() -> [Exchange] when
-      Exchange :: rabbit_types:exchange().
+-spec list() -> [ExchangeName] when
+      ExchangeName :: rabbit_exchange:name().
 %% @doc Lists the names of all exchanges.
 %%
 %% @returns a list of exchange names.
@@ -128,7 +126,7 @@ list_in_mnesia() ->
 %% -------------------------------------------------------------------
 
 -spec get(ExchangeName) -> Ret when
-      ExchangeName :: name(),
+      ExchangeName :: rabbit_exchange:name(),
       Ret :: {ok, Exchange :: rabbit_types:exchange()} | {error, not_found}.
 %% @doc Returns the record of the exchange named `Name'.
 %%
@@ -150,7 +148,7 @@ get_in_mnesia(Name) ->
 %% -------------------------------------------------------------------
 
 -spec get_many([ExchangeName]) -> [Exchange] when
-      ExchangeName :: name(),
+      ExchangeName :: rabbit_exchange:name(),
       Exchange :: rabbit_types:exchange().
 %% @doc Returns the records of the exchanges named `Name'.
 %%
@@ -192,7 +190,7 @@ count_in_mnesia() ->
 %% -------------------------------------------------------------------
 
 -spec update(ExchangeName, UpdateFun) -> ok when
-      ExchangeName :: name(),
+      ExchangeName :: rabbit_exchange:name(),
       UpdateFun :: fun((Exchange) -> Exchange).
 %% @doc Updates an existing exchange record using the result of
 %% `UpdateFun'.
@@ -214,6 +212,12 @@ update_in_mnesia(XName, Fun) ->
               ok
       end).
 
+-spec update_in_mnesia_tx(ExchangeName, UpdateFun) -> Ret when
+      ExchangeName :: rabbit_exchange:name(),
+      Exchange :: rabbit_types:exchange(),
+      UpdateFun :: fun((Exchange) -> Exchange),
+      Ret :: not_found | Exchange.
+
 update_in_mnesia_tx(Name, Fun) ->
     Table = {?MNESIA_TABLE, Name},
     case mnesia:wread(Table) of
@@ -294,7 +298,7 @@ set_in_mnesia(Xs) when is_list(Xs) ->
 %% -------------------------------------------------------------------
 
 -spec peek_serial(ExchangeName) -> Serial when
-      ExchangeName :: name(),
+      ExchangeName :: rabbit_exchange:name(),
       Serial :: integer().
 %% @doc Returns the next serial number without increasing it.
 %%
@@ -324,7 +328,7 @@ peek_serial_in_mnesia_tx(XName, LockType) ->
 %% -------------------------------------------------------------------
 
 -spec next_serial(ExchangeName) -> Serial when
-      ExchangeName :: name(),
+      ExchangeName :: rabbit_exchange:name(),
       Serial :: integer().
 %% @doc Returns the next serial number and increases it.
 %%
@@ -342,6 +346,10 @@ next_serial_in_mnesia(XName) ->
                                                    next_serial_in_mnesia_tx(XName)
                                            end).
 
+-spec next_serial_in_mnesia_tx(ExchangeName) -> Serial when
+      ExchangeName :: rabbit_exchange:name(),
+      Serial :: integer().
+
 next_serial_in_mnesia_tx(XName) ->
     Serial = peek_serial_in_mnesia_tx(XName, write),
     ok = mnesia:write(?MNESIA_SERIAL_TABLE,
@@ -353,7 +361,7 @@ next_serial_in_mnesia_tx(XName) ->
 %% -------------------------------------------------------------------
 
 -spec delete(ExchangeName, IfUnused) -> Ret when
-      ExchangeName :: name(),
+      ExchangeName :: rabbit_exchange:name(),
       IfUnused :: boolean(),
       Exchange :: rabbit_types:exchange(),
       Binding :: rabbit_types:binding(),
@@ -398,6 +406,14 @@ conditional_delete_in_mnesia(X = #exchange{name = XName}, OnlyDurable) ->
 unconditional_delete_in_mnesia(X, OnlyDurable) ->
     delete_in_mnesia(X, OnlyDurable, true).
 
+-spec delete_in_mnesia(Exchange, OnlyDurable, RemoveBindingsForSource) -> Ret when
+      Exchange :: rabbit_types:exchange(),
+      OnlyDurable :: boolean(),
+      RemoveBindingsForSource :: boolean(),
+      Exchange :: rabbit_types:exchange(),
+      Binding :: rabbit_types:binding(),
+      Deletions :: dict:dict(),
+      Ret :: {error, not_found} | {error, in_use} | {deleted, Exchange, [Binding], Deletions}.
 delete_in_mnesia(X = #exchange{name = XName}, OnlyDurable, RemoveBindingsForSource) ->
     ok = mnesia:delete({?MNESIA_TABLE, XName}),
     mnesia:delete({?MNESIA_DURABLE_TABLE, XName}),
@@ -409,7 +425,7 @@ delete_in_mnesia(X = #exchange{name = XName}, OnlyDurable, RemoveBindingsForSour
 %% -------------------------------------------------------------------
 
 -spec delete_serial(ExchangeName) -> ok when
-      ExchangeName :: name().
+      ExchangeName :: rabbit_exchange:name().
 %% @doc Deletes an exchange serial record from the database.
 %%
 %% @returns ok
@@ -494,7 +510,7 @@ match_in_mnesia(Pattern) ->
 %% -------------------------------------------------------------------
 
 -spec exists(ExchangeName) -> Exists when
-      ExchangeName :: name(),
+      ExchangeName :: rabbit_exchange:name(),
       Exists :: boolean().
 %% @doc Indicates if the exchange named `Name' exists.
 %%
@@ -532,9 +548,12 @@ clear_in_mnesia() ->
 %% maybe_auto_delete_in_mnesia().
 %% -------------------------------------------------------------------
 
--spec maybe_auto_delete_in_mnesia
-        (rabbit_types:exchange(), boolean())
-        -> 'not_deleted' | {'deleted', rabbit_binding:deletions()}.
+-spec maybe_auto_delete_in_mnesia(ExchangeName, boolean()) -> Ret when
+      ExchangeName :: rabbit_exchange:name(),
+      Exchange :: rabbit_types:exchange(),
+      Deletions :: rabbit_binding:deletions(),
+      Ret ::  {'not_deleted', 'undefined' | Exchange} |
+              {'deleted', Exchange, Deletions}.
 maybe_auto_delete_in_mnesia(XName, OnlyDurable) ->
     case mnesia:read({case OnlyDurable of
                           true  -> ?MNESIA_DURABLE_TABLE;

deps/rabbit/src/rabbit_db_queue.erl

@@ -374,7 +374,7 @@ get_many_durable(Names) when is_list(Names) ->
 
 -spec update(QName, UpdateFun) -> Ret when
       QName :: rabbit_amqqueue:name(),
-      Queue :: amqqueue:amqeueue(),
+      Queue :: amqqueue:amqqueue(),
       UpdateFun :: fun((Queue) -> Queue),
       Ret :: Queue | not_found. 
 %% @doc Updates an existing queue record using `UpdateFun'.
@@ -611,11 +611,7 @@ delete_transient_in_mnesia(FilterFun) ->
                                ]))
            end),
     lists:unzip(lists:flatten(
-                  [case delete_many_transient_in_mnesia(Queues) of
-                       {error, noproc} -> [];
-                       {error, {timeout, _}} -> [];
-                       Value -> Value
-                   end || Queues <- partition_queues(Qs)]
+                  [delete_many_transient_in_mnesia(Queues) || Queues <- partition_queues(Qs)]
                  )).
 
 -spec delete_many_transient_in_mnesia([QName]) -> Ret when
@@ -725,7 +721,7 @@ set_dirty_in_mnesia(Q) ->
 
 -spec update_in_mnesia_tx(QName, UpdateFun) -> Ret when
       QName :: rabbit_amqqueue:name(),
-      Queue :: amqqueue:amqeueue(),
+      Queue :: amqqueue:amqqueue(),
       UpdateFun :: fun((Queue) -> Queue),
       Ret :: Queue | not_found.
 

deps/rabbit/src/rabbit_db_topic_exchange.erl

@@ -38,7 +38,7 @@ set(#binding{source = XName, key = RoutingKey, destination = Destination, args =
 %% -------------------------------------------------------------------
 
 -spec delete_all_for_exchange(ExchangeName) -> ok when
-      ExchangeName :: rabbit_types:r('exchange').
+      ExchangeName :: rabbit_exchange:name().
 %% @doc Deletes all topic bindings for the exchange named `ExchangeName'
 %%
 %% @private
@@ -68,7 +68,7 @@ delete(Bs) when is_list(Bs) ->
 %% -------------------------------------------------------------------
 
 -spec match(ExchangeName, RoutingKey) -> ok when
-      ExchangeName :: rabbit_types:r('exchange'),
+      ExchangeName :: rabbit_exchange:name(),
       RoutingKey :: binary().
 %% @doc Finds the topic binding matching the given exchange and routing key and returns
 %% the destination of the binding

deps/rabbit/src/rabbit_db_user.erl

@@ -358,7 +358,7 @@ clear_matching_user_permissions_in_mnesia_tx(Username, VHostName) ->
 %% -------------------------------------------------------------------
 
 -spec get_topic_permissions(Username, VHostName, ExchangeName) -> Ret when
-      Username :: internal_topic:topicname(),
+      Username :: internal_user:username(),
       VHostName :: vhost:name(),
       ExchangeName :: binary(),
       Ret :: TopicPermission | undefined,
@@ -546,7 +546,7 @@ clear_topic_permissions_in_mnesia_tx(Username, VHostName, ExchangeName) ->
 
 -spec clear_matching_topic_permissions(Username, VHostName, ExchangeName) ->
     Ret when
-      Username :: internal_topic:topicname() | '_',
+      Username :: internal_user:username() | '_',
       VHostName :: vhost:name() | '_',
       ExchangeName :: binary() | '_',
       Ret :: [#topic_permission{}].

deps/rabbit/src/rabbit_exchange.erl

@@ -79,8 +79,7 @@ serialise_events(X = #exchange{type = Type, decorators = Decorators}) ->
               rabbit_exchange_decorator:select(all, Decorators))
         orelse (type_to_module(Type)):serialise_events().
 
--spec serial(rabbit_types:exchange()) ->
-                       fun((boolean()) -> 'none' | pos_integer()).
+-spec serial(rabbit_types:exchange()) -> 'none' | pos_integer().
 
 serial(X) ->
     case serialise_events(X) of
@@ -271,8 +270,8 @@ update_scratch_fun(App, Fun, Decorators) ->
                        decorators = Decorators}
     end.
 
--spec update_decorators(name(), [atom()] | none | undefined) -> 'ok'.
-
+-spec update_decorators(name(), {[Decorator], [Decorator]}) -> 'ok' when
+      Decorator :: atom().
 update_decorators(Name, Decorators) ->
     Fun = fun(X) -> X#exchange{decorators = Decorators} end,
     ok = rabbit_db_exchange:update(Name, Fun).
@@ -443,9 +442,7 @@ process_deletions({error, _} = E) ->
 process_deletions({deleted, #exchange{name = XName} = X, Bs, Deletions}) ->
     rabbit_binding:process_deletions(
       rabbit_binding:add_deletion(
-        XName, {X, deleted, Bs}, Deletions));
-process_deletions(Deletions) ->
-    rabbit_binding:process_deletions(Deletions).
+        XName, {X, deleted, Bs}, Deletions)).
 
 -spec validate_binding
         (rabbit_types:exchange(), rabbit_types:binding())

deps/rabbitmq_consistent_hash_exchange/BUILD.bazel

@@ -43,7 +43,7 @@ xref(
 plt(
     name = "base_plt",
     libs = ["//deps/rabbitmq_cli:elixir"],
-    deps = ["//deps/rabbitmq_cli:elixir"] + BUILD_DEPS + DEPS + RUNTIME_DEPS,
+    deps = ["//deps/rabbitmq_cli:elixir"] + BUILD_DEPS + DEPS,
 )
 
 dialyze(