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

INTEXT-104: Kafka: order for o-channel-adapter

Browse Source 
JIRA: https://jira.spring.io/browse/INTEXT-104

`order` attribute now honored if kafka outbound adapter is connected to a subscribable channel
unit tests and samples are updated

Polishing: use `<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>` for adapter tags to cover `SmartLifecycle` options.
This commit is contained in:
Soby Chacko
2014-07-22 23:44:36 -04:00
committed by Artem Bilan
parent e22a90ea1f
commit 999878f048
3 changed files with 20 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd
View File

@@ -9,7 +9,7 @@
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:import namespace="http://www.springframework.org/schema/integration"
schemaLocation="http://www.springframework.org/schema/integration/spring-integration.xsd"/>
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-4.0.xsd"/>
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -380,16 +380,7 @@
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attributeGroup ref="coreKafkaComponentAttributes"/>
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="send-timeout" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
@@ -419,8 +410,7 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="kafka-consumer-context-ref" use="required"
type="xsd:string">
<xsd:attribute name="kafka-consumer-context-ref" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Kafka Server Bean Name
@@ -430,30 +420,6 @@
</xsd:complexType>
</xsd:element>
<xsd:attributeGroup name="coreKafkaComponentAttributes">
<xsd:attribute name="id" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
Identifies the underlying Spring bean definition, which is an
instance of either 'EventDrivenConsumer' or 'PollingConsumer',
depending on whether the component's input channel is a
'SubscribableChannel' or 'PollableChannel'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" default="true" use="optional">
<xsd:annotation>
<xsd:documentation>
Flag to indicate that the component should start automatically
on startup (default true).
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
</xsd:attributeGroup>
<xsd:element name="outbound-channel-adapter">
<xsd:annotation>
<xsd:documentation>
@@ -466,42 +432,20 @@
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:attribute name="id" type="xsd:string" use="optional">
<xsd:annotation>
<xsd:documentation>
Identifies the underlying Spring bean definition, which is an
instance of either 'EventDrivenConsumer' or 'PollingConsumer',
depending on whether the component's input channel is a
'SubscribableChannel' or 'PollableChannel'.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="auto-startup" default="true" use="optional">
<xsd:annotation>
<xsd:documentation>
Flag to indicate that the component should start automatically
on startup (default true).
</xsd:documentation>
</xsd:annotation>
<xsd:simpleType>
<xsd:union memberTypes="xsd:boolean xsd:string"/>
</xsd:simpleType>
</xsd:attribute>
<xsd:attribute name="kafka-producer-context-ref" use="required"
type="xsd:string">
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="kafka-producer-context-ref" use="required" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
Kafka producer context reference.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="channel" type="xsd:string">
<xsd:attribute name="order">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.core.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
<xsd:documentation>
Specifies the order for invocation when this endpoint is connected as a
subscriber to a SubscribableChannel.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>

1
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests-context.xml
View File

@@ -17,6 +17,7 @@
kafka-producer-context-ref="kafkaProducerContext"
auto-startup="false"
channel="inputToKafka"
order="3"
>
<int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/>
</int-kafka:outbound-channel-adapter>

14
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.integration.kafka.config.xml;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.endpoint.PollingConsumer;
@@ -32,20 +33,23 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class KafkaOutboundAdapterParserTests<K,V> {
public class KafkaOutboundAdapterParserTests<K, V> {
@Autowired
private ApplicationContext appContext;
@Test
@SuppressWarnings("unchecked")
public void testOutboundAdapterConfiguration(){
final PollingConsumer pollingConsumer = appContext.getBean("kafkaOutboundChannelAdapter", PollingConsumer.class);
final KafkaProducerMessageHandler<K,V> messageHandler = appContext.getBean(KafkaProducerMessageHandler.class);
public void testOutboundAdapterConfiguration() {
final PollingConsumer pollingConsumer =
appContext.getBean("kafkaOutboundChannelAdapter", PollingConsumer.class);
final KafkaProducerMessageHandler<K, V> messageHandler = appContext.getBean(KafkaProducerMessageHandler.class);
Assert.assertNotNull(pollingConsumer);
Assert.assertNotNull(messageHandler);
final KafkaProducerContext<K,V> producerContext = messageHandler.getKafkaProducerContext();
Assert.assertEquals(messageHandler.getOrder(), 3);
final KafkaProducerContext<K, V> producerContext = messageHandler.getKafkaProducerContext();
Assert.assertNotNull(producerContext);
Assert.assertEquals(producerContext.getTopicsConfiguration().size(), 2);
}
}