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

INTEXT-175: Fix IEC population

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

Since `IntegrationEvaluationContextAware` infrastructure had had wrong design and has been deprecated in the SI-4.2, change its usage to the proper way according to the SI-core recipes.

Tested against Spring IO-2.0.
This commit is contained in:
Artem Bilan
2015-06-23 19:45:17 -04:00
committed by Artem Bilan
parent b77f621c51
commit 7695964e64
2 changed files with 37 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013-2014 the original author or authors.
* Copyright 2013-2015 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.
@@ -18,6 +18,7 @@ package org.springframework.integration.kafka.outbound;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.expression.IntegrationEvaluationContextAware;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.kafka.support.KafkaHeaders;
@@ -31,8 +32,7 @@ import org.springframework.util.Assert;
* @author Gary Russell
* @since 0.5
*/
public class KafkaProducerMessageHandler extends AbstractMessageHandler
implements IntegrationEvaluationContextAware {
public class KafkaProducerMessageHandler extends AbstractMessageHandler {
private final KafkaProducerContext kafkaProducerContext;
@@ -49,11 +49,6 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler
this.kafkaProducerContext = kafkaProducerContext;
}
@Override
public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
this.evaluationContext = evaluationContext;
}
public void setTopicExpression(Expression topicExpression) {
this.topicExpression = topicExpression;
}
@@ -70,31 +65,29 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler
return this.kafkaProducerContext;
}
@Override
protected void onInit() throws Exception {
super.onInit();
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
}
@Override
protected void handleMessageInternal(final Message<?> message) throws Exception {
String topic = this.topicExpression != null ?
this.topicExpression.getValue(getEvaluationContext(), message, String.class)
this.topicExpression.getValue(this.evaluationContext, message, String.class)
: message.getHeaders().get(KafkaHeaders.TOPIC, String.class);
Integer partitionId = this.partitionExpression != null ?
this.partitionExpression.getValue(getEvaluationContext(), message, Integer.class)
this.partitionExpression.getValue(this.evaluationContext, message, Integer.class)
: message.getHeaders().get(KafkaHeaders.PARTITION_ID, Integer.class);
Object messageKey = this.messageKeyExpression != null
? this.messageKeyExpression.getValue(getEvaluationContext(), message)
? this.messageKeyExpression.getValue(this.evaluationContext, message)
: message.getHeaders().get(KafkaHeaders.MESSAGE_KEY);
this.kafkaProducerContext.send(topic, partitionId, messageKey, message.getPayload());
}
private EvaluationContext getEvaluationContext() {
// Consider moving this back into onInit() once https://jira.spring.io/browse/INT-3749 is fixed
if (this.evaluationContext == null) {
throw new IllegalStateException("Evaluation context not initialized");
}
return this.evaluationContext;
}
@Override
public String getComponentType() {
return "kafka:outbound-channel-adapter";

53
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java
View File

@@ -32,20 +32,7 @@ import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import com.gs.collections.api.multimap.MutableMultimap;
import com.gs.collections.impl.factory.Multimaps;
import kafka.admin.AdminUtils;
import kafka.api.OffsetRequest;
import kafka.common.TopicExistsException;
import kafka.serializer.Decoder;
import kafka.serializer.Encoder;
import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
import org.springframework.context.expression.MapAccessor;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.kafka.core.DefaultConnectionFactory;
import org.springframework.integration.kafka.core.KafkaMessage;
import org.springframework.integration.kafka.core.ZookeeperConfiguration;
@@ -67,11 +54,22 @@ import org.springframework.integration.kafka.util.MessageUtils;
import org.springframework.integration.kafka.util.TopicUtils;
import org.springframework.messaging.support.MessageBuilder;
import org.junit.After;
import org.junit.Rule;
import org.junit.Test;
import com.gs.collections.api.multimap.MutableMultimap;
import com.gs.collections.impl.factory.Multimaps;
import kafka.admin.AdminUtils;
import kafka.api.OffsetRequest;
import kafka.common.TopicExistsException;
import kafka.serializer.Decoder;
import kafka.serializer.Encoder;
/**
* @author Gary Russell
* @author Marius Bogoevici
* @since 1.0
*
*/
public class OutboundTests {
@@ -136,9 +134,8 @@ public class OutboundTests {
SpelExpressionParser parser = new SpelExpressionParser();
handler.setMessageKeyExpression(parser.parseExpression("headers.foo"));
handler.setTopicExpression(parser.parseExpression("headers.bar"));
StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
evaluationContext.addPropertyAccessor(new MapAccessor());
handler.setIntegrationEvaluationContext(evaluationContext);
handler.afterPropertiesSet();
handler.handleMessage(MessageBuilder.withPayload("bar" + suffix)
.setHeader("foo", "3")
.setHeader("bar", TOPIC)
@@ -175,7 +172,7 @@ public class OutboundTests {
final String suffix = UUID.randomUUID().toString();
KafkaMessageListenerContainer kafkaMessageListenerContainer = createMessageListenerContainer(TOPIC,TOPIC2);
KafkaMessageListenerContainer kafkaMessageListenerContainer = createMessageListenerContainer(TOPIC, TOPIC2);
final Decoder<String> decoder = new StringDecoder();
@@ -212,9 +209,8 @@ public class OutboundTests {
SpelExpressionParser parser = new SpelExpressionParser();
handler.setMessageKeyExpression(parser.parseExpression("headers.foo"));
handler.setTopicExpression(parser.parseExpression("headers.bar"));
StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
evaluationContext.addPropertyAccessor(new MapAccessor());
handler.setIntegrationEvaluationContext(evaluationContext);
handler.afterPropertiesSet();
handler.handleMessage(MessageBuilder.withPayload("bar1" + suffix)
.setHeader("foo", "3")
.setHeader("bar", TOPIC)
@@ -278,9 +274,7 @@ public class OutboundTests {
handler.handleMessage(MessageBuilder.withPayload("fooTopic1" + suffix).build());
StandardEvaluationContext evaluationContext = new StandardEvaluationContext();
evaluationContext.addPropertyAccessor(new MapAccessor());
handler.setIntegrationEvaluationContext(evaluationContext);
handler.afterPropertiesSet();
producerContext.stop();
@@ -311,14 +305,17 @@ public class OutboundTests {
private KafkaProducerContext createProducerContext() throws Exception {
KafkaProducerContext kafkaProducerContext = new KafkaProducerContext();
Encoder<String> encoder = new StringEncoder();
ProducerMetadata<String, String> producerMetadata = new ProducerMetadata<String, String>(TOPIC, String.class, String.class, new EncoderAdaptingSerializer<String>(encoder), new EncoderAdaptingSerializer<String>(encoder));
ProducerMetadata<String, String> producerMetadata =
new ProducerMetadata<String, String>(TOPIC, String.class, String.class,
new EncoderAdaptingSerializer<>(encoder), new EncoderAdaptingSerializer<>(encoder));
Properties props = new Properties();
props.put("linger.ms", "15000");
ProducerFactoryBean<String, String> producer =
new ProducerFactoryBean<String, String>(producerMetadata, kafkaRule.getBrokersAsString(), props);
new ProducerFactoryBean<>(producerMetadata, kafkaRule.getBrokersAsString(), props);
ProducerConfiguration<String, String> config =
new ProducerConfiguration<String, String>(producerMetadata, producer.getObject());
Map<String, ProducerConfiguration<?, ?>> producerConfigurationMap = Collections.<String, ProducerConfiguration<?, ?>>singletonMap(TOPIC, config);
new ProducerConfiguration<>(producerMetadata, producer.getObject());
Map<String, ProducerConfiguration<?, ?>> producerConfigurationMap =
Collections.<String, ProducerConfiguration<?, ?>>singletonMap(TOPIC, config);
kafkaProducerContext.setProducerConfigurations(producerConfigurationMap);
return kafkaProducerContext;
}