Correct types when using msg store clients with module tag

variable queue state stores msg store clients with message store module tags

Author: hairyhum

src/rabbit_variable_queue.erl

@@ -1261,18 +1261,19 @@ trim_msg_status(MsgStatus) ->
         queue_index -> MsgStatus
     end.
 
-with_msg_store_state({MSCStateP, MSCStateT},  true, Fun) ->
-    {Result, MSCStateP1} = Fun(MSCStateP),
-    {Result, {MSCStateP1, MSCStateT}};
-with_msg_store_state({MSCStateP, MSCStateT}, false, Fun) ->
-    {Result, MSCStateT1} = Fun(MSCStateT),
-    {Result, {MSCStateP, MSCStateT1}}.
+with_msg_store_state({{Mod, MSCStatePInternal}, MSCStateT},  true, Fun) ->
+    {Result, MSCStatePInternal1} = Fun(Mod, MSCStatePInternal),
+    {Result, {{Mod, MSCStatePInternal1}, MSCStateT}};
+with_msg_store_state({MSCStateP, {Mod, MSCStateTInternal}}, false, Fun) ->
+    {Result, MSCStateTInternal1} = Fun(Mod, MSCStateTInternal),
+    {Result, {MSCStateP, {Mod, MSCStateTInternal1}}}.
 
 with_immutable_msg_store_state(MSCState, IsPersistent, Fun) ->
-    {Res, MSCState} = with_msg_store_state(MSCState, IsPersistent,
-                                           fun (MSCState1) ->
-                                                   {Fun(MSCState1), MSCState1}
-                                           end),
+    {Res, MSCState} = with_msg_store_state(
+        MSCState, IsPersistent,
+        fun (Mod, MSCState1) ->
+            {Fun(Mod, MSCState1), MSCState1}
+        end),
     Res.
 
 msg_store_client_init(MsgStore, MsgOnDiskFun, Callback, VHost) ->
@@ -1290,28 +1291,30 @@ msg_store_client_init(MsgStore, Ref, MsgOnDiskFun, Callback, VHost) ->
 msg_store_write(MSCState, IsPersistent, MsgId, Msg) ->
     with_immutable_msg_store_state(
       MSCState, IsPersistent,
-      fun ({Mod, MSCState1}) ->
-              Mod:write_flow(MsgId, Msg, MSCState1)
+      fun (Mod, MSCStateInternal) ->
+          Mod:write_flow(MsgId, Msg, MSCStateInternal)
       end).
 
 msg_store_read(MSCState, IsPersistent, MsgId) ->
     with_msg_store_state(
       MSCState, IsPersistent,
-      fun ({Mod, MSCState1}) ->
-              Mod:read(MsgId, MSCState1)
+      fun (Mod, MSCStateInternal) ->
+          Mod:read(MsgId, MSCStateInternal)
       end).
 
 msg_store_remove(MSCState, IsPersistent, MsgIds) ->
     with_immutable_msg_store_state(
       MSCState, IsPersistent,
-      fun ({Mod, MSCState1}) ->
-              Mod:remove(MsgIds, MSCState1)
+      fun (Mod, MSCStateInternal) ->
+              Mod:remove(MsgIds, MSCStateInternal)
       end).
 
 msg_store_close_fds(MSCState, IsPersistent) ->
     with_msg_store_state(
       MSCState, IsPersistent,
-      fun ({Mod, MSCState1}) -> Mod:close_all_indicated(MSCState1) end).
+      fun (Mod, MSCStateInternal) ->
+          Mod:close_all_indicated(MSCStateInternal)
+      end).
 
 msg_store_close_fds_fun(IsPersistent) ->
     fun (?MODULE, State = #vqstate { msg_store_clients = MSCState }) ->
@@ -2938,6 +2941,7 @@ start_new_store_sup() ->
     ok = rabbit_sup:start_child(?PERSISTENT_MSG_STORE_SUP,
                                 rabbit_msg_store_vhost_sup,
                                 [?PERSISTENT_MSG_STORE_SUP,
+                                 rabbit_msg_store,
                                  undefined,  {fun (ok) -> finished end, ok}]),
     ?PERSISTENT_MSG_STORE_SUP.
 

test/backing_queue_SUITE.erl

@@ -376,10 +376,11 @@ on_disk_stop(Pid) ->
 
 msg_store_client_init_capture(MsgStore, Ref) ->
     Pid = spawn(fun on_disk_capture/0),
-    {Pid, rabbit_msg_store_vhost_sup:client_init(
+    {rabbit_msg_store, MSCState} = rabbit_msg_store_vhost_sup:client_init(
             MsgStore, Ref, fun (MsgIds, _ActionTaken) ->
                                    Pid ! {on_disk, MsgIds}
-                           end, undefined, <<"/">>)}.
+                           end, undefined, <<"/">>),
+    {Pid, MSCState}.
 
 msg_store_contains(Atom, MsgIds, MSCState) ->
     Atom = lists:foldl(
@@ -456,7 +457,7 @@ test_msg_store_confirm_timer() ->
     Ref = rabbit_guid:gen(),
     MsgId  = msg_id_bin(1),
     Self = self(),
-    MSCState = rabbit_msg_store_vhost_sup:client_init(
+    {rabbit_msg_store, MSCState} = rabbit_msg_store_vhost_sup:client_init(
                  ?PERSISTENT_MSG_STORE, Ref,
                  fun (MsgIds, _ActionTaken) ->
                          case gb_sets:is_member(MsgId, MsgIds) of
@@ -1337,7 +1338,8 @@ nop(_) -> ok.
 nop(_, _) -> ok.
 
 msg_store_client_init(MsgStore, Ref) ->
-    rabbit_msg_store_vhost_sup:client_init(MsgStore, Ref, undefined, undefined, <<"/">>).
+    {rabbit_msg_store, MSCState} = rabbit_msg_store_vhost_sup:client_init(MsgStore, Ref, undefined, undefined, <<"/">>),
+    MSCState.
 
 variable_queue_init(Q, Recover) ->
     rabbit_variable_queue:init(