Merge pull request #5895 from rabbitmq/mqtt

Native MQTT

Author: michaelklishin

deps/rabbit/apps/rabbitmq_prelaunch/src/rabbit_prelaunch_conf.erl

@@ -122,8 +122,13 @@ set_default_config() ->
                 {schedule_ms_limit, 0},
                 {heap_word_limit, 0},
                 {busy_port, false},
-                {busy_dist_port, true}]}
-                | OsirisConfig
+                {busy_dist_port, true}]},
+              {mnesia,
+               [
+                {dump_log_write_threshold, 5000},
+                {dump_log_time_threshold, 90000}
+               ]}
+              | OsirisConfig
              ],
     apply_erlang_term_based_config(Config).
 

deps/rabbit/docs/rabbitmq.conf.example

@@ -755,7 +755,7 @@
 # stomp.tcp_listen_options.nodelay   = true
 #
 # stomp.tcp_listen_options.exit_on_close = true
-# stomp.tcp_listen_options.send_timeout  = 120
+# stomp.tcp_listen_options.send_timeout  = 120000
 
 ## Proxy protocol support
 ##
@@ -838,7 +838,7 @@
 # mqtt.tcp_listen_options.nodelay   = true
 #
 # mqtt.tcp_listen_options.exit_on_close = true
-# mqtt.tcp_listen_options.send_timeout  = 120
+# mqtt.tcp_listen_options.send_timeout  = 120000
 
 ## TLS listener settings
 ## ## See https://rabbitmq.com/mqtt.html and https://rabbitmq.com/ssl.html for details.

deps/rabbit/src/rabbit.erl

@@ -727,7 +727,8 @@ status() ->
                  true ->
                      [{virtual_host_count, rabbit_vhost:count()},
                       {connection_count,
-                       length(rabbit_networking:connections_local())},
+                       length(rabbit_networking:connections_local()) +
+                       length(rabbit_networking:local_non_amqp_connections())},
                       {queue_count, total_queue_count()}];
                  false ->
                      []
@@ -1163,12 +1164,12 @@ config_locations() ->
 % This event is necessary for the stats timer to be initialized with
 % the correct values once the management agent has started
 force_event_refresh(Ref) ->
-    % direct connections, e.g. MQTT, STOMP
+    % direct connections, e.g. STOMP
     ok = rabbit_direct:force_event_refresh(Ref),
     % AMQP connections
     ok = rabbit_networking:force_connection_event_refresh(Ref),
-    % "external" connections, which are not handled by the "AMQP core",
-    % e.g. connections to the stream plugin
+    % non-AMQP connections, which are not handled by the "AMQP core",
+    % e.g. connections to the stream and MQTT plugins
     ok = rabbit_networking:force_non_amqp_connection_event_refresh(Ref),
     ok = rabbit_channel:force_event_refresh(Ref),
     ok = rabbit_amqqueue:force_event_refresh(Ref).

deps/rabbit/src/rabbit_access_control.erl

@@ -18,7 +18,7 @@
 
 -export_type([permission_atom/0]).
 
--type permission_atom() :: 'configure' | 'read' | 'write'.
+-type permission_atom() :: 'configure' | 'write' | 'read'.
 
 %%----------------------------------------------------------------------------
 
@@ -194,8 +194,8 @@ check_resource_access(User = #user{username       = Username,
               check_access(
                 fun() -> Module:check_resource_access(
                            auth_user(User, Impl), Resource, Permission, Context) end,
-                Module, "access to ~ts refused for user '~ts'",
-                [rabbit_misc:rs(Resource), Username]);
+                Module, "~s access to ~s refused for user '~s'",
+                [Permission, rabbit_misc:rs(Resource), Username]);
          (_, Else) -> Else
       end, ok, Modules).
 
@@ -207,8 +207,8 @@ check_topic_access(User = #user{username = Username,
             check_access(
                 fun() -> Module:check_topic_access(
                     auth_user(User, Impl), Resource, Permission, Context) end,
-                Module, "access to topic '~ts' in exchange ~ts refused for user '~ts'",
-                [maps:get(routing_key, Context), rabbit_misc:rs(Resource), Username]);
+                Module, "~s access to topic '~s' in exchange ~s refused for user '~s'",
+                [Permission, maps:get(routing_key, Context), rabbit_misc:rs(Resource), Username]);
             (_, Else) -> Else
         end, ok, Modules).
 

deps/rabbit/src/rabbit_alarm.erl

@@ -29,6 +29,9 @@
 
 -export([remote_conserve_resources/3]). %% Internal use only
 
+-export_type([resource_alarm_source/0,
+              resource_alert/0]).
+
 -define(SERVER, ?MODULE).
 
 -define(FILE_DESCRIPTOR_RESOURCE, <<"file descriptors">>).
@@ -46,6 +49,9 @@
 -type resource_alarm_source() :: 'disk' | 'memory'.
 -type resource_alarm() :: {resource_limit, resource_alarm_source(), node()}.
 -type alarm() :: local_alarm() | resource_alarm().
+-type resource_alert() :: {WasAlarmSetForNode :: boolean(),
+                           IsThereAnyAlarmsWithSameSourceInTheCluster :: boolean(),
+                           NodeForWhichAlarmWasSetOrCleared :: node()}.
 
 %%----------------------------------------------------------------------------
 

deps/rabbit/src/rabbit_amqqueue.erl

@@ -1763,7 +1763,7 @@ basic_get(Q, NoAck, LimiterPid, CTag, QStates) ->
                     non_neg_integer(), rabbit_types:ctag(), boolean(),
                     rabbit_framing:amqp_table(), any(), rabbit_types:username(),
                     rabbit_queue_type:state()) ->
-    {ok, rabbit_queue_type:state(), rabbit_queue_type:actions()} |
+    {ok, rabbit_queue_type:state()} |
     {error, term()} |
     {protocol_error, Type :: atom(), Reason :: string(), Args :: term()}.
 basic_consume(Q, NoAck, ChPid, LimiterPid,

deps/rabbit/src/rabbit_amqqueue_process.erl

@@ -121,6 +121,7 @@
 
 -define(CREATION_EVENT_KEYS,
         [name,
+         type,
          durable,
          auto_delete,
          arguments,
@@ -129,7 +130,7 @@
          user_who_performed_action
         ]).
 
--define(INFO_KEYS, [pid | ?CREATION_EVENT_KEYS ++ ?STATISTICS_KEYS -- [name]]).
+-define(INFO_KEYS, [pid | ?CREATION_EVENT_KEYS ++ ?STATISTICS_KEYS -- [name, type]]).
 
 %%----------------------------------------------------------------------------
 

deps/rabbit/src/rabbit_binding.erl

@@ -11,8 +11,8 @@
 
 -export([recover/0, recover/2, exists/1, add/2, add/3, remove/1, remove/2, remove/3, remove/4]).
 -export([list/1, list_for_source/1, list_for_destination/1,
-         list_for_source_and_destination/2, list_explicit/0,
-         list_between/2, has_any_between/2]).
+         list_for_source_and_destination/2, list_for_source_and_destination/3,
+         list_explicit/0, list_between/2, has_any_between/2]).
 -export([new_deletions/0, combine_deletions/2, add_deletion/3,
          process_deletions/2, binding_action/3]).
 -export([info_keys/0, info/1, info/2, info_all/1, info_all/2, info_all/4]).
@@ -253,26 +253,18 @@ list(VHostPath) ->
 list_for_source(?DEFAULT_EXCHANGE(VHostPath)) ->
     implicit_bindings(VHostPath);
 list_for_source(SrcName) ->
-    mnesia:async_dirty(
-      fun() ->
-              Route = #route{binding = #binding{source = SrcName, _ = '_'}},
-              [B || #route{binding = B}
-                        <- mnesia:match_object(rabbit_route, Route, read)]
-      end).
+    Route = #route{binding = #binding{source = SrcName, _ = '_'}},
+    Fun = list_for_route(Route, false),
+    mnesia:async_dirty(Fun).
 
 -spec list_for_destination
         (rabbit_types:binding_destination()) -> bindings().
 
 list_for_destination(DstName = #resource{}) ->
-    ExplicitBindings = mnesia:async_dirty(
-                         fun() ->
-                                 Route = #route{binding = #binding{destination = DstName,
-                                                                   _ = '_'}},
-                                 [reverse_binding(B) ||
-                                  #reverse_route{reverse_binding = B} <-
-                                  mnesia:match_object(rabbit_reverse_route,
-                                                      reverse_route(Route), read)]
-                         end),
+    Route = #route{binding = #binding{destination = DstName,
+                                      _ = '_'}},
+    Fun = list_for_route(Route, true),
+    ExplicitBindings = mnesia:async_dirty(Fun),
     implicit_for_destination(DstName) ++ ExplicitBindings.
 
 -spec list_between(
@@ -316,27 +308,40 @@ implicit_for_destination(DstQueue = #resource{kind = queue,
 implicit_for_destination(_) ->
     [].
 
--spec list_for_source_and_destination
-        (rabbit_types:binding_source(), rabbit_types:binding_destination()) ->
-                                                bindings().
+-spec list_for_source_and_destination(rabbit_types:binding_source(), rabbit_types:binding_destination()) ->
+    bindings().
+list_for_source_and_destination(SrcName, DstName) ->
+    list_for_source_and_destination(SrcName, DstName, false).
 
+-spec list_for_source_and_destination(rabbit_types:binding_source(), rabbit_types:binding_destination(), boolean()) ->
+    bindings().
 list_for_source_and_destination(?DEFAULT_EXCHANGE(VHostPath),
                                 #resource{kind = queue,
                                           virtual_host = VHostPath,
-                                          name = QName} = DstQueue) ->
+                                          name = QName} = DstQueue,
+                                _Reverse) ->
     [#binding{source = ?DEFAULT_EXCHANGE(VHostPath),
               destination = DstQueue,
               key = QName,
               args = []}];
-list_for_source_and_destination(SrcName, DstName) ->
-    mnesia:async_dirty(
-      fun() ->
-              Route = #route{binding = #binding{source      = SrcName,
-                                                destination = DstName,
-                                                _           = '_'}},
-              [B || #route{binding = B} <- mnesia:match_object(rabbit_route,
-                                                               Route, read)]
-      end).
+list_for_source_and_destination(SrcName, DstName, Reverse) ->
+    Route = #route{binding = #binding{source      = SrcName,
+                                      destination = DstName,
+                                      _           = '_'}},
+    Fun = list_for_route(Route, Reverse),
+    mnesia:async_dirty(Fun).
+
+list_for_route(Route, false) ->
+    fun() ->
+            [B || #route{binding = B} <- mnesia:match_object(rabbit_route, Route, read)]
+    end;
+list_for_route(Route, true) ->
+    fun() ->
+            [reverse_binding(B) ||
+             #reverse_route{reverse_binding = B} <-
+             mnesia:match_object(rabbit_reverse_route,
+                                 reverse_route(Route), read)]
+    end.
 
 -spec info_keys() -> rabbit_types:info_keys().