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