anchor.diff:              #state{subscriber = Pid} = State) ->
     error_logger:info_msg("~p ~p subscriber ~p is down",
                           [?MODULE, self(), Pid]),
  -  NewState = reset_buffer(State#state{subscriber = undefined}),
  +  NewState = reset_buffer(State#state{ subscriber      = undefined
  +                                     , subscriber_mref = undefined
  +                                     }),
     {noreply, NewState};
   handle_info({'DOWN', _MonitorRef, process, Pid, Reason},
               #state{socket_pid = Pid} = State) ->
  -  {stop, {socket_down, Reason}, State};
  +  Timeout = ?SOCKET_RETRY_DELAY_MS,
  +  error_logger:info_msg("~p ~p socket ~p is down, reason: ~p, "
  +                        "retrying in ~p ms",
  +                        [?MODULE, self(), Pid, Reason, Timeout]),
  +  erlang:send_after(Timeout, self(), init_socket),
  +  State1 = State#state{socket_pid = undefined},
  +  NewState = reset_buffer(State1),
  +  {noreply, NewState};
   handle_info(Info, State) ->
     error_logger:warning_msg("~p ~p got unexpected info: ~p",
                             [?MODULE, self(), Info]),
  
positive.code: -define(DEFAULT_SLEEP_TIMEOUT, 1000). % 1 sec
  -define(DEFAULT_PREFETCH_COUNT, 1).
  -define(ERROR_COOLDOWN, 1000).
  
  -define(SEND_FETCH_REQUEST, send_fetch_request).
  
  
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.
  