anchor.diff:      gen_fsm:start_link(?MODULE, #state{channel_id = ChannelId,
                                          owner = Owner}, []).
   
  -active_once(Buffer) ->
  -    Buffer ! {active, once}.
  +-spec set_active(pid() | atom(), logplex_drain_buffer:framing_fun()) -> any().
  +set_active(Buffer, Fun) when is_function(Fun, 1) ->
  +    Buffer ! {active, Fun}.
   
   %% ------------------------------------------------------------------
   %% gen_fsm Function Definitions
  
positive.code:             send(S)
      end.
  
  send(S = #state{owner = Owner, buf = Buf}) ->
      {Data, _Count, NewBuf} = logplex_drain_buffer:to_pkts(Buf, 4096, pkt_fmt()),
      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.
  
negative.code: %% ------------------------------------------------------------------
  
  -export([start_link/2
           ,set_active/2
          ]).
  
  -export([active/2,
  