Added option to override message headers (#1049)

without this change there's no easy way to work with message headers for Spring Integration
with this change we're allowing to register beans that will override the default behaviour

fixes gh-1032

docs/src/main/asciidoc/spring-cloud-sleuth.adoc

@@ -1237,6 +1237,12 @@ By default, all channels but `hystrixStreamOutput` channel are included.
 IMPORTANT: When using the `Executor` to build a Spring Integration `IntegrationFlow`, you must use the untraced version of the `Executor`.
 Decorating the Spring Integration Executor Channel with `TraceableExecutorService` causes the spans to be improperly closed.
 
+If you want to customize the way tracing context is read from and written to message headers,
+it's enough for you to register beans of types:
+
+* `Propagation.Setter<MessageHeaderAccessor, String>` - for writing headers to the message
+* `Propagation.Getter<MessageHeaderAccessor, String>` - for reading headers from the message
+
 ==== Spring RabbitMq
 
 We instrument the `RabbitTemplate` so that tracing headers get injected

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TraceSpringIntegrationAutoConfiguration.java

@@ -17,9 +17,11 @@
 package org.springframework.cloud.sleuth.instrument.messaging;
 
 import brave.Tracing;
+import brave.propagation.Propagation;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.cloud.sleuth.autoconfig.TraceAutoConfiguration;
@@ -27,6 +29,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.channel.interceptor.GlobalChannelInterceptorWrapper;
 import org.springframework.integration.config.GlobalChannelInterceptor;
+import org.springframework.messaging.support.MessageHeaderAccessor;
 
 /**
  * {@link org.springframework.boot.autoconfigure.EnableAutoConfiguration
@@ -57,8 +60,22 @@ public class TraceSpringIntegrationAutoConfiguration {
 	}
 
 	@Bean
-	TracingChannelInterceptor traceChannelInterceptor(Tracing tracing) {
-		return new TracingChannelInterceptor(tracing);
+	TracingChannelInterceptor traceChannelInterceptor(Tracing tracing,
+			Propagation.Setter<MessageHeaderAccessor, String> traceMessagePropagationSetter,
+			Propagation.Getter<MessageHeaderAccessor, String> traceMessagePropagationGetter) {
+		return new TracingChannelInterceptor(tracing, traceMessagePropagationSetter, traceMessagePropagationGetter);
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	Propagation.Setter<MessageHeaderAccessor, String> traceMessagePropagationSetter() {
+		return MessageHeaderPropagation.INSTANCE;
+	}
+
+	@Bean
+	@ConditionalOnMissingBean
+	Propagation.Getter<MessageHeaderAccessor, String> traceMessagePropagationGetter() {
+		return MessageHeaderPropagation.INSTANCE;
 	}
 
 }

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptor.java

@@ -20,12 +20,14 @@ import brave.Span;
 import brave.SpanCustomizer;
 import brave.Tracer;
 import brave.Tracing;
+import brave.propagation.Propagation;
 import brave.propagation.ThreadLocalSpan;
 import brave.propagation.TraceContext;
 import brave.propagation.TraceContextOrSamplingFlags;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.springframework.aop.support.AopUtils;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.sleuth.util.SpanNameUtil;
 import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.DirectChannel;
@@ -85,14 +87,20 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
 	final boolean integrationObjectSupportPresent;
 	private final boolean hasDirectChannelClass;
 
+	@Autowired
 	TracingChannelInterceptor(Tracing tracing) {
+		this(tracing, MessageHeaderPropagation.INSTANCE, MessageHeaderPropagation.INSTANCE);
+	}
+
+	TracingChannelInterceptor(Tracing tracing, Propagation.Setter<MessageHeaderAccessor, String> setter,
+			Propagation.Getter<MessageHeaderAccessor, String> getter) {
 		this.tracing = tracing;
 		this.tracer = tracing.tracer();
 		this.threadLocalSpan = ThreadLocalSpan.create(this.tracer);
 		this.injector = tracing.propagation()
-				.injector(MessageHeaderPropagation.INSTANCE);
+				.injector(setter);
 		this.extractor = tracing.propagation()
-				.extractor(MessageHeaderPropagation.INSTANCE);
+				.extractor(getter);
 		this.integrationObjectSupportPresent = ClassUtils.isPresent(
 				"org.springframework.integration.context.IntegrationObjectSupport",
 				null);