Use JSON for span transport with Stream by default

User can still override it and Stream should adjust itself, but
now we have JSON flowing between instrumented apps and the Zipkin
server.

Fixes gh-217

spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/Span.java

@@ -27,6 +27,9 @@ import java.util.Map;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonInclude;
+
 /**
  * Class for gathering and reporting statistics about a block of execution.
  * <p>
@@ -63,6 +66,7 @@ import org.springframework.util.StringUtils;
  * like scoped tracers. Sleuth spans are DTOs, whose sole responsibility is the current
  * span in the trace tree.
  */
+@JsonInclude(JsonInclude.Include.NON_DEFAULT)
 public class Span {
 
 	public static final String SAMPLED_NAME = "X-B3-Sampled";
@@ -129,6 +133,11 @@ public class Span {
 	private final List<Log> logs;
 	private final Span savedSpan;
 
+	@SuppressWarnings("unused")
+	private Span() {
+		this(-1,-1,"dummy",0,Collections.emptyList(),0,false,false,null);
+	}
+
 	/**
 	 * Creates a new span that still tracks tags and logs of the
 	 * current span. This is crucial when continuing spans
@@ -194,6 +203,7 @@ public class Span {
 	 * Return the total amount of time elapsed since start was called, if running, or
 	 * difference between stop and start
 	 */
+	@JsonIgnore
 	public synchronized long getAccumulatedMillis() {
 		if (this.begin == 0) {
 			return 0;
@@ -207,6 +217,7 @@ public class Span {
 	/**
 	 * Has the span been started and not yet stopped?
 	 */
+	@JsonIgnore
 	public synchronized boolean isRunning() {
 		return this.begin != 0 && this.end == 0;
 	}
@@ -253,6 +264,7 @@ public class Span {
 	 * <p>
 	 * Might be null
 	 */
+	@JsonIgnore
 	public Span getSavedSpan() {
 		return this.savedSpan;
 	}

spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/Host.java

@@ -20,6 +20,7 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.nio.ByteBuffer;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonInclude;
 
 /**
@@ -36,12 +37,17 @@ public class Host {
 	private String address;
 	private Integer port;
 
+	@SuppressWarnings("unused")
+	private Host() {
+	}
+
 	public Host(String serviceName, String address, Integer port) {
 		this.serviceName = serviceName;
 		this.address = address;
 		this.port = port;
 	}
 
+	@JsonIgnore
 	public int getIpv4() {
 		InetAddress inetAddress = null;
 		try {

spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/SleuthStreamAutoConfiguration.java

@@ -72,9 +72,9 @@ public class SleuthStreamAutoConfiguration {
 	}
 
 	@Bean
-	public StreamSpanListener sleuthStreamSpanListener(HostLocator endpointLocator,
+	public StreamSpanReporter sleuthStreamSpanReporter(HostLocator endpointLocator,
 			SpanMetricReporter spanMetricReporter) {
-		return new StreamSpanListener(endpointLocator, spanMetricReporter);
+		return new StreamSpanReporter(endpointLocator, spanMetricReporter);
 	}
 
 	@Configuration

spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/Spans.java

@@ -35,6 +35,10 @@ public class Spans {
 
 	private Host host;
 	private List<Span> spans = Collections.emptyList();
+	
+	@SuppressWarnings("unused")
+	private Spans() {
+	}
 
 	public Spans(Host host, List<Span> spans) {
 		this.host = host;

spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/StreamEnvironmentPostProcessor.java

@@ -76,6 +76,8 @@ public class StreamEnvironmentPostProcessor implements EnvironmentPostProcessor
 		// compete with each other and only one gets each message.
 		map.put("spring.cloud.stream.bindings." + SleuthSink.INPUT + ".group",
 				environment.getProperty("spring.sleuth.stream.group", SleuthSink.INPUT));
+		map.put("spring.cloud.stream.bindings." + SleuthSink.INPUT + ".content-type",
+				environment.getProperty("spring.sleuth.stream.content-type", "application/json"));
 		addOrReplace(environment.getPropertySources(), map);
 	}
 

spring-cloud-sleuth-stream/src/main/java/org/springframework/cloud/sleuth/stream/StreamSpanReporter.java

@@ -37,13 +37,13 @@ import org.springframework.integration.annotation.MessageEndpoint;
  * @since 1.0.0
  */
 @MessageEndpoint
-public class StreamSpanListener implements SpanReporter {
+public class StreamSpanReporter implements SpanReporter {
 
 	private Collection<Span> queue = new ConcurrentLinkedQueue<>();
 	private final HostLocator endpointLocator;
 	private final SpanMetricReporter spanMetricReporter;
 
-	public StreamSpanListener(HostLocator endpointLocator, SpanMetricReporter spanMetricReporter) {
+	public StreamSpanReporter(HostLocator endpointLocator, SpanMetricReporter spanMetricReporter) {
 		this.endpointLocator = endpointLocator;
 		this.spanMetricReporter = spanMetricReporter;
 	}

spring-cloud-sleuth-stream/src/test/java/org/springframework/cloud/sleuth/stream/StreamSpanListenerTests.java

@@ -65,7 +65,7 @@ public class StreamSpanListenerTests {
 	@Autowired Tracer tracer;
 	@Autowired ApplicationContext application;
 	@Autowired ZipkinTestConfiguration test;
-	@Autowired StreamSpanListener listener;
+	@Autowired StreamSpanReporter listener;
 	@Autowired CounterService counterService;
 	@Autowired SpanReporter spanReporter;
 
@@ -139,7 +139,7 @@ public class StreamSpanListenerTests {
 		private List<Span> spans = new ArrayList<>();
 
 		@Autowired
-		StreamSpanListener listener;
+		StreamSpanReporter listener;
 
 		@ServiceActivator(inputChannel=SleuthSource.OUTPUT)
 		public void handle(Message<?> msg) {

spring-cloud-sleuth-zipkin-stream/src/main/java/org/springframework/cloud/sleuth/zipkin/stream/StreamEnvironmentPostProcessor.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.sleuth.zipkin.stream;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.env.EnvironmentPostProcessor;
+import org.springframework.cloud.sleuth.stream.SleuthSink;
+import org.springframework.cloud.sleuth.stream.Spans;
+import org.springframework.core.env.ConfigurableEnvironment;
+import org.springframework.core.env.MapPropertySource;
+import org.springframework.core.env.MutablePropertySources;
+import org.springframework.core.env.PropertySource;
+
+/**
+ * {@link EnvironmentPostProcessor} that sets the default properties for Sleuth Zipkin
+ * Stream.
+ *
+ * @author Dave Syer
+ *
+ * @since 1.0.0
+ */
+public class StreamEnvironmentPostProcessor implements EnvironmentPostProcessor {
+
+	private static final String PROPERTY_SOURCE_NAME = "defaultProperties";
+
+	@Override
+	public void postProcessEnvironment(ConfigurableEnvironment environment,
+			SpringApplication application) {
+		Map<String, Object> map = new HashMap<String, Object>();
+		map.put("spring.cloud.stream.bindings." + SleuthSink.INPUT + ".content-type",
+				environment.getProperty("spring.sleuth.stream.content-type",
+						"application/x-java-object;type=" + Spans.class.getName()));
+		addOrReplace(environment.getPropertySources(), map);
+	}
+
+	private void addOrReplace(MutablePropertySources propertySources,
+			Map<String, Object> map) {
+		MapPropertySource target = null;
+		if (propertySources.contains(PROPERTY_SOURCE_NAME)) {
+			PropertySource<?> source = propertySources.get(PROPERTY_SOURCE_NAME);
+			if (source instanceof MapPropertySource) {
+				target = (MapPropertySource) source;
+				for (String key : map.keySet()) {
+					target.getSource().put(key, map.get(key));
+				}
+			}
+		}
+		if (target == null) {
+			target = new MapPropertySource(PROPERTY_SOURCE_NAME, map);
+		}
+		if (!propertySources.contains(PROPERTY_SOURCE_NAME)) {
+			propertySources.addLast(target);
+		}
+	}
+
+}

spring-cloud-sleuth-zipkin-stream/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,3 @@
+# Environment Post Processor
+org.springframework.boot.env.EnvironmentPostProcessor=\
+org.springframework.cloud.sleuth.zipkin.stream.StreamEnvironmentPostProcessor