anchor.diff:  -define(DEFAULT_SLEEP_TIMEOUT, 1000). % 1 sec
   -define(DEFAULT_PREFETCH_COUNT, 1).
   -define(ERROR_COOLDOWN, 1000).
  +-define(SOCKET_RETRY_DELAY_MS, 1000).
   
   -define(SEND_FETCH_REQUEST, send_fetch_request).
   
  
positive.code:             #state{subscriber = Pid} = State) ->
    error_logger:info_msg("~p ~p subscriber ~p is down",
                          [?MODULE, self(), Pid]),
    NewState = reset_buffer(State#state{subscriber = undefined}),
    {noreply, NewState};
  handle_info({'DOWN', _MonitorRef, process, Pid, Reason},
              #state{socket_pid = Pid} = State) ->
    {stop, {socket_down, Reason}, State};
  handle_info(Info, State) ->
    error_logger:warning_msg("~p ~p got unexpected info: ~p",
                            [?MODULE, self(), Info]),
  
negative.code: maybe_send_fetch_request(#state{ subscriber     = Subscriber
                                 , pending_acks   = PendingAcks
                                 , prefetch_count = PrefetchCount
                                 , socket_pid     = SocketPid
                                 } = State) ->
    case is_pid(SocketPid) andalso
         is_pid(Subscriber) andalso
         length(PendingAcks) < PrefetchCount of
      true ->
        case send_fetch_request(State) of
          {ok, CorrId} ->
            State#state{last_corr_id = CorrId};
          {error, {sock_down, _Reason}} ->
            %% ignore error here, the socket pid 'DOWN' message
            %% should trigger the socket re-init loop
            State
        end;
      false ->
        State
    end.
  