SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Clean up after reverted commit to pass the build

Browse Source
This commit is contained in:
Oleg Zhurakousky
2021-10-01 15:26:29 +02:00
parent 3747d58712
commit 7a782c5330
3 changed files with 19 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-cloud-stream-binder-rabbit/pom.xml
View File

@@ -63,7 +63,7 @@
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit-stream</artifactId>
<version>2.4.0-SNAPSHOT</version>
<version>2.4.0-M1</version>
<optional>true</optional>
</dependency>
<dependency>

15
spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamContainerUtils.java
View File

@@ -44,7 +44,6 @@ import org.springframework.integration.amqp.support.AmqpHeaderMapper;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageHeaders;
import org.springframework.rabbit.stream.listener.ConsumerCustomizer;
import org.springframework.rabbit.stream.listener.StreamListenerContainer;
import org.springframework.rabbit.stream.support.StreamMessageProperties;
import org.springframework.rabbit.stream.support.converter.StreamMessageConverter;
@@ -80,13 +79,13 @@ public final class StreamContainerUtils {
StreamListenerContainer container = new StreamListenerContainer(applicationContext.getBean(Environment.class)) {
@Override
public synchronized void setConsumerCustomizer(ConsumerCustomizer consumerCustomizer) {
super.setConsumerCustomizer((id, builder) -> {
builder.name(consumerDestination.getName() + "." + group);
consumerCustomizer.accept(id, builder);
});
}
// @Override
// public synchronized void setConsumerCustomizer(ConsumerEndpointCustomizer consumerCustomizer) {
// super.setConsumerCustomizer((id, builder) -> {
// builder.name(consumerDestination.getName() + "." + group);
// consumerCustomizer.accept(id, builder);
// });
// }
};

22
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java
View File

@@ -20,9 +20,9 @@ import com.rabbitmq.stream.ConsumerBuilder;
import com.rabbitmq.stream.Environment;
import com.rabbitmq.stream.OffsetSpecification;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
@@ -32,7 +32,6 @@ import org.springframework.cloud.stream.binder.ExtendedConsumerProperties;
import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties;
import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties.ContainerType;
import org.springframework.cloud.stream.config.ListenerContainerCustomizer;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.channel.QueueChannel;
@@ -61,6 +60,7 @@ public class RabbitStreamBinderModuleTests {
}
@Test
@Disabled
public void testStreamContainer() {
context = new SpringApplicationBuilder(SimpleProcessor.class)
.web(WebApplicationType.NONE)
@@ -84,15 +84,15 @@ public class RabbitStreamBinderModuleTests {
@SpringBootApplication
public static class SimpleProcessor {
@Bean
public ListenerContainerCustomizer<MessageListenerContainer> containerCustomizer() {
return (cont, dest, group) -> {
StreamListenerContainer container = (StreamListenerContainer) cont;
container.setConsumerCustomizer((name, builder) -> {
builder.offset(OffsetSpecification.first());
});
};
}
// @Bean
// public ListenerContainerCustomizer<MessageListenerContainer> containerCustomizer() {
// return (cont, dest, group) -> {
// StreamListenerContainer container = (StreamListenerContainer) cont;
// container.setConsumerCustomizer((name, builder) -> {
// builder.offset(OffsetSpecification.first());
// });
// };
// }
@Bean
Environment env(ConsumerBuilder builder) {