anchor.diff:  %%
   
   %% @private
  -ensure_headers(Headers, Frame) -> ensure_headers(Headers, Frame, []).
  -
  -%% @private
  -ensure_headers([], _Frame, Acc) ->
  +with_headers(Headers, Frame, State) ->
  +    with_headers(Headers, Frame, [], State).
  +
  +%% @private
  +with_headers([], _Frame, Acc, _State) ->
       {ok, lists:reverse(Acc)};
  -ensure_headers([H|T], Frame, Acc) ->
  +with_headers([H|T], Frame, Acc, State) ->
       case restomp:header(Frame, H) of
  -        {ok, Value} -> ensure_headers(T, Frame, [Value|Acc]);
  -        not_found   -> {error, missing_header, "Missing header: " ++ H}
  +        {ok, Value} -> with_headers(T, Frame, [Value|Acc]);
  +        not_found   -> send_error(missing_header, "Missing header " ++ H, State)
       end.
   
   %% @private
  
positive.code: send_error(Msg, State) -> send_error(Msg, [], State).
  
  %% @private
  send_error(Msg, Details, State) ->
      Headers = [{"message", atom_to_list(Msg)},
                 ?CONTENT_TEXT,
                 ?VERSION_HEADER],
      case send_frame("ERROR", Headers, Details, State) of
          {noreply, NewState} -> {stop, Msg, NewState};
          Error               -> Error
      end.
  
  %% @private
  send_frame(Cmd, Headers, Body, State) ->
      send_frame(#stomp_frame{command = Cmd, headers = Headers, body = Body}, State).
  
  -spec send_frame(restomp:frame(), inet:socket()) -> #s{}.
  %% @private
  
negative.code: %%
  
  %% @private
  connect(_Frame, State) ->
      Session = riak_core_util:unique_id_62(),
      Headers = [{"session", Session},
                 ?SERVER_HEADER,
                 ?VERSION_HEADER],
      send_frame("CONNECTED", Headers, "", State#s{session = Session}).
  
  %% @private
  send(Frame = #stomp_frame{command = _Cmd, headers = _Headers, body = _Body},
       State) ->
      %% Check destination header exists
      case with_headers(["destination"], Frame, State) of
          {ok, [Dest]} ->
              case split_destination(Dest) of
                  {topic, _Topic}  -> {noreply, State};
                  _Other           -> send_error(invalid_destination, State)
              end;
          Error ->
              Error
      end.
  
  %% @private
  subscribe(Frame, State) ->
      case with_headers(["ack", "id", "destination"], Frame, State) of
          {ok, ["client", _Id, Dest]} ->
              case split_destination(Dest) of
                  {queue, _Topic, _Queue} -> {noreply, State};
  