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

Changed occurances of map calls on kafka streams to mapValues

Browse Source 
Resolves #357
This commit is contained in:
Danish Garg
2018-04-11 19:38:42 +05:30
committed by Soby Chacko
parent 5543b4ed1e
commit a2592698c9
2 changed files with 13 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java
View File

@@ -73,7 +73,7 @@ class KafkaStreamsMessageConversionDelegate {
String contentType = this.kstreamBindingInformationCatalogue.getContentType(outboundBindTarget);
MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
return outboundBindTarget.map((k, v) -> {
return outboundBindTarget.mapValues((v) -> {
Message<?> message = v instanceof Message<?> ? (Message<?>) v :
MessageBuilder.withPayload(v).build();
Map<String, Object> headers = new HashMap<>(message.getHeaders());
@@ -81,9 +81,9 @@ class KafkaStreamsMessageConversionDelegate {
headers.put(MessageHeaders.CONTENT_TYPE, contentType);
}
MessageHeaders messageHeaders = new MessageHeaders(headers);
return new KeyValue<>(k,
return
messageConverter.toMessage(message.getPayload(),
messageHeaders).getPayload());
messageHeaders).getPayload();
});
}
@@ -137,10 +137,10 @@ class KafkaStreamsMessageConversionDelegate {
processErrorFromDeserialization(bindingTarget, branch[1]);
//first branch above is the branch where the messages are converted, let it go through further processing.
return branch[0].map((o, o2) -> {
KeyValue<Object, Object> objectObjectKeyValue = keyValueThreadLocal.get();
return branch[0].mapValues((o2) -> {
Object objectValue = keyValueThreadLocal.get().value;
keyValueThreadLocal.remove();
return objectObjectKeyValue;
return objectValue;
});
}

15
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java
View File

@@ -27,7 +27,6 @@ import org.apache.commons.logging.LogFactory;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.Consumed;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
@@ -300,18 +299,18 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene
else {
LOG.info("Native decoding is disabled for " + inboundName + ". Inbound message conversion done by Spring Cloud Stream.");
}
stream = stream.map((key, value) -> {
KeyValue<Object, Object> keyValue;
stream = stream.mapValues(value -> {
Object returnValue;
String contentType = bindingProperties.getContentType();
if (!StringUtils.isEmpty(contentType) && !nativeDecoding) {
Message<?> message = MessageBuilder.withPayload(value)
.setHeader(MessageHeaders.CONTENT_TYPE, contentType).build();
keyValue = new KeyValue<>(key, message);
returnValue = message;
} else {
returnValue = value;
}
else {
keyValue = new KeyValue<>(key, value);
}
return keyValue;
return returnValue;
});
return stream;
}