anchor.diff:          {vmq_writer, exit, Reason} ->
               teardown(State, Reason);
           {Proto, Socket, Data} ->
  -            NewParserState = process_bytes(SessionPid, <<Buffer/binary, Data/binary>>, ParserState),
  +            NewParserState = process_bytes(SessionPid, Data, ParserState),
               vmq_systree:incr_bytes_received(byte_size(Data)),
               loop(State#st{parser_state=NewParserState});
           {ProtoClosed, Socket} ->
  
positive.code:              session_monitor=MRef,
               proto_tag=proto_tag(Transport)}).
  
  loop(#st{buffer=Buffer, socket=Socket, transport=Transport,
              session=SessionPid, parser_state=ParserState,
              proto_tag={Proto, ProtoClosed, ProtoError}} = State) ->
      Transport:setopts(Socket, [{active, once}]),
  
negative.code: -spec init(_,_,atom() | tuple(),maybe_improper_list()) -> any().
  init(Ref, Socket, Transport, Opts) ->
      ok = ranch:accept_ack(Ref),
      case lists:keyfind(tune_buffer_size, 1, Opts) of
          {_, true} ->
              ok = tune_buffer_size(Transport, Socket);
          _ ->
              ok
      end,
      ok = Transport:setopts(Socket, [{nodelay, true},
                                      {packet, raw},
                                      {active, once}]),
  