Refactoring for CLI-driven definition import

4 files changed, 27 insertions, 15 deletions

diff --git a/src/rabbit_auth_backend_internal.erl b/src/rabbit_auth_backend_internal.erl
index 8777774775..09392bf7b8 100644
--- a/src/rabbit_auth_backend_internal.erl
+++ b/src/rabbit_auth_backend_internal.erl
@@ -797,5 +797,5 @@ hashing_algorithm(User, Version) ->
                 <<"3.0.", _/binary>> -> rabbit_password_hashing_md5;
                 _                    -> rabbit_password:hashing_mod()
             end;
-        Alg       -> binary_to_atom(Alg, utf8)
+        Alg       -> rabbit_data_coercion:to_atom(Alg, utf8)
     end.
diff --git a/src/rabbit_definitions.erl b/src/rabbit_definitions.erl
index fc50232e2a..709bfc0a85 100644
--- a/src/rabbit_definitions.erl
+++ b/src/rabbit_definitions.erl
@@ -20,7 +20,7 @@
 %% automatic import on boot
 -export([maybe_load_definitions/0, maybe_load_definitions_from/2]).
 %% import
--export([import_definitions/1, import_definitions/2,
+-export([import_raw/1, import_raw/2, import_parsed/1, import_parsed/2,
          apply_defs/2, apply_defs/3, apply_defs/4, apply_defs/5]).
 %% export
 -export([all_definitions/0]).
@@ -45,22 +45,33 @@ maybe_load_core_definitions() ->
 maybe_load_management_definitions() ->
     maybe_load_definitions(rabbitmq_management, load_definitions).
 
--spec import_definitions(Body :: binary() | iolist()) -> ok | {error, term()}.
-import_definitions(Body) ->
+-spec import_raw(Body :: binary() | iolist()) -> ok | {error, term()}.
+import_raw(Body) ->
     rabbit_log:info("Asked to import definitions. Acting user: ~s", [?INTERNAL_USER]),
     case decode([], Body) of
         {error, E}   -> {error, E};
         {ok, _, Map} -> apply_defs(Map, ?INTERNAL_USER)
     end.
 
--spec import_definitions(Body :: binary() | iolist(), VHost :: vhost:name()) -> ok | {error, term()}.
-import_definitions(Body, VHost) ->
+-spec import_raw(Body :: binary() | iolist(), VHost :: vhost:name()) -> ok | {error, term()}.
+import_raw(Body, VHost) ->
     rabbit_log:info("Asked to import definitions. Acting user: ~s", [?INTERNAL_USER]),
     case decode([], Body) of
         {error, E}   -> {error, E};
         {ok, _, Map} -> apply_defs(Map, ?INTERNAL_USER, fun() -> ok end, VHost)
     end.
 
+-spec import_parsed(Defs :: #{any() => any()}) -> ok | {error, term()}.
+import_parsed(Body0) ->
+    rabbit_log:info("Asked to import definitions. Acting user: ~s", [?INTERNAL_USER]),
+    Body = atomise_map_keys(Body0),
+    apply_defs(Body, ?INTERNAL_USER).
+
+-spec import_parsed(Defs :: #{any() => any()}, VHost :: vhost:name()) -> ok | {error, term()}.
+import_parsed(Body0, VHost) ->
+    rabbit_log:info("Asked to import definitions. Acting user: ~s", [?INTERNAL_USER]),
+    Body = atomise_map_keys(Body0),
+    apply_defs(Body, ?INTERNAL_USER, fun() -> ok end, VHost).
 
 -spec all_definitions() -> map().
 all_definitions() ->
@@ -132,7 +143,7 @@ load_definitions_from_file(File) ->
     case file:read_file(File) of
         {ok, Body} ->
             rabbit_log:info("Applying definitions from ~s", [File]),
-            import_definitions(Body);
+            import_raw(Body);
         {error, E} ->
             rabbit_log:error("Could not read definitions from ~s, Error: ~p", [File, E]),
             {error, {could_not_read_defs, {File, E}}}
@@ -157,13 +168,16 @@ decode(<<"">>) ->
 decode(Body) ->
     try
       Decoded = rabbit_json:decode(Body),
-      Normalised = maps:fold(fun(K, V, Acc) ->
-                     Acc#{binary_to_atom(K, utf8) => V}
-                   end, Decoded, Decoded),
+      Normalised = atomise_map_keys(Decoded),
       {ok, Normalised}
     catch error:_ -> {error, not_json}
     end.
 
+atomise_map_keys(Decoded) ->
+    maps:fold(fun(K, V, Acc) ->
+        Acc#{rabbit_data_coercion:to_atom(K, utf8) => V}
+              end, Decoded, Decoded).
+
 -spec apply_defs(Map :: #{atom() => any()}, ActingUser :: rabbit_types:username()) -> 'ok'.
 
 apply_defs(Map, ActingUser) ->
@@ -178,8 +192,6 @@ apply_defs(Map, ActingUser, VHost) when is_binary(VHost) ->
     apply_defs(Map, ActingUser, fun () -> ok end, VHost);
 
 apply_defs(Map, ActingUser, SuccessFun) when is_function(SuccessFun) ->
-    rabbit_log:info("Asked to import definitions for multiple virtual hosts. Acting user: ~p",
-                    [ActingUser]),
     Version = maps:get(rabbitmq_version, Map, maps:get(rabbit_version, Map, undefined)),
     try
         rabbit_log:info("Importing users..."),
@@ -427,7 +439,7 @@ add_binding_int(Binding, Source, Destination, ActingUser) ->
       ActingUser).
 
 dest_type(Binding) ->
-    list_to_atom(binary_to_list(maps:get(destination_type, Binding, undefined))).
+    rabbit_data_coercion:to_atom(maps:get(destination_type, Binding, undefined)).
 
 r(Type, Props) -> r(Type, name, Props).
 
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 86c11c4ad2..0d945929af 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -174,7 +174,7 @@ store_ram(X) ->
         (binary()) -> atom() | rabbit_types:connection_exit().
 
 check_type(TypeBin) ->
-    case rabbit_registry:binary_to_type(TypeBin) of
+    case rabbit_registry:binary_to_type(rabbit_data_coercion:to_binary(TypeBin)) of
         {error, not_found} ->
             rabbit_misc:protocol_error(
               command_invalid, "unknown exchange type '~s'", [TypeBin]);
diff --git a/src/rabbit_policy.erl b/src/rabbit_policy.erl
index c718c743f9..78157072e3 100644
--- a/src/rabbit_policy.erl
+++ b/src/rabbit_policy.erl
@@ -146,7 +146,7 @@ merge_policy_value(Name, PolicyVal, OpVal) ->
     end.
 
 policy_merge_strategy(Name) ->
-    case rabbit_registry:binary_to_type(Name) of
+    case rabbit_registry:binary_to_type(rabbit_data_coercion:to_binary(Name)) of
         {error, not_found} ->
             {error, not_found};
         T                  ->