anchor.diff:  %% Description: Handling call messages
   %% @hidden
   %%--------------------------------------------------------------------
  -handle_call({register_worker, {WorkerType, Url, Reader}}, {From, _Ref}, State = #state{ maps=Maps }) ->
  -    {WorkerType, Ring} = lists:keyfind(WorkerType, 1, Maps),
  -    Ring1 = case lists:keyfind(Url, 1, Ring) of
  -               {Url, {async, From}} ->
  -                   lists:keyreplace(Url, 1, Ring, {Url, Reader});
  -               _ -> Ring
  -           end,
  -    Ring2 = lists:map(fun ({Frag, {async, _}}) ->
  -                              {Frag, undefined};
  -                          ({_Frag, Pool}=Mapping) when is_pid(Pool) ->
  -                              Mapping
  -                      end, Ring1),
  -    io:format("RING2 ~p~n", [Ring2]),
  -    {ok, Map, Interval} = redis_shard:generate_map_and_interval(Ring2),
  -    io:format("~p ~p ~p~n", [WorkerType, Map, Interval]),
  +notify_complete(State0=#state{ reply_to=Pid }) 
  +  when Pid =:= self; Pid =:= undefined ->
  +    State0#state{ reply_to=undefined };
  +
  +notify_complete(State0=#state{ maps=TempTable, reply_to=Pid }) ->
  +    case dict:size(TempTable) of
  +        0 ->
  +            gen_server:reply(Pid, ok),
  +            State0#state{ reply_to=undefined };
  +        _ -> State0
  +    end.
  +
  +update_worker_pid(Url, From, Pid, Ring) ->
  +    case lists:keyfind(Url, 1, Ring) of
  +        {Url, {async, From}} ->
  +            lists:keyreplace(Url, 1, Ring, {Url, Pid});
  +        _ -> Ring
  +    end.
  +
  +save_shard_info(WorkerType, Ring0) ->
  +    MapFun = fun ({Frag, {async, _}}, AccIn) ->
  +                     {{Frag, undefined}, AccIn};
  +                 ({_Frag, Pool}=Mapping, AccIn) when is_pid(Pool) ->
  +                     {Mapping, AccIn+1}
  +             end,
  +    {Ring1, Completed} = lists:mapfoldl(MapFun, 0, Ring0),
  +    {ok, Map, Interval} = redis_shard:generate_map_and_interval(Ring1),
       logplex_shard_info:save(WorkerType, Map, Interval),
  -    Maps1 = lists:keyreplace(WorkerType, 1, Maps, {WorkerType, Ring1}),
  -    {reply, ok, State#state{ maps=Maps1 }};
  +    case Completed =:= length(Ring1) of
  +        true -> done;
  +        false -> incomplete
  +    end.
  +
  +handle_call({register_worker, {WorkerType, Url, Reader}}, {From, _Ref}, State = #state{ maps=TempTable0 }) ->
  +    Ring = dict:fetch(WorkerType, TempTable0),
  +    Ring1 = update_worker_pid(Url, From, Reader, Ring),
  +    TempTable = case save_shard_info(WorkerType, Ring1) of
  +                    done ->
  +                        dict:erase(WorkerType, TempTable0);
  +                    incomplete ->
  +                        dict:store(WorkerType, Ring1, TempTable0)
  +                end,
  +    State1 = notify_complete(State#state{ maps=TempTable }),
  +    {reply, ok, State1};
   
   handle_call({commit, new_shard_info}, _From, State) ->
       backup_shard_info(),
  
positive.code:             {reply, {error, {C, E}}, State}
      end;
  
  handle_call({prepare, {new_shard_info, NewShardInfo}}, _From, State) ->
      {reply, prepare_new_shard_info(NewShardInfo), State};
  
  handle_call({make_permanent, new_shard_info}, _From, State) ->
      try
  
negative.code: -define(CURRENT_WRITE_MAP, logplex_redis_buffer_map).
  -define(BACKUP_WRITE_MAP, backup_logplex_redis_buffer_map).
  
  -record(state, {urls :: list(),
                  maps = dict:new(),
                  reply_to = undefined :: {pid(), reference()}}).
  
  -define(TIMEOUT, 30000).
  
  