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

INTEXT-175 Replace onInit() check for IntegrationEvaluationContext

Browse Source 
- replace the `onInit()` check for an `IntegrationEvaluationContext` with a getter
This commit is contained in:
Marius Bogoevici
2015-06-18 15:18:20 -04:00
committed by Artem Bilan
parent 0956c0ad5a
commit b77f621c51
1 changed files with 11 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -70,28 +70,31 @@ public class KafkaProducerMessageHandler extends AbstractMessageHandler
return this.kafkaProducerContext;
}
@Override
protected void onInit() throws Exception {
Assert.notNull(this.evaluationContext);
}
@Override
protected void handleMessageInternal(final Message<?> message) throws Exception {
String topic = this.topicExpression != null ?
this.topicExpression.getValue(this.evaluationContext, message, String.class)
this.topicExpression.getValue(getEvaluationContext(), message, String.class)
: message.getHeaders().get(KafkaHeaders.TOPIC, String.class);
Integer partitionId = this.partitionExpression != null ?
this.partitionExpression.getValue(this.evaluationContext, message, Integer.class)
this.partitionExpression.getValue(getEvaluationContext(), message, Integer.class)
: message.getHeaders().get(KafkaHeaders.PARTITION_ID, Integer.class);
Object messageKey = this.messageKeyExpression != null
? this.messageKeyExpression.getValue(this.evaluationContext, message)
? this.messageKeyExpression.getValue(getEvaluationContext(), 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";