anchor.diff:              try rabl_amqp:subscribe(Channel, SinkQueue, self()) of
                   {ok, Tag} ->
                       State2 = State#state{subscription_tag=Tag},
  +                    lager:info("consumer ready"),
                       {next_state, consuming, State2};
                   Error ->
                       lager:error("Error ~p subscribing to ~p ~n", [Error, SinkQueue]),
                       Delay2 = backoff(Delay),
  +                    erlang:send_after(Delay, self(), connect_timeout),
                       {next_state, unsubscribed, State#state{reconnect_delay_millis=Delay2,
  -                                                           connection_attempt_counter=CAC+1}, Delay}
  +                                                           connection_attempt_counter=CAC+1}}
               catch exit:Ex when is_tuple(Ex) andalso element(1, Ex) == noproc ->
                       %% connection | channel busted, destroy and start
                       %% over
                       disconnect(State),
  -                    {next_state, disconnected, State, 0}
  +                    erlang:send_after(0, self(), connect_timeout),
  +                    {next_state, disconnected, State}
               end
       end.
   
  
positive.code:     PrefetchCount = application:get_env(rabl, prefetch_count, ?DEFAULT_PREFETCH),
      Delay2 = backoff(Delay),
  
      case CAC >= MaxRetries of
          true ->
              {stop, max_connection_retry_limit_reached, State};
          false ->
              case rabl_amqp:connection_start(AMQPParams) of
  
negative.code:             State=#state{connection_monitor=ConnMonRef, connection=Connection}) ->
      lager:error("Connection Down with reason ~p~n", [Reason]),
      connect(State);
  handle_info(connect_timeout, StateName, State) ->
      ?MODULE:StateName(connect_timeout, State);
  handle_info(Info, AnyState, State) ->
      %% TODO implement throttle here?
      Time = os:timestamp(),
  