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

KStream binder - nativeEncoding and Serde changes

Browse Source
This commit is contained in:
Soby Chacko
2018-01-13 21:48:20 -05:00
committed by Oleg Zhurakousky
parent 3a5aed61c9
commit 9500ccfe46
1 changed files with 14 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-cloud-stream-binder-kstream/src/main/java/org/springframework/cloud/stream/binder/kstream/KStreamBinder.java
View File

@@ -100,23 +100,17 @@ public class KStreamBinder extends
private Serde<?> getKeySerde(ExtendedProducerProperties<KStreamProducerProperties> properties) {
Serde<?> keySerde;
try {
if (properties.isUseNativeEncoding()) {
keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray();
if (StringUtils.hasText(properties.getExtension().getKeySerde())) {
keySerde = Utils.newInstance(properties.getExtension().getKeySerde(), Serde.class);
if (keySerde instanceof Configurable) {
((Configurable) keySerde).configure(streamsConfig.originals());
}
}
else {
if (StringUtils.hasText(properties.getExtension().getKeySerde())) {
keySerde = Utils.newInstance(properties.getExtension().getKeySerde(), Serde.class);
if (keySerde instanceof Configurable) {
((Configurable) keySerde).configure(streamsConfig.originals());
}
}
else {
keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray();
}
keySerde = this.binderConfigurationProperties.getConfiguration().containsKey("key.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("key.serde"), Serde.class) : Serdes.ByteArray();
}
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
@@ -128,12 +122,6 @@ public class KStreamBinder extends
Serde<?> valueSerde;
try {
if (properties.isUseNativeEncoding()) {
valueSerde = this.binderConfigurationProperties.getConfiguration().containsKey("value.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("value.serde"), Serde.class) : Serdes.ByteArray();
}
else {
if (StringUtils.hasText(properties.getExtension().getValueSerde())) {
valueSerde = Utils.newInstance(properties.getExtension().getValueSerde(), Serde.class);
if (valueSerde instanceof Configurable) {
@@ -141,9 +129,13 @@ public class KStreamBinder extends
}
}
else {
valueSerde = Serdes.ByteArray();
valueSerde = this.binderConfigurationProperties.getConfiguration().containsKey("value.serde") ?
Utils.newInstance(this.binderConfigurationProperties.getConfiguration().get("value.serde"), Serde.class) : Serdes.ByteArray();
}
}
else {
valueSerde = Serdes.ByteArray();
}
}
catch (ClassNotFoundException e) {
throw new IllegalStateException("Serde class not found: ", e);
@@ -160,7 +152,7 @@ public class KStreamBinder extends
}
if (!isNativeEncoding) {
outboundBindTarget.map(keyValueMapper).to(name, Produced.with(keySerde, valueSerde));
}
}
else {
outboundBindTarget.to(name, Produced.with(keySerde, valueSerde));
}