anchor.diff:  -module(pulsedb_realtime).
   -export([subscribe/2, unsubscribe/1]).
  --export([start_link/0, init/1, terminate/2, handle_info/2, handle_cast/2]).
  +-export([start_link/0, init/1, terminate/2, handle_info/2, handle_call/2]).
   
   -record(subscriptions, 
    {
  
positive.code:   ok.
  
  
  handle_cast({subscribe, Pid, Query, Tag}, #subscriptions{clients=Clients0}=State) ->
    erlang:monitor(process, Pid),
    Clients = case lists:keytake(Query, 1, Clients0) of
      {value, {Q, UTC, Pids0}, Rest} ->
        Pids = lists:usort([{Pid,Tag}|Pids0]),
        [{Q,UTC, Pids} | Rest];
      false ->
        [{Query,undefined,[{Pid,Tag}]} | Clients0]
    end,
    
    {noreply, State#subscriptions{clients=Clients}}.
  
  
  handle_info(tick, #subscriptions{clients=Clients}=State) ->
    T1 = erlang:now(),
    Clients1 = lists:map(fun({Query,LastUTC,Pids} = Entry) ->
      case pulsedb:read(Query, memory) of
        {ok, [], _} -> 
  
negative.code:         {UTC, V} = lists:last(Data),
          if
            LastUTC == undefined orelse UTC > LastUTC ->
              [Pid ! {pulse, Tag, UTC, V} || {Pid,Tag,_} <- Pids],
              {Query, UTC, Pids};
            true ->
              Entry
          end                 
      end
    end, Clients),
    {_,Delay} = pulse:current_second(),
    erlang:send_after(Delay, self(), tick),
    {noreply, State#subscriptions{clients = Clients1}};
  
  
  
  handle_info({'DOWN', _, _, Pid, _}, #subscriptions{clients=Clients0}=State) ->
    Clients1 = [{Query, UTC, lists:keydelete(Pid,1,Pids)} 
               || {Query, UTC, Pids} <- Clients0],
    Clients = [C || {Pids,_,_}=C <- Clients1, length(Pids) > 0],
    {noreply, State#subscriptions{clients=Clients}}.
  