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

Allow configuration of Kafka embedded headers in the bus

Browse Source
This commit is contained in:
Marius Bogoevici
2015-09-29 16:13:51 -04:00
committed by Mark Fisher
parent 1a1ebe7b78
commit deecae93a6
1 changed files with 15 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -24,6 +24,7 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.kafka.support.ZookeeperConnect;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
/**
@@ -44,6 +45,8 @@ public class KafkaMessageChannelBinderConfiguration {
private String defaultBrokerPort;
private String[] headers;
private KafkaMessageChannelBinder.Mode mode;
private String offsetStoreTopic;
@@ -81,8 +84,10 @@ public class KafkaMessageChannelBinderConfiguration {
@Bean
KafkaMessageChannelBinder kafkaMessageChannelBinder() {
KafkaMessageChannelBinder kafkaMessageChannelBinder = new KafkaMessageChannelBinder(zookeeperConnect(),
getKafkaConnectionString(), getZkConnectionString());
KafkaMessageChannelBinder kafkaMessageChannelBinder = ObjectUtils.isEmpty(headers) ?
new KafkaMessageChannelBinder(zookeeperConnect(), getKafkaConnectionString(), getZkConnectionString())
: new KafkaMessageChannelBinder(zookeeperConnect(), getKafkaConnectionString(), getZkConnectionString(),
headers);
kafkaMessageChannelBinder.setCodec(codec);
kafkaMessageChannelBinder.setMode(mode);
kafkaMessageChannelBinder.setOffsetStoreTopic(offsetStoreTopic);
@@ -129,6 +134,14 @@ public class KafkaMessageChannelBinderConfiguration {
this.defaultBrokerPort = defaultBrokerPort;
}
public String[] getHeaders() {
return headers;
}
public void setHeaders(String[] headers) {
this.headers = headers;
}
public void setMode(KafkaMessageChannelBinder.Mode mode) {
this.mode = mode;
}