anchor.diff:              send(S)
       end.
   
  -send(S = #state{owner = Owner, buf = Buf}) ->
  -    {Data, _Count, NewBuf} = logplex_drain_buffer:to_pkts(Buf, 4096, pkt_fmt()),
  +send(S = #state{owner = Owner, buf = Buf,
  +                active_fun = Fun}) ->
  +    {Data, _Count, NewBuf} = logplex_drain_buffer:to_pkts(Buf, 4096, Fun),
       Owner ! {logplex_tail_data, self(), Data},
       {next_state, passive,
        S#state{buf=NewBuf}}.
  -
  -pkt_fmt() ->
  -    fun ({loss_indication, N, When}) ->
  -            M = #msg{time = logplex_syslog_utils:datetime(now),
  -                     source = <<"logplex">>,
  -                     ps = <<"1">>,
  -                     content=io_lib:format("Tail buffer overflowed. "
  -                                           "~p messages lost since ~s.",
  -                                           [N, logplex_syslog_utils:datetime(When)]
  -                                          )},
  -            {frame, logplex_utils:format(M)};
  -        ({msg, Msg}) ->
  -            M = logplex_utils:parse_msg(Msg),
  -            {frame, logplex_utils:format(M)}
  -    end.
  
positive.code: 
  -record(state, {buf = logplex_drain_buffer:new() :: logplex_drain_buffer:buf(),
                  channel_id :: logplex_channel:id(),
                  owner :: pid()}).
  
  %-type pstates() :: 'passive' | 'active'.
  
  
negative.code: %% ------------------------------------------------------------------
  
  -export([start_link/2
           ,set_active/2
          ]).
  
  -export([active/2,
  