AMQP-801: Move ConsumeOkEvent to handleConsumeOk

JIRA: https://jira.spring.io/browse/AMQP-801

The proper place to emit the `ConsumeOkEvent` is really a
`handleConsumeOk()` callback when broker answers us that consumer has
been subscribed to the queue

**Cherry-pick to 1.7.x**

Author: artembilan

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/BlockingQueueConsumer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -677,9 +677,6 @@ private void consumeFromQueue(String queue) throws IOException {
 		else {
 			logger.error("Null consumer tag received for queue " + queue);
 		}
-		if (this.applicationEventPublisher != null) {
-			this.applicationEventPublisher.publishEvent(new ConsumeOkEvent(this, queue, consumerTag));
-		}
 	}
 
 	private void attemptPassiveDeclarations() {
@@ -875,7 +872,12 @@ private final class InternalConsumer extends DefaultConsumer {
 		public void handleConsumeOk(String consumerTag) {
 			super.handleConsumeOk(consumerTag);
 			if (logger.isDebugEnabled()) {
-				logger.debug("ConsumeOK : " + BlockingQueueConsumer.this);
+				logger.debug("ConsumeOK: " + BlockingQueueConsumer.this);
+			}
+			if (BlockingQueueConsumer.this.applicationEventPublisher != null) {
+				String queueName = BlockingQueueConsumer.this.consumerTags.get(consumerTag);
+				BlockingQueueConsumer.this.applicationEventPublisher
+						.publishEvent(new ConsumeOkEvent(this, queueName, consumerTag));
 			}
 		}