diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/StreamRabbitListenerContainerFactory.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/StreamRabbitListenerContainerFactory.java index 411a24be..7d24a61f 100644 --- a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/StreamRabbitListenerContainerFactory.java +++ b/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 consumerCustomizer; + private ConsumerCustomizer consumerCustomizer; private ContainerCustomizer 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 consumerCustomizer) { + public void setConsumerCustomizer(ConsumerCustomizer consumerCustomizer) { this.consumerCustomizer = consumerCustomizer; } diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/ConsumerCustomizer.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/ConsumerCustomizer.java new file mode 100644 index 00000000..501cb3ff --- /dev/null +++ b/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 { +} diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java index dc83799c..4063b935 100644 --- a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java +++ b/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 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 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(); } } diff --git a/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java b/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java index 1375b446..5ce1602c 100644 --- a/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java +++ b/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 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;