anchor.diff:              end;
           {ok, [_Ack, _Id, _Dest]} ->
               send_error(must_set_ack_client, State);
  -        {error, Msg, Detail} ->
  -            send_error(Msg, Detail, State)
  +        Error ->
  +            Error
       end.
   
   %% @private
   unsubscribe(Frame, State) ->
  -    case restomp:header(Frame, "id") of
  -        {ok, _Id} ->
  +    case with_headers(["id"], Frame, State) of
  +        {ok, [_Id]} ->
               {noreply, State};
  -        not_found ->
  -            send_error(no_id, State)
  +        Error ->
  +            Error
       end.
   
   %% @private
   ack(Frame, State) ->
  -    case ensure_headers(["message-id", "subscription"], Frame) of
  +    case with_headers(["message-id", "subscription"], Frame, State) of
           {ok, [_Id, _Sub]} ->
               {noreply, State};
  -        {error, Msg, Detail} ->
  -            send_error(Msg, Detail, State)
  +        Error ->
  +            Error
       end.
   
   %% @private
   nack(Frame, State) ->
  -    case ensure_headers(["message-id", "subscription"], Frame) of
  +    case with_headers(["message-id", "subscription"], Frame, State) of
           {ok, [_Id, _Sub]} ->
               {noreply, State};
  -        {error, Msg, Detail} ->
  -            send_error(Msg, Detail, State)
  +        Error ->
  +            Error
       end.
   
   %% @private
  
positive.code: %%
  
  %% @private
  connect(_Frame, State = #s{session = undefined}) ->
      Session = riak_core_util:unique_id_62(),
      Headers = [{"session", Session},
                 ?SERVER_HEADER,
                 ?VERSION_HEADER],
      send_frame("CONNECTED", Headers, "", State#s{session = Session});
  connect(_Frame, State) ->
      send_error(already_connected, State).
  
  %% @private
  send(Frame = #stomp_frame{command = _Cmd, headers = _Headers, body = _Body},
       State) ->
      %% Check destination header exists
      case restomp:header(Frame, "destination") of
          {ok, Dest} ->
              case split_destination(Dest) of
                  {topic, _Topic}  -> {noreply, State};
                  _Other           -> send_error(invalid_destination, State)
              end;
          not_found ->
              send_error(no_destination, State)
      end.
  
  %% @private
  subscribe(Frame, State) ->
      case ensure_headers(["ack", "id", "destination"], Frame) of
          {ok, ["client", _Id, Dest]} ->
              case split_destination(Dest) of
                  {queue, _Topic, _Queue} -> {noreply, State};
  
negative.code: send_error(Msg, State) -> send_error(Msg, [], State).
  
  %% @private
  send_error(Msg, Body, State) ->
      Headers = [{"message", atom_to_list(Msg)},
                 ?CONTENT_TEXT,
                 ?VERSION_HEADER],
      case send_frame("ERROR", Headers, [Body], 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
  