anchor.diff:  %%% riak_core_broadcast_handler callbacks
   %%%===================================================================
   
  +-type clock() :: vclock:vclock().
  +
  +-type broadcast_message() :: #broadcast{}.
  +-type broadcast_id() :: id().
  +-type broadcast_clock() :: clock().
  +-type broadcast_payload() :: {id(), type(), metadata(), value()}.
  +
  +%% @doc Returns from the broadcast message the identifier and the payload.
  +-spec broadcast_data(broadcast_message()) -> {{broadcast_id(), broadcast_clock()}, broadcast_payload()}.
  +broadcast_data(#broadcast{id=Id, type=Type, clock=Clock, metadata=Metadata, value=Value}) ->
  +    {{Id, Clock}, {Id, Type, Metadata, Value}}.
  +
   %% @todo doc
  -%% @todo spec
  -broadcast_data(#broadcast{id=Id, type=Type, value=Value}=Msg) ->
  -    lager:warning("id: ~p, msg: ~p", [Id, Msg]),
  -    {{Id, Type, Value}, {Id, Type, Value}}.
  -
  -%% @todo doc
  -%% @todo spec
  -merge({Id, Type, Value}, {Id, Type, Value}) ->
  -    lager:warning("id: ~p, type: ~p value: ~p", [Id, Type, Value]),
  -
  -    case is_stale({Id, Type, Value}) of
  +-spec merge({broadcast_id(), broadcast_clock()}, broadcast_payload()) -> boolean().
  +merge({Id, Clock}, {Id, Type, Metadata, Value}) ->
  +    lager:info("id: ~p, clock: ~p", [Id, Clock]),
  +
  +    case is_stale({Id, Clock}) of
           true ->
               false;
           false ->
  -            {ok, _} = local_bind(Id, Type, Value),
  +            {ok, _} = local_bind(Id, Type, Metadata, Value),
               true
       end.
   
  -%% @todo doc
  -%% @todo spec
  -is_stale({Id, Type, Value}) ->
  -    lager:warning("id: ~p, type: ~p value: ~p", [Id, Type, Value]),
  -
  -    Result = case read(Id, undefined) of
  -        {ok, {_I, T, _M, V}} ->
  -            not lasp_lattice:is_lattice_strict_inflation(T, V, Value);
  -        _ ->
  -            %% We don't know about the value yet; not stale.
  -            false
  -    end,
  -
  -    lager:warning("returning: ~p", [Result]),
  -
  +%% @doc Use the clock on the object to determine if this message is
  +%% stale or not.
  +-spec is_stale({broadcast_id(), broadcast_clock()}) -> boolean().
  +is_stale({Id, Clock}) ->
  +    lager:info("id: ~p, clock: ~p", [Id, Clock]),
  +    gen_server:call(?MODULE, {is_stale, Id, Clock}, infinity).
  +
  +%% @doc Given a message identifier and a clock, return a given message.
  +-spec graft({broadcast_id(), broadcast_clock()}) ->
  +    stale | {ok, broadcast_payload()} | {error, term()}.
  +graft({Id, Clock}) ->
  +    Result = gen_server:call(?MODULE, {graft, Id, Clock}, infinity),
  +    lager:info("id: ~p, clock: ~p, result: ~p", [Id, Clock, Result]),
       Result.
  -
  -%% @todo doc
  -%% @todo spec
  -graft(Id) ->
  -    lager:warning("id: ~p", [Id]),
  -    {error, not_implemented}.
   
   %% @todo doc
   %% @todo spec
  
positive.code: %%      the value to bind.
  %%
  -spec bind(id(), value()) -> {ok, var()}.
  bind(Id, Value) ->
      {ok, {Id, Type, Metadata, Value}} = gen_server:call(?MODULE, {bind, Id, Value}, infinity),
      broadcast({Id, Type, Metadata, Value}),
      {ok, {Id, Type, Metadata, Value}}.
  
  
negative.code: 
  -define(BLOCKING, fun() -> {noreply, State} end).
  
  %% Clock mutation macros.
  -define(CLOCK_INIT, fun(Metadata) ->
              VClock = vclock:increment(Actor, vclock:fresh()),
              orddict:store(clock, VClock, Metadata)
      end).
  
  -define(CLOCK_MERG, fun(Metadata) ->
              vclock:merge([orddict:fetch(clock, Metadata0), orddict:fetch(clock, Metadata)])
      end).
  
  %%%===================================================================
  %%% API
  %%%===================================================================
  