fixed problem in vq:tx_commit, moved expiry recalculation to post_msg_store callback

1 files changed, 15 insertions, 14 deletions

diff --git a/src/rabbit_variable_queue.erl b/src/rabbit_variable_queue.erl
index 6dc24e80b9..202f2c99d2 100644
--- a/src/rabbit_variable_queue.erl
+++ b/src/rabbit_variable_queue.erl
@@ -613,22 +613,18 @@ tx_rollback(Txn, State = #vqstate { durable = IsDurable }) ->
 tx_commit(Txn, Fun, MsgPropsFun, State = #vqstate { durable = IsDurable }) ->
     #tx { pending_acks = AckTags, pending_messages = Pubs } = lookup_tx(Txn),
     erase_tx(Txn),
-    PubsOrdered = lists:foldl(
-                    fun ({Msg, MsgProps}, Acc) ->
-                            [{Msg, MsgPropsFun(MsgProps)} | Acc]
-                    end, [], Pubs),
     AckTags1 = lists:append(AckTags),
-    PersistentGuids = persistent_guids(PubsOrdered),
+    PersistentGuids = persistent_guids(Pubs),
     HasPersistentPubs = PersistentGuids =/= [],
     {AckTags1,
      a(case IsDurable andalso HasPersistentPubs of
            true  -> ok = rabbit_msg_store:sync(
                            ?PERSISTENT_MSG_STORE, PersistentGuids,
-                           msg_store_callback(PersistentGuids,
-                                              PubsOrdered, AckTags1, Fun)),
+                           msg_store_callback(PersistentGuids,Pubs, AckTags1, 
+                                              Fun, MsgPropsFun)),
                     State;
-           false -> tx_commit_post_msg_store(
-                      HasPersistentPubs, PubsOrdered, AckTags1, Fun, State)
+           false -> tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags1, 
+                                             Fun, MsgPropsFun, State)
        end)}.
 
 requeue(AckTags, MsgPropsFun, State) ->
@@ -901,11 +897,12 @@ update_rate(Now, Then, Count, {OThen, OCount}) ->
 %% Internal major helpers for Public API
 %%----------------------------------------------------------------------------
 
-msg_store_callback(PersistentGuids, Pubs, AckTags, Fun) ->
+msg_store_callback(PersistentGuids, Pubs, AckTags, Fun, MsgPropsFun) ->
     Self = self(),
     F = fun () -> rabbit_amqqueue:maybe_run_queue_via_backing_queue(
                     Self, fun (StateN) -> tx_commit_post_msg_store(
-                                            true, Pubs, AckTags, Fun, StateN)
+                                            true, Pubs, AckTags, 
+                                            Fun, MsgPropsFun, StateN)
                           end)
         end,
     fun () -> spawn(fun () -> ok = rabbit_misc:with_exit_handler(
@@ -916,7 +913,7 @@ msg_store_callback(PersistentGuids, Pubs, AckTags, Fun) ->
                     end)
     end.
 
-tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags, Fun,
+tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags, Fun, MsgPropsFun,
                          State = #vqstate {
                            on_sync     = OnSync = #sync {
                                            acks_persistent = SPAcks,
@@ -935,17 +932,21 @@ tx_commit_post_msg_store(HasPersistentPubs, Pubs, AckTags, Fun,
                                 end];
             false -> []
         end,
+    PubsOrdered = lists:foldl(
+                    fun ({Msg, MsgProps}, Acc) ->
+                            [{Msg, MsgPropsFun(MsgProps)} | Acc]
+                    end, [], Pubs),
     case IsDurable andalso (HasPersistentPubs orelse PersistentAcks =/= []) of
         true  -> State #vqstate { on_sync = #sync {
                                     acks_persistent = [PersistentAcks | SPAcks],
                                     acks_all        = [AckTags | SAcks],
-                                    pubs            = [Pubs | SPubs],
+                                    pubs            = [PubsOrdered | SPubs],
                                     funs            = [Fun | SFuns] }};
         false -> State1 = tx_commit_index(
                             State #vqstate { on_sync = #sync {
                                                acks_persistent = [],
                                                acks_all        = [AckTags],
-                                               pubs            = [Pubs],
+                                               pubs            = [PubsOrdered],
                                                funs            = [Fun] } }),
                  State1 #vqstate { on_sync = OnSync }
     end.