anchor.diff:  init(Parent) ->
       io:format("init ~p~n", [?MODULE]),
       {ok, RE} = re:compile("^<\\d+>\\S+ \\S+ \\S+ (t[.]\\S+) "),
  -    RedisBuffers = [{logplex_queue:get(Pid, redis_url), Pid} || {_Id, Pid, worker, _Modules} <- supervisor:which_children(logplex_redis_buffer_sup)],
  +    RedisBuffers = [{logplex_queue:get(Pid, redis_url), Pid}
  +                    || {_Id, Pid, worker, _Modules} <- supervisor:which_children(logplex_redis_buffer_sup)],
       {ok, Map, Interval} = redis_shard:generate_map_and_interval(lists:sort(RedisBuffers)),
       proc_lib:init_ack(Parent, {ok, self()}),
       loop(#state{regexp=RE, map=Map, interval=Interval}).
   
  -loop(#state{regexp=RE, map=Map, interval=Interval, drain_accepting=DrainAccepting}=State) ->
  -    DrainAccepting1 =
  -        receive
  -            {logplex_drain_buffer, stop_accepting} ->
  -                io:format("logplex_worker event=producer_callback action=stop_accepting~n"),
  -                false;
  -            {logplex_drain_buffer, start_accepting} ->
  -                io:format("logplex_worker event=producer_callback action=start_accepting~n"),
  -                true
  -        after 0 -> DrainAccepting
  -        end,
  +loop(#state{regexp=RE, map=Map, interval=Interval}=State) ->
       case catch logplex_queue:out(logplex_work_queue) of
           timeout ->
               ok;
  
positive.code:         {1, [Msg]} ->
              case re:run(Msg, RE, [{capture, all_but_first, binary}]) of
                  {match, [Token]} ->
                      route(Token, Map, Interval, Msg, DrainAccepting1);
                  _ ->
                      ok
              end
      end,
      ?MODULE:loop(State#state{drain_accepting=DrainAccepting1}).
  
  route(Token, Map, Interval, Msg, DrainAccepting) when is_binary(Token), is_binary(Msg) ->
      case logplex_token:lookup(Token) of
          #token{channel_id=ChannelId, name=TokenName, app_id=AppId, drains=Drains} ->
              BufferPid = logplex_shard:lookup(integer_to_list(ChannelId), Map, Interval),
              logplex_stats:incr(logplex_stats_channels, {message_processed, AppId, ChannelId}),
              Msg1 = iolist_to_binary(re:replace(Msg, Token, TokenName)),
              process_drains(AppId, ChannelId, Drains, Msg1, DrainAccepting),
              process_tails(ChannelId, Msg1),
              process_msg(ChannelId, BufferPid, Msg1);
          _ ->
              ok
      end.
  
  process_drains(_AppId, _ChannelId, _Drains, _Msg, false = _DrainAccepting) ->
      logplex_stats:incr("drain_buffer_dropped"),
      logplex_realtime:incr("drain_buffer_dropped"),
      ok;
  
  process_drains(AppId, ChannelId, Drains, Msg, true = _DrainAccepting) ->
      [logplex_queue:in(logplex_drain_buffer, {TcpDrain, AppId, ChannelId, Token, Host, Port, Msg}) ||
          #drain{token=Token, resolved_host=Host, port=Port, tcp=TcpDrain} <- Drains],
      ok.
  
  process_tails(ChannelId, Msg) ->
  
negative.code: 
  -include_lib("logplex.hrl").
  
  -record(state, {regexp, map, interval}).
  
  %% API functions
  start_link(_QueuePid) ->
  