Allow message id override on merge

When performing a broadcast merge it might be the
case that the application wants to return a new message
id that results from the merge and have that broadcast instead
of the originally received message id. Support this by handling
an extra return tuple from the callback.

Author: lrascao

src/plumtree_broadcast.erl

@@ -380,6 +380,16 @@ handle_broadcast(false, _MessageId, _Message, Mod, _Round, Root, From, State) ->
     State1 = add_lazy(From, Root, State),
     _ = send({prune, Root, myself()}, Mod, From),
     State1;
+%% The next clause is designed to allow the callback to override the message id
+%% after the merge, suppose node A eager pushed a change to node B, node B would then lazy
+%% push it to node C, at this point the message id being sent to C is the one that originated
+%% from A. Concurrently B takes an update that subsumes the previous one, now node C receives the
+%% lazy push and hasn't seen this message and asks B to graft it, if C now sends the message id
+%% that it got from A, node B will not answer the graft since it is deemed stale.
+%% With this extra clause the callback is able to return a new message id that resulted from
+%% the merge and have that be propagated.
+handle_broadcast({true, MessageId}, _OldMessageId, Message, Mod, Round, Root, From, State) ->
+    handle_broadcast(true, MessageId, Message, Mod, Round, Root, From, State);
 handle_broadcast(true, MessageId, Message, Mod, Round, Root, From, State) -> %% valid msg
     %% remove sender from lazy and set as eager
     plumtree_util:log(debug, "moving peer ~p from lazy to eager on tree rooted at ~p",

src/plumtree_broadcast_handler.erl

@@ -24,7 +24,8 @@
 
 %% Given the message id and payload, merge the message in the local state.
 %% If the message has already been received return `false', otherwise return `true'
--callback merge(any(), any()) -> boolean().
+%% If a new message id is to be propagated after the merge return `{true, MessageId}`
+-callback merge(any(), any()) -> boolean() | {true, any()}.
 
 %% Return true if the message (given the message id) has already been received.
 %% `false' otherwise

test/plumtree_test_broadcast_handler.erl

@@ -122,7 +122,7 @@ broadcast_data({Key, Object}) ->
 
 %% Given the message id and payload, merge the message in the local state.
 %% If the message has already been received return `false', otherwise return `true'
--spec merge(any(), any()) -> boolean().
+-spec merge(any(), any()) -> boolean() | {true, any()}.
 merge({Key, _Context} = MsgId, RemoteObj) ->
     Existing = dbread(Key),
     lager:info("merge msg id ~p, remote object: ~p, existing object: ~p",
@@ -133,7 +133,7 @@ merge({Key, _Context} = MsgId, RemoteObj) ->
             lager:info("merge object has ben reconciled to ~p",
                       [Reconciled]),
             dbwrite(Key, Reconciled),
-            true
+            {true, {Key, plumtree_test_object:context(Reconciled)}}
     end.
 
 %% Return true if the message (given the message id) has already been received.