diff --git a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc index 08115545e..87c4c6ba1 100644 --- a/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc +++ b/spring-cloud-stream-core-docs/src/main/asciidoc/spring-cloud-stream-overview.adoc @@ -1004,7 +1004,13 @@ Effective only for messaging middleware that does not support message headers na Useful when producing data for non-Spring Cloud Stream applications. + Default: `embeddedHeaders`. - +useNativeEncoding:: + When set to `true`, the outbound message is serialized directly by client library, which must be configured correspondingly (e.g. setting an appropriate Kafka producer value serializer). +When this configuration is being used, the outbound message marshalling is not based on the `contentType` of the binding. +When native encoding is used, it is the responsibility of the consumer to use appropriate decoder (ex: Kafka consumer value de-serializer) to deserialize the inbound message. +Also, when native encoding/decoding is used the `headerMode` property is ignored and headers will not be embedded into the message. ++ +Default: `false`. [[contenttypemanagement]] == Content Type and Transformation diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index 8c1ff4981..3aac7c84d 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -43,11 +43,13 @@ import org.springframework.util.MimeType; *
  • {@link #createConsumerDestinationIfNecessary(String, String, ConsumerProperties)}
  • *
  • {@link #createConsumerEndpoint(String, String, CD, ConsumerProperties)}
  • * - * @author Marius Bogoevici - * @param the consumer properties type - * @param

    the producer properties type + * + * @param the consumer properties type + * @param

    the producer properties type * @param the consumer destination type * @param the producer destination type + * @author Marius Bogoevici + * @author Ilayaperumal Gopinathan * @since 1.1 */ public abstract class AbstractMessageChannelBinder @@ -84,6 +86,7 @@ public abstract class AbstractMessageChannelBinder(destination, null, outputChannel, producerMessageHandler instanceof Lifecycle ? (Lifecycle) producerMessageHandler : null) { @@ -131,6 +134,7 @@ public abstract class AbstractMessageChannelBinder * *

    + * * @param destination the name of the target destination * @param producerProperties the producer properties * @return the message handler for sending data to the target middleware @@ -159,6 +164,7 @@ public abstract class AbstractMessageChannelBinder requestMessage) { + if (!(requestMessage.getPayload() instanceof byte[])) { + return requestMessage; + } MessageValues messageValues; if (this.extractEmbeddedHeaders) { try { @@ -303,16 +316,25 @@ public abstract class AbstractMessageChannelBinder message) throws Exception { + Message messageToSend = (this.useNativeEncoding) ? + message : serializeAndEmbedHeadersIfApplicable(message); + this.delegate.handleMessage(messageToSend); + } + + private Message serializeAndEmbedHeadersIfApplicable(Message message) throws Exception { MessageValues transformed = serializePayloadIfNecessary(message); byte[] payload; if (this.embedHeaders) { @@ -346,9 +368,7 @@ public abstract class AbstractMessageChannelBinder