"formatting"

Author: kjnilsson

deps/rabbitmq_stream/src/Elixir.RabbitMQ.CLI.Ctl.Commands.AddSuperStreamCommand.erl

@@ -56,7 +56,8 @@ validate([], _Opts) ->
 validate([_Name], #{partitions := _, routing_keys := _}) ->
     {validation_failure,
      "Specify --partitions or routing-keys, not both."};
-validate([_Name], #{exchange_type := <<"x-super-stream">>, routing_keys := _}) ->
+validate([_Name],
+         #{exchange_type := <<"x-super-stream">>, routing_keys := _}) ->
     {validation_failure,
      "Exchange type x-super-stream cannot be used with routing-keys."};
 validate([_Name], #{partitions := Partitions}) when Partitions < 1 ->
@@ -153,8 +154,7 @@ usage() ->
       "s <partitions>] [--routing-keys <routing-keys>]">>.
 
 usage_additional() ->
-    [[<<"<name>">>,
-      <<"The name of the super stream.">>],
+    [[<<"<name>">>, <<"The name of the super stream.">>],
      [<<"--vhost <vhost>">>,
       <<"The virtual host the super stream is added to.">>],
      [<<"--partitions <partitions>">>,
@@ -180,33 +180,33 @@ run([SuperStream],
       timeout := Timeout,
       partitions := Partitions} =
         Opts) ->
-    Spec0 = maps:with([vhost,
-                       exchange_type], Opts),
-    Spec = Spec0#{username => cli_acting_user(),
-                  name => SuperStream,
-                  partitions_source => {partition_count, Partitions},
-                  arguments => stream_arguments(Opts)},
+    Spec0 = maps:with([vhost, exchange_type], Opts),
+    Spec =
+        Spec0#{username => cli_acting_user(),
+               name => SuperStream,
+               partitions_source => {partition_count, Partitions},
+               arguments => stream_arguments(Opts)},
     create_super_stream(NodeName, Timeout, Spec);
 run([SuperStream],
     #{node := NodeName,
       timeout := Timeout,
       routing_keys := RoutingKeysStr} =
         Opts) ->
-    Spec0 = maps:with([vhost,
-                       exchange_type], Opts),
-    RoutingKeys =
-        [K || K <- string:lexemes(RoutingKeysStr, ", ")],
-    Spec = Spec0#{username => cli_acting_user(),
-                  name => SuperStream,
-                  partitions_source => {routing_keys, RoutingKeys},
-                  arguments => stream_arguments(Opts)},
+    Spec0 = maps:with([vhost, exchange_type], Opts),
+    RoutingKeys = [K || K <- string:lexemes(RoutingKeysStr, ", ")],
+    Spec =
+        Spec0#{username => cli_acting_user(),
+               name => SuperStream,
+               partitions_source => {routing_keys, RoutingKeys},
+               arguments => stream_arguments(Opts)},
     create_super_stream(NodeName, Timeout, Spec).
 
 stream_arguments(Opts) ->
     stream_arguments(#{}, Opts).
 
 %% Something strange, dialyzer infers that map_size/1 returns positive_integer()
 -dialyzer({no_match, stream_arguments/2}).
+
 stream_arguments(Acc, Arguments) when map_size(Arguments) =:= 0 ->
     Acc;
 stream_arguments(Acc, #{max_length_bytes := Value} = Arguments) ->
@@ -245,13 +245,12 @@ duration_to_seconds([{sign, _},
                      {seconds, S}]) ->
     Y * 365 * 86400 + M * 30 * 86400 + D * 86400 + H * 3600 + Mn * 60 + S.
 
-create_super_stream(NodeName,
-                    Timeout,
-                    Spec) ->
+create_super_stream(NodeName, Timeout, Spec) ->
     case rabbit_misc:rpc_call(NodeName,
                               rabbit_stream_manager,
                               create_super_stream,
-                              [Spec], Timeout)
+                              [Spec],
+                              Timeout)
     of
         ok ->
             {ok,

deps/rabbitmq_stream/src/rabbit_exchange_type_super_stream.erl

@@ -60,25 +60,41 @@ route(#exchange{name = Name},
             end
     end.
 
-info(_) -> [].
-info(_, _) -> [].
-validate(_X) -> ok.
+info(_) ->
+    [].
+
+info(_, _) ->
+    [].
+
+validate(_X) ->
+    ok.
 
 validate_binding(_X, #binding{key = K}) ->
     try
         %% just check the Key is an integer
         _ = binary_to_integer(K),
         ok
-    catch error:badarg ->
+    catch
+        error:badarg ->
             {error,
              {binding_invalid, "The binding key must be an integer: ~tp", [K]}}
     end.
 
-create(_Serial, _X) -> ok.
-delete(_Serial, _X) -> ok.
-policy_changed(_X1, _X2) -> ok.
-add_binding(_Serial, _X, _B) -> ok.
-remove_bindings(_Serial, _X, _Bs) -> ok.
+create(_Serial, _X) ->
+    ok.
+
+delete(_Serial, _X) ->
+    ok.
+
+policy_changed(_X1, _X2) ->
+    ok.
+
+add_binding(_Serial, _X, _B) ->
+    ok.
+
+remove_bindings(_Serial, _X, _Bs) ->
+    ok.
+
 assert_args_equivalence(X, Args) ->
     rabbit_exchange:assert_args_equivalence(X, Args).
 

deps/rabbitmq_stream/src/rabbit_stream.erl

@@ -40,8 +40,7 @@
 
 start(_Type, _Args) ->
     rabbit_stream_metrics:init(),
-    rabbit_global_counters:init([{protocol, stream}],
-                                ?PROTOCOL_COUNTERS),
+    rabbit_global_counters:init([{protocol, stream}], ?PROTOCOL_COUNTERS),
     rabbit_global_counters:init([{protocol, stream},
                                  {queue_type, ?STREAM_QUEUE_TYPE}]),
     rabbit_stream_sup:start_link().
@@ -130,8 +129,10 @@ kill_connection(ConnectionName) ->
                          {ConnectionPid,
                           #{<<"connection_name">> := ConnectionNameBin}} ->
                              exit(ConnectionPid, kill);
-                         {ConnectionPid, _ClientProperties} -> ok
-                     after 1000 -> ok
+                         {ConnectionPid, _ClientProperties} ->
+                             ok
+                     after 1000 ->
+                         ok
                      end
                   end,
                   pg_local:get_members(rabbit_stream_connections)).

deps/rabbitmq_stream/src/rabbit_stream_manager.erl

@@ -30,9 +30,6 @@
          create/4,
          delete/3,
          create_super_stream/1,
-         %% obsolete use create_super_stream/1
-         % create_super_stream/3,
-         %% obsolete use create_super_stream/1
          create_super_stream/6,
          delete_super_stream/3,
          lookup_leader/2,
@@ -43,17 +40,22 @@
          partitions/2,
          partition_index/3]).
 
+         %% obsolete use create_super_stream/1
+         % create_super_stream/3,
+         %% obsolete use create_super_stream/1
+
 -record(state, {configuration}).
 
 -type super_stream_spec() ::
-#{name := binary(),
-  vhost := binary(),
-  username := binary(),
-  partitions_source := {partition_count, pos_integer()} |
-                       {routing_keys, [binary()]},
-  arguments => map(),
-  exchange_type => binary() %<<"direct">> | <<"x-super-stream">>,
- }.
+    #{name := binary(),
+      vhost := binary(),
+      username := binary(),
+      partitions_source :=
+          {partition_count, pos_integer()} | {routing_keys, [binary()]},
+      arguments => map(),
+      exchange_type => binary()}.
+
+                            %<<"direct">> | <<"x-super-stream">>,
 
 start_link(Conf) ->
     gen_server:start_link({local, ?MODULE}, ?MODULE, [Conf], []).
@@ -88,19 +90,16 @@ create_super_stream(VirtualHost,
                     Arguments,
                     RoutingKeys,
                     Username) ->
-    Options = #{partitions => Partitions,
-                args => Arguments,
-                routing_keys => RoutingKeys,
-                username => Username},
+    Options =
+        #{partitions => Partitions,
+          args => Arguments,
+          routing_keys => RoutingKeys,
+          username => Username},
     create_super_stream(VirtualHost, Name, Options).
 
-
-
--spec create_super_stream(binary(),
-                          binary(),
-                          map()) -> ok | {error, term()}.
-create_super_stream(VirtualHost,
-                    Name,
+-spec create_super_stream(binary(), binary(), map()) ->
+                             ok | {error, term()}.
+create_super_stream(VirtualHost, Name,
                     #{username := Username} = Options) ->
     Type = maps:get(exchange_type, Options, <<"direct">>),
     Partitions = maps:get(partitions, Options, []),
@@ -116,30 +115,29 @@ create_super_stream(VirtualHost,
                      RoutingKeys,
                      Username}).
 
-
 -spec create_super_stream(super_stream_spec()) ->
-    ok | {error, term()}.
+                             ok | {error, term()}.
 create_super_stream(#{exchange_type := <<"x-super-stream">>,
                       partitions_source := {routing_keys, _}}) ->
     {error, unsupported_specification};
 create_super_stream(#{name := Name,
                       vhost := VHost,
                       username := Username,
-                      partitions_source := PartitionSource} = Spec) ->
+                      partitions_source := PartitionSource} =
+                        Spec) ->
     Type = maps:get(exchange_type, Spec, <<"direct">>),
     Arguments = maps:get(arguments, Spec, #{}),
     {Partitions, RoutingKeys} =
         case PartitionSource of
             {partition_count, Count} ->
-                Streams = [rabbit_stream_utils:partition_name(Name, K)
-                           || K <- lists:seq(0, Count - 1)],
-                Keys = [integer_to_binary(K) ||
-                               K <- lists:seq(0, Count - 1)],
+                Streams =
+                    [rabbit_stream_utils:partition_name(Name, K)
+                     || K <- lists:seq(0, Count - 1)],
+                Keys = [integer_to_binary(K) || K <- lists:seq(0, Count - 1)],
                 {Streams, Keys};
             {routing_keys, Keys} ->
                 Streams =
-                    [rabbit_stream_utils:partition_name(Name, K)
-                     || K <- Keys],
+                    [rabbit_stream_utils:partition_name(Name, K) || K <- Keys],
                 {Streams, Keys}
         end,
 
@@ -825,15 +823,16 @@ add_super_stream_binding(VirtualHost,
     ExchangeName = rabbit_misc:r(VirtualHost, exchange, ExchangeNameBin),
     QueueName = rabbit_misc:r(VirtualHost, queue, QueueNameBin),
     Pid = self(),
-    Arguments = case ExchangeType of
-                    <<"direct">> ->
-                        rabbit_misc:set_table_value([],
-                                                    <<"x-stream-partition-order">>,
-                                                    long,
-                                                    Order);
-                    _ ->
-                        []
-                end,
+    Arguments =
+        case ExchangeType of
+            <<"direct">> ->
+                rabbit_misc:set_table_value([],
+                                            <<"x-stream-partition-order">>,
+                                            long,
+                                            Order);
+            _ ->
+                []
+        end,
     case rabbit_binding:add(#binding{source = ExchangeName,
                                      destination = QueueName,
                                      key = RoutingKey,
@@ -952,22 +951,24 @@ is_resource_stream_queue(#resource{kind = queue} = Resource) ->
 is_resource_stream_queue(_) ->
     false.
 
-partition_index(#exchange{name = ExchangeName,
-                          type = ExchangeType} = Exchange, Stream) ->
+partition_index(#exchange{name = ExchangeName, type = ExchangeType} =
+                    Exchange,
+                Stream) ->
     UnorderedBindings =
         [Binding
          || Binding = #binding{destination = #resource{name = Q} = D}
-            <- rabbit_binding:list_for_source(ExchangeName),
+                <- rabbit_binding:list_for_source(ExchangeName),
             is_resource_stream_queue(D), Q == Stream],
     case UnorderedBindings of
         [] ->
             {error, stream_not_found};
         _ when ExchangeType =:= direct ->
-            Bindings = rabbit_stream_utils:sort_partitions(Exchange, UnorderedBindings),
+            Bindings =
+                rabbit_stream_utils:sort_partitions(Exchange,
+                                                    UnorderedBindings),
             Binding = lists:nth(1, Bindings),
             #binding{args = Args} = Binding,
-            case rabbit_misc:table_lookup(Args,
-                                          <<"x-stream-partition-order">>)
+            case rabbit_misc:table_lookup(Args, <<"x-stream-partition-order">>)
             of
                 {_, Order} ->
                     Index = rabbit_data_coercion:to_integer(Order),