From 7695964e6480a5ccf6b25ccfc93fc7b392f63054 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 23 Jun 2015 19:45:17 -0400 Subject: [PATCH] INTEXT-175: Fix IEC population 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. --- .../outbound/KafkaProducerMessageHandler.java | 31 +++++------ .../kafka/outbound/OutboundTests.java | 53 +++++++++---------- 2 files changed, 37 insertions(+), 47 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index 6c693d6c75..4cfb9e49cd 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -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"; diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java index daa763666a..4be47e3825 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java @@ -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 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 encoder = new StringEncoder(); - ProducerMetadata producerMetadata = new ProducerMetadata(TOPIC, String.class, String.class, new EncoderAdaptingSerializer(encoder), new EncoderAdaptingSerializer(encoder)); + ProducerMetadata producerMetadata = + new ProducerMetadata(TOPIC, String.class, String.class, + new EncoderAdaptingSerializer<>(encoder), new EncoderAdaptingSerializer<>(encoder)); Properties props = new Properties(); props.put("linger.ms", "15000"); ProducerFactoryBean producer = - new ProducerFactoryBean(producerMetadata, kafkaRule.getBrokersAsString(), props); + new ProducerFactoryBean<>(producerMetadata, kafkaRule.getBrokersAsString(), props); ProducerConfiguration config = - new ProducerConfiguration(producerMetadata, producer.getObject()); - Map> producerConfigurationMap = Collections.>singletonMap(TOPIC, config); + new ProducerConfiguration<>(producerMetadata, producer.getObject()); + Map> producerConfigurationMap = + Collections.>singletonMap(TOPIC, config); kafkaProducerContext.setProducerConfigurations(producerConfigurationMap); return kafkaProducerContext; }