anchor.diff:                         {next_state, consumer_state(), #state{}} |
                          {next_state, consumer_state(), #state{}, timeout()} |
                          {stop, Reason::atom(), #state{}}.
  -unsubscribed(timeout, State) ->
  +unsubscribed(connect_timeout, State) ->
  +    lager:info("connect_timeout in state unsubscribed"),
       subscribe(State).
   
   %%--------------------------------------------------------------------
  
positive.code:             try rabl_amqp:subscribe(Channel, SinkQueue, self()) of
                  {ok, Tag} ->
                      State2 = State#state{subscription_tag=Tag},
                      {next_state, consuming, State2};
                  Error ->
                      lager:error("Error ~p subscribing to ~p ~n", [Error, SinkQueue]),
                      Delay2 = backoff(Delay),
                      {next_state, unsubscribed, State#state{reconnect_delay_millis=Delay2,
                                                             connection_attempt_counter=CAC+1}, Delay}
              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}
              end
      end.
  
  
negative.code:                     case rabl_amqp:channel_open(Connection) of
                          {ok, Channel} ->
                              ChanMonRef = erlang:monitor(process, Channel),
                              ok = rabl_amqp:set_prefetch_count(Channel, PrefetchCount),
                              subscribe(State#state{channel=Channel,
                                                    channel_monitor=ChanMonRef,
  