GH-1352: ConsumerCustomizer Improvements

- pass in the listener id, if available
- narrow to a specific interface to aid Boot auto configuration
- add null check

* Apply Suggestion to add @FuntionalInterface

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/StreamRabbitListenerContainerFactory.java

@@ -17,7 +17,6 @@
 package org.springframework.rabbit.stream.config;
 
 import java.lang.reflect.Method;
-import java.util.function.Consumer;
 
 import org.springframework.amqp.rabbit.batch.BatchingStrategy;
 import org.springframework.amqp.rabbit.config.BaseRabbitListenerContainerFactory;
@@ -26,11 +25,11 @@ import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint;
 import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint;
 import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
 import org.springframework.lang.Nullable;
+import org.springframework.rabbit.stream.listener.ConsumerCustomizer;
 import org.springframework.rabbit.stream.listener.StreamListenerContainer;
 import org.springframework.rabbit.stream.listener.adapter.StreamMessageListenerAdapter;
 import org.springframework.util.Assert;
 
-import com.rabbitmq.stream.ConsumerBuilder;
 import com.rabbitmq.stream.Environment;
 
 /**
@@ -47,7 +46,7 @@ public class StreamRabbitListenerContainerFactory
 
 	private boolean nativeListener;
 
-	private Consumer<ConsumerBuilder> consumerCustomizer;
+	private ConsumerCustomizer consumerCustomizer;
 
 	private ContainerCustomizer<StreamListenerContainer> containerCustomizer;
 
@@ -72,7 +71,7 @@ public class StreamRabbitListenerContainerFactory
 	 * Customize the consumer builder before it is built.
 	 * @param consumerCustomizer the customizer.
 	 */
-	public void setConsumerCustomizer(java.util.function.Consumer<ConsumerBuilder> consumerCustomizer) {
+	public void setConsumerCustomizer(ConsumerCustomizer consumerCustomizer) {
 		this.consumerCustomizer = consumerCustomizer;
 	}
 

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/ConsumerCustomizer.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2021 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
+ *
+ *      https://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.rabbit.stream.listener;
+
+import java.util.function.BiConsumer;
+
+import com.rabbitmq.stream.ConsumerBuilder;
+
+/**
+ * Customizer for {@link ConsumerBuilder}.
+ *
+ * @author Gary Russell
+ * @since 2.4
+ *
+ */
+@FunctionalInterface
+public interface ConsumerCustomizer extends BiConsumer<String, ConsumerBuilder> {
+}

spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java

@@ -50,7 +50,7 @@ public class StreamListenerContainer implements MessageListenerContainer, BeanNa
 
 	private StreamMessageConverter messageConverter;
 
-	private java.util.function.Consumer<ConsumerBuilder> consumerCustomizer = c -> { };
+	private ConsumerCustomizer consumerCustomizer = (id, con) -> { };
 
 	private Consumer consumer;
 
@@ -112,7 +112,8 @@ public class StreamListenerContainer implements MessageListenerContainer, BeanNa
 	 * Customize the consumer builder before it is built.
 	 * @param consumerCustomizer the customizer.
 	 */
-	public synchronized void setConsumerCustomizer(java.util.function.Consumer<ConsumerBuilder> consumerCustomizer) {
+	public synchronized void setConsumerCustomizer(ConsumerCustomizer consumerCustomizer) {
+		Assert.notNull(consumerCustomizer, "'consumerCustomizer' cannot be null");
 		this.consumerCustomizer = consumerCustomizer;
 	}
 
@@ -167,7 +168,7 @@ public class StreamListenerContainer implements MessageListenerContainer, BeanNa
 	@Override
 	public synchronized void start() {
 		if (this.consumer == null) {
-			this.consumerCustomizer.accept(this.builder);
+			this.consumerCustomizer.accept(getListenerId(), this.builder);
 			this.consumer = this.builder.build();
 		}
 	}

spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java

@@ -61,15 +61,14 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
 
 	@Test
 	void simple(@Autowired RabbitTemplate template) throws InterruptedException {
-
 		template.convertAndSend("test.stream.queue1", "foo");
 		assertThat(this.config.latch1.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.config.received).isEqualTo("foo");
+		assertThat(this.config.id).isEqualTo("test");
 	}
 
 	@Test
 	void nativeMsg(@Autowired RabbitTemplate template) throws InterruptedException {
-
 		template.convertAndSend("test.stream.queue2", "foo");
 		assertThat(this.config.latch2.await(10, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.config.receivedNative).isNotNull();
@@ -97,6 +96,8 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
 
 		volatile Context context;
 
+		volatile String id;
+
 		@Bean
 		Environment environment() {
 			return Environment.builder()
@@ -140,13 +141,16 @@ public class RabbitListenerTests extends AbstractIntegrationTests {
 		RabbitListenerContainerFactory<StreamListenerContainer> nativeFactory(Environment env) {
 			StreamRabbitListenerContainerFactory factory = new StreamRabbitListenerContainerFactory(env);
 			factory.setNativeListener(true);
-			factory.setConsumerCustomizer(builder -> builder.name("myConsumer")
-					.offset(OffsetSpecification.first())
-					.manualTrackingStrategy());
+			factory.setConsumerCustomizer((id, builder) -> {
+				builder.name("myConsumer")
+						.offset(OffsetSpecification.first())
+						.manualTrackingStrategy();
+				this.id = id;
+			});
 			return factory;
 		}
 
-		@RabbitListener(queues = "test.stream.queue2", containerFactory = "nativeFactory")
+		@RabbitListener(id = "test", queues = "test.stream.queue2", containerFactory = "nativeFactory")
 		void nativeMsg(Message in, Context context) {
 			this.receivedNative = in;
 			this.context = context;