WIP. fixing tests

Author: Daniil Fedotov

src/rabbit_variable_queue.erl

@@ -1291,27 +1291,27 @@ msg_store_write(MSCState, IsPersistent, MsgId, Msg) ->
     with_immutable_msg_store_state(
       MSCState, IsPersistent,
       fun ({Mod, MSCState1}) ->
-              Mod:write_flow(MsgId, Msg, MSCState1)
+              {Mod, Mod:write_flow(MsgId, Msg, MSCState1)}
       end).
 
 msg_store_read(MSCState, IsPersistent, MsgId) ->
     with_msg_store_state(
       MSCState, IsPersistent,
       fun ({Mod, MSCState1}) ->
-              Mod:read(MsgId, MSCState1)
+              {Mod, Mod:read(MsgId, MSCState1)}
       end).
 
 msg_store_remove(MSCState, IsPersistent, MsgIds) ->
     with_immutable_msg_store_state(
       MSCState, IsPersistent,
       fun ({Mod, MSCState1}) ->
-              Mod:remove(MsgIds, MSCState1)
+              {Mod, Mod:remove(MsgIds, MSCState1)}
       end).
 
 msg_store_close_fds(MSCState, IsPersistent) ->
     with_msg_store_state(
       MSCState, IsPersistent,
-      fun ({Mod, MSCState1}) -> Mod:close_all_indicated(MSCState1) end).
+      fun ({Mod, MSCState1}) -> {Mod, Mod:close_all_indicated(MSCState1)} end).
 
 msg_store_close_fds_fun(IsPersistent) ->
     fun (?MODULE, State = #vqstate { msg_store_clients = MSCState }) ->
@@ -2938,6 +2938,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(