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

GH-204: DSL Register container as bean if needed

Browse Source 
Fixes https://github.com/spring-projects/spring-integration-kafka/issues/204

When an external container is provided to the DSL, register it as a bean
if it is not already a bean.

Polishing id from PR comments; add gateway support too.

* Polishing JavaDocs and omissions in the test-case
This commit is contained in:
Gary Russell
2018-05-14 12:43:01 -04:00
committed by Artem Bilan
parent e296d66a4b
commit 8decbe7810
4 changed files with 56 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/Kafka.java
View File

@@ -104,7 +104,11 @@ public final class Kafka {
}
/**
* Create an initial {@link KafkaMessageDrivenChannelAdapterSpec}.
* Create an initial {@link KafkaMessageDrivenChannelAdapterSpec}. If the listener
* container is not already a bean it will be registered in the application context.
* If the adapter spec has an {@code id}, the bean name will be that id appended with
* '.container'. Otherwise, the bean name will be generated from the container class
* name.
* @param listenerContainer the {@link AbstractMessageListenerContainer}.
* @param <K> the Kafka message key type.
* @param <V> the Kafka message value type.
@@ -117,7 +121,11 @@ public final class Kafka {
}
/**
* Create an initial {@link KafkaMessageDrivenChannelAdapterSpec}.
* Create an initial {@link KafkaMessageDrivenChannelAdapterSpec}. If the listener
* container is not already a bean it will be registered in the application context.
* If the adapter spec has an {@code id}, the bean name will be that id appended with
* '.container'. Otherwise, the bean name will be generated from the container class
* name.
* @param listenerContainer the {@link AbstractMessageListenerContainer}.
* @param listenerMode the {@link KafkaMessageDrivenChannelAdapter.ListenerMode}.
* @param <K> the Kafka message key type.
@@ -316,7 +324,10 @@ public final class Kafka {
/**
* Create an initial {@link KafkaInboundGatewaySpec} with the provided container and
* template.
* template. If the listener container is not already a bean it will be registered in
* the application context. If the adapter spec has an {@code id}, the bean name will
* be that id appended with '.container'. Otherwise, the bean name will be generated
* from the container class name.
* @param container the container.
* @param template the template.
* @param <K> the Kafka message key type.

12
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.kafka.dsl;
import java.util.Collections;
import java.util.Map;
import java.util.function.Consumer;
@@ -44,12 +45,16 @@ import org.springframework.util.Assert;
* @since 3.0.2
*/
public class KafkaInboundGatewaySpec<K, V, R, S extends KafkaInboundGatewaySpec<K, V, R, S>>
extends MessagingGatewaySpec<S, KafkaInboundGateway<K, V, R>> {
extends MessagingGatewaySpec<S, KafkaInboundGateway<K, V, R>>
implements ComponentsRegistration {
private final AbstractMessageListenerContainer<K, V> container;
KafkaInboundGatewaySpec(AbstractMessageListenerContainer<K, V> messageListenerContainer,
KafkaTemplate<K, R> kafkaTemplate) {
super(new KafkaInboundGateway<>(messageListenerContainer, kafkaTemplate));
this.container = messageListenerContainer;
}
/**
@@ -86,6 +91,11 @@ public class KafkaInboundGatewaySpec<K, V, R, S extends KafkaInboundGatewaySpec<
return _this();
}
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.container, getId() == null ? null : getId() + ".container");
}
/**
* A {@link ConcurrentMessageListenerContainer} configuration {@link KafkaInboundGatewaySpec}
* extension.

13
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-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.
@@ -46,11 +46,15 @@ import org.springframework.util.Assert;
* @since 3.0
*/
public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDrivenChannelAdapterSpec<K, V, S>>
extends MessageProducerSpec<S, KafkaMessageDrivenChannelAdapter<K, V>> {
extends MessageProducerSpec<S, KafkaMessageDrivenChannelAdapter<K, V>>
implements ComponentsRegistration {
private final AbstractMessageListenerContainer<K, V> container;
KafkaMessageDrivenChannelAdapterSpec(AbstractMessageListenerContainer<K, V> messageListenerContainer,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
super(new KafkaMessageDrivenChannelAdapter<>(messageListenerContainer, listenerMode));
this.container = messageListenerContainer;
}
/**
@@ -152,6 +156,11 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
return _this();
}
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.container, getId() == null ? null : getId() + ".container");
}
/**
* A {@link ConcurrentMessageListenerContainer} configuration {@link KafkaMessageDrivenChannelAdapterSpec}
* extension.

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

@@ -52,12 +52,14 @@ import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessage
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.ContainerProperties.AckMode;
import org.springframework.kafka.listener.GenericMessageListenerContainer;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.MessageListenerContainer;
@@ -100,8 +102,8 @@ public class KafkaDslTests {
private static final String TEST_TOPIC5 = "test-topic5";
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3,
TEST_TOPIC4, TEST_TOPIC5);
public static KafkaEmbedded embeddedKafka =
new KafkaEmbedded(1, true, TEST_TOPIC1, TEST_TOPIC2, TEST_TOPIC3, TEST_TOPIC4, TEST_TOPIC5);
@Autowired
@Qualifier("sendToKafkaFlow.input")
@@ -212,7 +214,7 @@ public class KafkaDslTests {
@Test
public void testGateways() throws Exception {
assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS));
assertThat(this.config.replyContainerLatch.await(30, TimeUnit.SECONDS)).isTrue();
assertThat(this.gate.exchange(TEST_TOPIC4, "foo")).isEqualTo("FOO");
}
@@ -259,16 +261,23 @@ public class KafkaDslTests {
.get();
}
@Bean
public ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<Integer, String> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckMode(AckMode.MANUAL);
factory.setRecoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(),
new RawRecordHeaderErrorMessageStrategy()));
factory.setRetryTemplate(new RetryTemplate());
return factory;
}
@Bean
public IntegrationFlow topic2ListenerFromKafkaFlow() {
return IntegrationFlows
.from(Kafka.messageDrivenChannelAdapter(consumerFactory(),
KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC2)
.configureListenerContainer(c ->
c.ackMode(ContainerProperties.AckMode.MANUAL))
.recoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(),
new RawRecordHeaderErrorMessageStrategy()))
.retryTemplate(new RetryTemplate())
.from(Kafka.messageDrivenChannelAdapter(kafkaListenerContainerFactory().createContainer(TEST_TOPIC2),
KafkaMessageDrivenChannelAdapter.ListenerMode.record)
.filterInRetry(true))
.filter(Message.class, m ->
m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101,
@@ -380,4 +389,5 @@ public class KafkaDslTests {
String exchange(@Header(KafkaHeaders.TOPIC) String topic, String out);
}
}