anchor.diff:      logplex_realtime:incr(message_received),
       logplex_queue:in(logplex_work_queue, Msg);
   process_msg({malformed, Msg}) ->
  -    io:format("[~p] malformed_syslog_message=\"~p\"~n",
  -              [?MODULE, Msg]),
  +    ?WARN("err=malformed_syslog_message data=\"~p\"~n",
  +          [Msg]),
       logplex_stats:incr(message_received_malformed).
  
positive.code:          terminate/2,
           code_change/3]).
  
  -record(state, {sock, buffer = <<>>}).
  
  %%====================================================================
  %% API functions
  
negative.code: %% gen_server callbacks
  %%====================================================================
  init([]) ->
      {ok, #state{}}.
  
  handle_call(Msg, _From, State) ->
      ?WARN("err=unexpected_call data=~p", [Msg]),
      {noreply, State}.
  
  handle_cast({set_socket, CSock}, State) ->
      case inet:peername(CSock) of
          {ok, PeerName = {PHost, PPort}} ->
              {ok, {Host, Port}} = inet:sockname(CSock),
              ?INFO("at=new_connection peer=~s local=~s",
                    [logplex_logging:dest(PHost, PPort),
                     logplex_logging:dest(Host, Port)]),
              inet:setopts(CSock, [{active, once}]),
              {noreply, State#state{sock=CSock,
                                    peername=PeerName}};
          {error, Reason} ->
              ?INFO("at=new_connection err=\"~p\"",
                    [Reason]),
              {stop, normal, State}
      end;
  
  handle_cast(Msg, State) ->
      ?WARN("err=unexpected_cast data=~p", [Msg]),
      {noreply, State}.
  
  handle_info({tcp, Sock, Packet},
              #state{sock=Sock, buffer=Buffer}=State) ->
      {Result, Msgs, NewBuf} = syslog_parser:push(Packet, Buffer),
      case Result of
          ok -> ok;
          {error, Err} ->
              ?INFO("[~p] event=parse_error, txt=\"~p\"",
                        [?MODULE, Err]),
              ok
      end,
  