anchor.diff:      notify_all(BindingList, Value),
       reply_to_all(Threads, {ok, Value, Next}).
   
  -reply_to_all([], _Result) ->
  -    ok;
  -
   reply_to_all([H|T], Result) ->
       {server, undefined, {Address, Ref}} = H,
       gen_server:reply({Address, Ref}, Result),
  -    reply_to_all(T, Result).
  -
  -next_key(NextKey0, Type, State) ->
  -    case NextKey0 of
  -        undefined ->
  -            {ok, NextKey} = declare_next(Type, State),
  -            NextKey;
  -        _ ->
  -            NextKey0
  -    end.
  -
  -notify_all(L, Value) ->
  -    case L of
  -        [H|T] ->
  -            notify_value(H, Value),
  -            notify_all(T, Value);
  -        [] ->
  -            ok
  -    end.
  +    reply_to_all(T, Result);
  +reply_to_all([], _Result) ->
  +    ok.
  +
  +next_key(undefined, Type, State) ->
  +    {ok, NextKey} = declare_next(Type, State),
  +    NextKey;
  +next_key(NextKey0, _, _) ->
  +    NextKey0.
  +
  +notify_all([H|T], Value) ->
  +    notify_value(H, Value),
  +    notify_all(T, Value);
  +notify_all([], _) ->
  +    ok.
   
   %% @doc Declare the next object for streams.
   declare_next(Type, State=#state{partition=Partition, node=Node}) ->
  
positive.code:                 end
      end;
  
  handle_command({read, X, Function}, From,
                 State=#state{variables=Variables}) ->
      [{_Key, V=#dv{value=Value,
                    bound=Bound,
  
negative.code: 
  write(Type, Value, Next, Functions0, Key, Variables) ->
      lager:info("Writing key: ~p next: ~p", [Key, Next]),
      [{_Key, #dv{waiting_threads=Threads,
                  binding_list=BindingList,
                  lazy=Lazy}}] = ets:lookup(Variables, Key),
      Functions = lists:usort(Functions0),
      V1 = #dv{type=Type, value=Value, functions=Functions, next=Next,
               lazy=Lazy, bound=true},
  