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

Use StreamListener for coercing messages from JSON Span (#386)

Browse Source 
* Use StreamListener for coercing messages from JSON `Span`

Rely on the `contentType` header of the transported message to
tell Spring Cloud Stream how to coerce the message to `Span`.
This commit is contained in:
Marius Bogoevici
2016-09-08 09:56:29 -04:00
committed by Marcin Grzejszczak
parent 10a169ed26
commit d331f5a212
3 changed files with 11 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -74,9 +74,9 @@ public class StreamEnvironmentPostProcessor implements EnvironmentPostProcessor
// Technically this is only needed on the consumer, but it's fine to be explicit
// on producers as well. It puts all consumers in the same "group", meaning they
// compete with each other and only one gets each message.
map.put("spring.cloud.stream.bindings." + SleuthSink.INPUT + ".group",
map.put("spring.cloud.stream.bindings." + SleuthSource.OUTPUT + ".group",
environment.getProperty("spring.sleuth.stream.group", SleuthSink.INPUT));
map.put("spring.cloud.stream.bindings." + SleuthSink.INPUT + ".content-type",
map.put("spring.cloud.stream.bindings." + SleuthSource.OUTPUT + ".content-type",
environment.getProperty("spring.sleuth.stream.content-type", "application/json"));
addOrReplace(environment.getPropertySources(), map);
}

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

@@ -22,7 +22,6 @@ 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;
@@ -43,9 +42,14 @@ public class StreamEnvironmentPostProcessor implements EnvironmentPostProcessor
public void postProcessEnvironment(ConfigurableEnvironment environment,
SpringApplication application) {
Map<String, Object> map = new HashMap<String, Object>();
// Clearing the content type on the inbound channel means that the payload
// of inbound messages can be coerced by a `@StreamListener` method to
// its argument type based on the 'contentType' header of the inbound message.
// Necessary to be done explicitly because the property is set by by
// org.springframework.cloud.sleuth.stream.StreamEnvironmentPostProcessor to
// 'application/json' for outbound channels
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);
}

4
spring-cloud-sleuth-zipkin-stream/src/main/java/org/springframework/cloud/sleuth/zipkin/stream/ZipkinMessageListener.java
View File

@@ -8,6 +8,7 @@ import org.springframework.cloud.sleuth.Span;
import org.springframework.cloud.sleuth.stream.SleuthSink;
import org.springframework.cloud.sleuth.stream.Spans;
import org.springframework.cloud.sleuth.zipkin.stream.ZipkinMessageListener.NotSleuthStreamClient;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.context.annotation.ConditionContext;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Lazy;
@@ -16,7 +17,6 @@ import org.springframework.core.env.Environment;
import org.springframework.core.env.MapPropertySource;
import org.springframework.core.type.AnnotatedTypeMetadata;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
import zipkin.Annotation;
import zipkin.BinaryAnnotation;
import zipkin.BinaryAnnotation.Type;
@@ -62,7 +62,7 @@ public class ZipkinMessageListener {
.metrics(metrics.forTransport("stream")).build();
}
@ServiceActivator(inputChannel = SleuthSink.INPUT)
@StreamListener(SleuthSink.INPUT)
public void sink(Spans input) {
List<zipkin.Span> converted = ConvertToZipkinSpanList.convert(input);
this.collector.accept(converted, Callback.NOOP);