SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

simplify KafkaMessageChannelBinder ctor call

Browse Source
This commit is contained in:
Matt Benson
2016-05-06 10:48:11 -05:00
committed by Marius Bogoevici
parent 216d5a1ac6
commit 9119e794c8
2 changed files with 6 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -183,16 +183,16 @@ public class KafkaMessageChannelBinder
this.zookeeperConnect = zookeeperConnect;
this.brokers = brokers;
this.zkAddress = zkAddress;
if (headersToMap.length > 0) {
if (ObjectUtils.isEmpty(headersToMap)) {
this.headersToMap = BinderHeaders.STANDARD_HEADERS;
}
else {
String[] combinedHeadersToMap = Arrays.copyOfRange(BinderHeaders.STANDARD_HEADERS, 0,
BinderHeaders.STANDARD_HEADERS.length + headersToMap.length);
System.arraycopy(headersToMap, 0, combinedHeadersToMap, BinderHeaders.STANDARD_HEADERS.length,
headersToMap.length);
this.headersToMap = combinedHeadersToMap;
}
else {
this.headersToMap = BinderHeaders.STANDARD_HEADERS;
}
}
String getZkAddress() {
@@ -793,7 +793,6 @@ public class KafkaMessageChannelBinder
}
@Override
@SuppressWarnings("unchecked")
protected Object handleRequestMessage(Message<?> requestMessage) {
if (HeaderMode.embeddedHeaders.equals(consumerProperties.getHeaderMode())) {
MessageValues messageValues = extractMessageValues(requestMessage);

7
spring-cloud-stream-binders/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java
View File

@@ -32,7 +32,6 @@ import org.springframework.integration.codec.Codec;
import org.springframework.integration.kafka.support.LoggingProducerListener;
import org.springframework.integration.kafka.support.ProducerListener;
import org.springframework.integration.kafka.support.ZookeeperConnect;
import org.springframework.util.ObjectUtils;
/**
* @author David Turanski
@@ -71,10 +70,8 @@ public class KafkaBinderConfiguration {
String[] headers = kafkaBinderConfigurationProperties.getHeaders();
String kafkaConnectionString = kafkaBinderConfigurationProperties.getKafkaConnectionString();
String zkConnectionString = kafkaBinderConfigurationProperties.getZkConnectionString();
KafkaMessageChannelBinder kafkaMessageChannelBinder = ObjectUtils.isEmpty(headers) ?
new KafkaMessageChannelBinder(zookeeperConnect(), kafkaConnectionString, zkConnectionString)
: new KafkaMessageChannelBinder(zookeeperConnect(), kafkaConnectionString, zkConnectionString,
headers);
KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder(
zookeeperConnect(), kafkaConnectionString, zkConnectionString, headers);
kafkaMessageChannelBinder.setCodec(codec);
kafkaMessageChannelBinder.setOffsetUpdateTimeWindow(kafkaBinderConfigurationProperties.getOffsetUpdateTimeWindow());
kafkaMessageChannelBinder.setOffsetUpdateCount(kafkaBinderConfigurationProperties.getOffsetUpdateCount());