anchor.diff:  -export([stream_init/0, stream_init/1]).
   -export([stream_start/1, stream_cancel/1]).
   -export([stream/1, stream/2, stream/3, stream_ack/1, stream_ack/2]).
  +-export([stream2/1, stream2/2, stream2/3, stream_last/1, stream_last/2]).
   
   -include_lib("rexi/include/rexi.hrl").
   
  
positive.code:         exit(timeout)
      end.
  
  %% @equiv stream_ack(Client, 1)
  stream_ack(Client) ->
      erlang:send(Client, {rexi_ack, 1}).
  
negative.code: %% internal functions %%
  
  cast_msg(Msg) -> {'$gen_cast', Msg}.
  
  maybe_init_stream(Timeout) ->
      case get(rexi_STREAM_INITED) of
          true ->
              ok;
          _ ->
              init_stream(Timeout)
      end.
  
  init_stream(Timeout) ->
      case sync_reply(rexi_STREAM_INIT, Timeout) of
          rexi_STREAM_START ->
              put(rexi_STREAM_INITED, true),
              ok;
          rexi_STREAM_CANCEL ->
              exit(normal);
          timeout ->
              exit(timeout);
          Else ->
              exit({invalid_stream_message, Else})
      end.
  
  maybe_wait(Limit, Timeout) ->
      case get(rexi_unacked) of
  