GH-1399 Added support for listener container customizer

This work is basically parsing out some rabbit specific work from https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/pull/145 into a an AbstractMessageChannelBinder

Resolves #1399
Resolves #1400

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java

@@ -27,6 +27,7 @@ import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
 import org.springframework.beans.factory.support.DefaultSingletonBeanRegistry;
+import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
 import org.springframework.cloud.stream.provisioning.ConsumerDestination;
 import org.springframework.cloud.stream.provisioning.ProducerDestination;
 import org.springframework.cloud.stream.provisioning.ProvisioningException;
@@ -53,6 +54,8 @@ import org.springframework.messaging.support.ChannelInterceptorAdapter;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.util.Assert;
 
+
+
 /**
  * {@link AbstractBinder} that serves as base class for {@link MessageChannel} binders.
  * Implementors must implement the following methods:
@@ -94,23 +97,18 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 	 */
 	private final String[] headersToEmbed;
 
+	private final ListenerContainerCustomizer<?> containerCustomizer;
+
 	private ApplicationEventPublisher applicationEventPublisher;
 
-	public AbstractMessageChannelBinder(String[] headersToEmbed,
-			PP provisioningProvider) {
-
-		this.headersToEmbed = headersToEmbed == null ? new String[0] : headersToEmbed;
-		this.provisioningProvider = provisioningProvider;
+	public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider) {
+		this(headersToEmbed, provisioningProvider, null);
 	}
 
-	/**
-	 * @deprecated As of release 2.0. Please use other constructors.
-	 */
-	@Deprecated
-	protected AbstractMessageChannelBinder(boolean supportsHeadersNatively, String[] headersToEmbed,
-			PP provisioningProvider) {
-
-		this(headersToEmbed, provisioningProvider);
+	public AbstractMessageChannelBinder(String[] headersToEmbed, PP provisioningProvider, ListenerContainerCustomizer<?> containerCustomizer) {
+		this.headersToEmbed = headersToEmbed == null ? new String[0] : headersToEmbed;
+		this.provisioningProvider = provisioningProvider;
+		this.containerCustomizer = containerCustomizer == null ? (c, q, g) -> { } : containerCustomizer;
 	}
 
 	@Override
@@ -122,6 +120,11 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		return this.applicationEventPublisher;
 	}
 
+	@SuppressWarnings("unchecked")
+	protected <L> ListenerContainerCustomizer<L> getContainerCustomizer() {
+		return (ListenerContainerCustomizer<L>) this.containerCustomizer;
+	}
+
 	/**
 	 * Binds an outbound channel to a given destination. The implementation delegates to
 	 * {@link ProvisioningProvider#provisionProducerDestination(String, ProducerProperties)}

spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/ListenerContainerCustomizer.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.stream.config;
+
+/**
+ * If a single bean of this type is in the application context, listener
+ * containers created by the binder can be further customized after all
+ * the properties are set. For example, to configure less-common
+ * properties.
+ *
+ * @author Gary Russell
+ * @author Oleg Zhurakousky
+ *
+ * @since 2.1
+ *
+ */
+@FunctionalInterface
+public interface ListenerContainerCustomizer<T> {
+
+	/**
+	 * Configure the container that is being created for the supplied queue name and
+	 * consumer group.
+	 * @param container the container.
+	 * @param destinationName the destination name.
+	 * @param group the consumer group.
+	 */
+	void configure(T container, String destinationName, String group);
+
+}

spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java

@@ -474,7 +474,7 @@ public class BindingServiceTests {
 				Binding.class);
 		int n = 0;
 		while (n++ < 300 && delegate == null) {
-			Thread.sleep(200);
+			Thread.sleep(400);
 		}
 		assertThat(delegate).isSameAs(mockBinding);
 		service.unbindConsumers(inputChannelName);