anchor.diff:      end,
       proplists:get_value(Key, Options, Default).
   
  +notify(Message, State) ->
  +    Subscriptions = State#?STATE.subscriptions,
  +    CallbackState = State#?STATE.callback_state,
  +    {MsgType, _} = DecodedMsg = of_msg_lib:decode(Message),
  +    Subscribers = ets:lookup(Subscriptions, MsgType),
  +    notify_subscriber(Subscribers, DecodedMsg, CallbackState).
  +
  +notify_subscriber([], _Message, CallbackState) ->
  +    {ok, CallbackState};
  +notify_subscriber([{_Type, Module, Filter}|Rest], DecodedMsg, CallbackState) ->
  +    case subscriber_filter(Filter, DecodedMsg) of
  +        true ->
  +            case do_callback(Module, handle_message, [DecodedMsg, CallbackState]) of
  +                {ok, NewCallbackState} ->
  +                    notify_subscriber(Rest, DecodedMsg, NewCallbackState);
  +                {terminate, Reason} ->
  +                    {error, Reason}
  +            end;
  +        _ ->
  +            notify_subscriber(Rest, DecodedMsg, CallbackState)
  +    end.
  +
  +subscriber_filter(true, _DecodedMsg) -> true;
  +subscriber_filter(Fn, DecodedMsg) when is_function(Fn) ->
  +    % catch errors?
  +    Fn(DecodedMsg).
  +
  +
   % XXX process commands through ofs_store
   send(Msg, State) ->
       Conn = State#?STATE.main_connection,
  
positive.code:         callback_mod,
          callback_state,
          generation_id,
          opt
      }
  ).
  
  
negative.code:     ok.
  
  subscribe(Module, in_packet, State) ->
      subscribe(Module, {in_packet, true}, State);
  subscribe(_Module, {in_packet, _FilterFn}, _State) ->
      % XXX store subscription in ets?
      ok;
  