Adding missing function specs

Author: ChunyiLyu

deps/rabbitmq_mqtt/src/mqtt_machine.erl

@@ -172,6 +172,8 @@ apply(_Meta, Unknown, State) ->
     logger:error("MQTT Raft state machine v1 received unknown command ~tp", [Unknown]),
     {State, {error, {unknown_command, Unknown}}, []}.
 
+-spec state_enter(ra_server:ra_state(), state()) ->
+    ra_machine:effects().
 state_enter(leader, State) ->
     %% re-request monitors for all known pids, this would clean up
     %% records for all connections are no longer around, e.g. right after node restart
@@ -188,6 +190,7 @@ overview(#machine_state{client_ids = ClientIds,
 %% ==========================
 
 %% Avoids blocking the Raft leader.
+-spec notify_connection(pid(), 'duplicate_id' | 'decommission_node') -> pid().
 notify_connection(Pid, Reason) ->
   spawn(fun() -> gen_server2:cast(Pid, Reason) end).
 

deps/rabbitmq_mqtt/src/mqtt_machine_v0.erl

@@ -113,6 +113,8 @@ apply(_Meta, Unknown, State) ->
     logger:error("MQTT Raft state machine received an unknown command ~tp", [Unknown]),
     {State, {error, {unknown_command, Unknown}}, []}.
 
+-spec state_enter(ra_server:ra_state(), state()) ->
+    ra_machine:effects().
 state_enter(leader, State) ->
     %% re-request monitors for all known pids, this would clean up
     %% records for all connections are no longer around, e.g. right after node restart
@@ -123,6 +125,7 @@ state_enter(_, _) ->
 %% ==========================
 
 %% Avoids blocking the Raft leader.
+-spec notify_connection(pid(), duplicate_id | decommission_node) -> pid().
 notify_connection(Pid, Reason) ->
   spawn(fun() -> gen_server2:cast(Pid, Reason) end).
 

deps/rabbitmq_mqtt/src/rabbit_mqtt.erl

@@ -41,6 +41,7 @@ start(normal, []) ->
 stop(_) ->
     rabbit_mqtt_sup:stop_listeners().
 
+-spec emit_connection_info_all([node()], [atom()], reference(), pid()) -> term().
 emit_connection_info_all(Nodes, Items, Ref, AggregatorPid) ->
     case rabbit_mqtt_ff:track_client_id_in_ra() of
         true ->
@@ -57,6 +58,7 @@ emit_connection_info_all(Nodes, Items, Ref, AggregatorPid) ->
             rabbit_control_misc:await_emitters_termination(Pids)
     end.
 
+-spec emit_connection_info_local([atom()], reference(), pid()) -> ok.
 emit_connection_info_local(Items, Ref, AggregatorPid) ->
     LocalPids = local_connection_pids(),
     emit_connection_info(Items, Ref, AggregatorPid, LocalPids).

deps/rabbitmq_mqtt/src/rabbit_mqtt_collector.erl

@@ -36,6 +36,7 @@ register(ServerId, ClientId, Pid) ->
     erlang:send_after(5000, self(), {ra_event, undefined, register_timeout}),
     {ok, Corr}.
 
+-spec unregister(binary(), pid()) -> ok.
 unregister(ClientId, Pid) ->
     {ClusterName, _} = mqtt_node:server_id(),
     case ra_leaderboard:lookup_leader(ClusterName) of
@@ -49,6 +50,7 @@ unregister(ClientId, Pid) ->
 list_pids() ->
     list(fun(#machine_state{pids = Pids}) -> maps:keys(Pids) end).
 
+-spec list() -> term().
 list() ->
     list(fun(#machine_state{client_ids = Ids}) -> maps:to_list(Ids) end).
 
@@ -76,6 +78,7 @@ list(QF) ->
             end
     end.
 
+-spec leave(binary()) ->  ok | timeout | nodedown.
 leave(NodeBin) ->
     Node = binary_to_atom(NodeBin, utf8),
     ServerId = mqtt_node:server_id(),

deps/rabbitmq_mqtt/src/rabbit_mqtt_ff.erl

@@ -24,5 +24,6 @@
       callbacks     => #{enable => {mqtt_node, delete}}
      }}).
 
+-spec track_client_id_in_ra() -> boolean().
 track_client_id_in_ra() ->
     not rabbit_feature_flags:is_enabled(delete_ra_cluster_mqtt_node).

deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl

@@ -87,6 +87,14 @@
 
 -opaque state() :: #state{}.
 
+-type info_commands() :: host | port | peer_host | peer_port
+    | connected_at | ssl_login_name | vhost | user_who_performed_action | user
+    | clean_sess | will_msg | retainer_pid | exchange | prefetch
+    | messages_unconfirmed | messages_unacknowledged
+    | node
+    | client_id | client_properties
+    | channel_max | frame_max | auth_mechanism.
+
 -spec initial_state(Socket :: any(), ConnectionName :: binary()) ->
     state().
 initial_state(Socket, ConnectionName) ->
@@ -1329,9 +1337,13 @@ serialise_and_send_to_client(Packet, #state{cfg = #cfg{proto_ver = ProtoVer,
                          [Sock, Error, Packet#mqtt_packet.fixed, Packet#mqtt_packet.variable])
     end.
 
+-spec serialise(#mqtt_packet{}, state()) ->
+    iodata().
 serialise(Packet, #state{cfg = #cfg{proto_ver = ProtoVer}}) ->
     rabbit_mqtt_packet:serialise(Packet, ProtoVer).
 
+-spec terminate(any(), binary(), atom(),rabbit_mqtt_processor:state()) ->
+    ok.
 terminate(SendWill, ConnName, ProtoFamily, State) ->
     maybe_send_will(SendWill, ConnName, State),
     Infos = [{name, ConnName},
@@ -1427,11 +1439,15 @@ delete_queue(QName, Username) ->
               ok
       end).
 
+-spec handle_pre_hibernate() -> ok.
 handle_pre_hibernate() ->
     erase(permission_cache),
     erase(topic_permission_cache),
     ok.
 
+-spec handle_ra_event(register_timeout
+| {applied, [{reference(), ok}]}
+| {not_leader, term(), reference()}, state()) -> state().
 handle_ra_event({applied, [{Corr, ok}]},
                 State = #state{register_state = {pending, Corr}}) ->
     %% success case - command was applied transition into registered state
@@ -1796,6 +1812,7 @@ throttle(Conserve, Connected, #state{queues_soft_limit_exceeded = QSLE,
     not sets:is_empty(QSLE) orelse
     credit_flow:blocked().
 
+-spec info(info_commands(), state()) -> any.
 info(host, #state{cfg = #cfg{host = Val}}) -> Val;
 info(port, #state{cfg = #cfg{port = Val}}) -> Val;
 info(peer_host, #state{cfg = #cfg{peer_host = Val}}) -> Val;

deps/rabbitmq_mqtt/src/rabbit_mqtt_retained_msg_store_dets.erl

@@ -17,29 +17,37 @@
   table
 }).
 
+-type store_state() :: #store_state{}.
 
+-spec new(file:name_all(), rabbit_types:vhost()) -> store_state().
 new(Dir, VHost) ->
   Tid = open_table(Dir, VHost),
   #store_state{table = Tid}.
 
+-spec recover(file:name_all(), rabbit_types:vhost()) ->
+  {error, uninitialized} | {ok, store_state()}.
 recover(Dir, VHost) ->
   case open_table(Dir, VHost) of
     {error, _} -> {error, uninitialized};
     {ok, Tid}  -> {ok, #store_state{table = Tid}}
   end.
 
+-spec insert(binary(), mqtt_msg(), store_state()) -> ok.
 insert(Topic, Msg, #store_state{table = T}) ->
   ok = dets:insert(T, #retained_message{topic = Topic, mqtt_msg = Msg}).
 
+-spec lookup(binary(), store_state()) -> retained_message() | not_found.
 lookup(Topic, #store_state{table = T}) ->
   case dets:lookup(T, Topic) of
     []      -> not_found;
     [Entry] -> Entry
   end.
 
+-spec delete(binary(), store_state()) -> ok.
 delete(Topic, #store_state{table = T}) ->
   ok = dets:delete(T, Topic).
 
+-spec terminate(store_state()) -> ok.
 terminate(#store_state{table = T}) ->
   ok = dets:close(T).
 

deps/rabbitmq_mqtt/src/rabbit_mqtt_retained_msg_store_ets.erl

@@ -19,14 +19,18 @@
   filename
 }).
 
+-type store_state() :: #store_state{}.
 
+-spec new(file:name_all(), rabbit_types:vhost()) -> store_state().
 new(Dir, VHost) ->
   Path = rabbit_mqtt_util:path_for(Dir, VHost),
   TableName = rabbit_mqtt_util:vhost_name_to_table_name(VHost),
   _ = file:delete(Path),
   Tid = ets:new(TableName, [set, public, {keypos, #retained_message.topic}]),
   #store_state{table = Tid, filename = Path}.
 
+-spec recover(file:name_all(), rabbit_types:vhost()) ->
+  {error, uninitialized} | {ok, store_state()}.
 recover(Dir, VHost) ->
   Path = rabbit_mqtt_util:path_for(Dir, VHost),
   case ets:file2tab(Path) of
@@ -35,19 +39,23 @@ recover(Dir, VHost) ->
     {error, _} -> {error, uninitialized}
   end.
 
+-spec insert(binary(), mqtt_msg(), store_state()) -> ok.
 insert(Topic, Msg, #store_state{table = T}) ->
   true = ets:insert(T, #retained_message{topic = Topic, mqtt_msg = Msg}),
   ok.
 
+-spec lookup(binary(), store_state()) -> retained_message() | not_found.
 lookup(Topic, #store_state{table = T}) ->
   case ets:lookup(T, Topic) of
     []      -> not_found;
     [Entry] -> Entry
   end.
 
+-spec delete(binary(), store_state()) -> ok.
 delete(Topic, #store_state{table = T}) ->
   true = ets:delete(T, Topic),
   ok.
 
+-spec terminate(store_state()) -> ok.
 terminate(#store_state{table = T, filename = Path}) ->
   ok = ets:tab2file(T, Path, [{extended_info, [object_count]}]).

deps/rabbitmq_mqtt/src/rabbit_mqtt_retainer_sup.erl

@@ -17,6 +17,7 @@
 start_link(SupName) ->
     supervisor:start_link(SupName, ?MODULE, []).
 
+-spec child_for_vhost(binary()) -> pid().
 child_for_vhost(VHost) when is_binary(VHost) ->
   case rabbit_mqtt_retainer_sup:start_child(VHost) of
     {ok, Pid}                       -> Pid;

deps/rabbitmq_mqtt/src/rabbit_mqtt_sup.erl

@@ -70,6 +70,7 @@ init([{Listeners, SslListeners0}]) ->
         )
       ]}}.
 
+-spec stop_listeners() -> ok.
 stop_listeners() ->
     _ = rabbit_networking:stop_ranch_listener_of_protocol(?TCP_PROTOCOL),
     _ = rabbit_networking:stop_ranch_listener_of_protocol(?TLS_PROTOCOL),