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

INTEXT-213: Outbound Adapter: Add RH Advice Chain

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

s/1.4/2.0/
This commit is contained in:
Gary Russell
2016-01-05 13:02:24 -05:00
committed by Artem Bilan
parent 5a25a9bd09
commit d3e724d134
4 changed files with 26 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-kafka/src/main/resources/META-INF/spring.schemas
View File

@@ -1,2 +1,2 @@
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-1.3.xsd=org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka-2.0.xsd=org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd
http\://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd=org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd

8
spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd → spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-2.0.xsd
View File

@@ -491,9 +491,11 @@
</xsd:annotation>
<xsd:complexType>
<xsd:sequence>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1"/>
</xsd:sequence>
<xsd:all>
<xsd:element ref="integration:poller" minOccurs="0" maxOccurs="1" />
<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType"
minOccurs="0" maxOccurs="1" />
</xsd:all>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="kafka-producer-context-ref" use="required" type="xsd:string">
<xsd:annotation>

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

@@ -22,6 +22,9 @@
message-key-expression="'bar'"
partition-id-expression="2">
<int:poller fixed-delay="1000" time-unit="MILLISECONDS" receive-timeout="0" task-executor="taskExecutor"/>
<int-kafka:request-handler-advice-chain>
<bean class="org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice" />
</int-kafka:request-handler-advice-chain>
</int-kafka:outbound-channel-adapter>
<int-kafka:outbound-channel-adapter id="kafkaOutboundChannelAdapterWithHeaderRoutingDisabled"

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

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2014 the original author or authors.
* Copyright 2013-2016 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.
@@ -16,21 +16,30 @@
package org.springframework.integration.kafka.config.xml;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.aop.Advisor;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.handler.advice.RequestHandlerCircuitBreakerAdvice;
import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler;
import org.springframework.integration.kafka.rule.KafkaEmbedded;
import org.springframework.integration.kafka.rule.KafkaRule;
import org.springframework.integration.kafka.support.KafkaProducerContext;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.MessageHandler;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -42,6 +51,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@DirtiesContext
public class KafkaOutboundAdapterParserTests {
@ClassRule
@@ -71,6 +81,11 @@ public class KafkaOutboundAdapterParserTests {
= this.appContext.getBean("org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler#1",
KafkaProducerMessageHandler.class);
assertEquals(false, TestUtils.getPropertyValue(messageHandler2, "enableHeaderRouting"));
MessageHandler advisedHandler = TestUtils.getPropertyValue(pollingConsumer, "handler", MessageHandler.class);
assertTrue(AopUtils.isJdkDynamicProxy(advisedHandler));
Advisor[] advisors = ((Advised) advisedHandler).getAdvisors();
assertEquals(1, advisors.length);
assertThat(advisors[0].getAdvice(), instanceOf(RequestHandlerCircuitBreakerAdvice.class));
}
}