Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix DSL for inner bean names generation (#8639)

Browse Source 
The `IntegrationFlowBeanPostProcessor.processIntegrationComponentSpec()`
uses a wrong `generateBeanName()` for component to register making the
provided `id` as a prefix

* Use `generateBeanName(Object instance, String prefix, @Nullable String fallbackId, boolean useFlowIdAsPrefix)`
instead to properly "fallback" to the provided name

**Cherry-pick to `6.1.x`**
This commit is contained in:
Artem Bilan
2023-06-08 09:50:32 -04:00
committed by GitHub
parent 79408d7b2c
commit 8b8a7a4c16
2 changed files with 18 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java
View File

@@ -356,7 +356,7 @@ public class IntegrationFlowBeanPostProcessor
.filter(component -> noBeanPresentForComponent(component.getKey(), beanName))
.forEach(component ->
registerComponent(component.getKey(),
generateBeanName(component.getKey(), component.getValue())));
generateBeanName(component.getKey(), beanName, component.getValue(), false)));
}
}

28
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -244,8 +244,10 @@ public class KafkaDslTests {
}
@Test
void channels(@Autowired MessageChannel topic6Channel, @Autowired PollableKafkaChannel topic8Channel,
@Autowired PollableKafkaChannel topic9Channel) {
void channels(@Qualifier("topic6Channel") MessageChannel topic6Channel,
@Qualifier("topic8Channel") PollableKafkaChannel topic8Channel,
@Qualifier("topic9Channel") PollableKafkaChannel topic9Channel) {
topic6Channel.send(new GenericMessage<>("foo"));
Message<?> received = topic8Channel.receive();
assertThat(received)
@@ -344,7 +346,8 @@ public class KafkaDslTests {
}
@Bean
public IntegrationFlow sendToKafkaFlow() {
public IntegrationFlow sendToKafkaFlow(
KafkaProducerMessageHandlerSpec<Integer, String, ?> kafkaMessageHandlerTopic2) {
return f -> f
.<String>split(p -> Stream.generate(() -> p).limit(101).iterator(), null)
.enrichHeaders(h -> h.header(KafkaIntegrationHeaders.FUTURE_TOKEN, "foo"))
@@ -353,17 +356,15 @@ public class KafkaDslTests {
kafkaMessageHandler(producerFactory(), TEST_TOPIC1)
.timestampExpression("T(Long).valueOf('1487694048633')"),
e -> e.id("kafkaProducer1")))
.subscribe(sf -> sf.handle(
kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
.flush(msg -> true)
.timestamp(m -> 1487694048644L),
e -> e.id("kafkaProducer2")))
.subscribe(sf -> sf.handle(kafkaMessageHandlerTopic2, e -> e.id("kafkaProducer2")))
);
}
@Bean
public DefaultKafkaHeaderMapper mapper() {
return new DefaultKafkaHeaderMapper();
public KafkaProducerMessageHandlerSpec<Integer, String, ?> kafkaMessageHandlerTopic2() {
return kafkaMessageHandler(producerFactory(), TEST_TOPIC2)
.flush(msg -> true)
.timestamp(m -> 1487694048644L);
}
private KafkaProducerMessageHandlerSpec<Integer, String, ?> kafkaMessageHandler(
@@ -382,6 +383,11 @@ public class KafkaDslTests {
.configureKafkaTemplate(t -> t.id("kafkaTemplate:" + topic));
}
@Bean
public DefaultKafkaHeaderMapper mapper() {
return new DefaultKafkaHeaderMapper();
}
@Bean
public IntegrationFlow sourceFlow() {
@@ -427,7 +433,7 @@ public class KafkaDslTests {
@Bean
public IntegrationFlow channels(KafkaTemplate<Integer, String> template,
ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory,
KafkaMessageSource<?, ?> channelSource,
@Qualifier("channelSource") KafkaMessageSource<?, ?> channelSource,
PublishSubscribeKafkaChannel publishSubscribeKafkaChannel) {
return IntegrationFlow.from(topic6Channel(template, containerFactory))