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

formatting

Browse Source
This commit is contained in:
Adrian Cole
2020-02-06 15:36:50 +08:00
parent 3c0932bb32
commit 2f0e3ed2c7
3 changed files with 158 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-cloud-sleuth-core/src/main/java/org/springframework/cloud/sleuth/instrument/messaging/SleuthMessagingProperties.java
View File

@@ -57,9 +57,11 @@ public class SleuthMessagingProperties {
/**
* An array of patterns against which channel names will be matched.
* @see org.springframework.integration.config.GlobalChannelInterceptor#patterns()
* Defaults to any channel name not matching the Hystrix Stream and functional Stream channel names.
* Defaults to any channel name not matching the Hystrix Stream and functional
* Stream channel names.
*/
private String[] patterns = new String[] { "!hystrixStreamOutput*", "*", "!channel*"};
private String[] patterns = new String[] { "!hystrixStreamOutput*", "*",
"!channel*" };
/**
* Enable Spring Integration sleuth instrumentation.

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

@@ -59,7 +59,8 @@ import org.springframework.util.ClassUtils;
*
* @author Marcin Grzejszczak
*/
public final class TracingChannelInterceptor extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor {
public final class TracingChannelInterceptor extends ChannelInterceptorAdapter
implements ExecutorChannelInterceptor {
/**
* Name of the class in Spring Cloud Stream that is a direct channel.
@@ -108,22 +109,25 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
@Autowired
TracingChannelInterceptor(Tracing tracing) {
this(tracing, MessageHeaderPropagation.INSTANCE, MessageHeaderPropagation.INSTANCE);
this(tracing, MessageHeaderPropagation.INSTANCE,
MessageHeaderPropagation.INSTANCE);
}
TracingChannelInterceptor(Tracing tracing, Propagation.Setter<MessageHeaderAccessor, String> setter,
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(setter);
this.extractor = tracing.propagation().extractor(getter);
this.integrationObjectSupportPresent = ClassUtils
.isPresent("org.springframework.integration.context.IntegrationObjectSupport", null);
this.hasDirectChannelClass = ClassUtils.isPresent("org.springframework.integration.channel.DirectChannel",
null);
this.directWithAttributesChannelClass = ClassUtils.isPresent(STREAM_DIRECT_CHANNEL, null)
? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null;
this.integrationObjectSupportPresent = ClassUtils.isPresent(
"org.springframework.integration.context.IntegrationObjectSupport", null);
this.hasDirectChannelClass = ClassUtils
.isPresent("org.springframework.integration.channel.DirectChannel", null);
this.directWithAttributesChannelClass = ClassUtils
.isPresent(STREAM_DIRECT_CHANNEL, null)
? ClassUtils.resolveClassName(STREAM_DIRECT_CHANNEL, null) : null;
}
public static TracingChannelInterceptor create(Tracing tracing) {
@@ -166,7 +170,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
MessageHeaderAccessor headers = mutableHeaderAccessor(retrievedMessage);
TraceContextOrSamplingFlags extracted = this.extractor.extract(headers);
Span span = this.threadLocalSpan.next(extracted);
MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys());
MessageHeaderPropagation.removeAnyTraceHeaders(headers,
this.tracing.propagation().keys());
this.injector.inject(span.context(), headers);
if (!span.isNoop()) {
span.kind(Span.Kind.PRODUCER).name("send").start();
@@ -195,19 +200,24 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
return REMOTE_SERVICE_NAME;
}
private Message<?> outputMessage(Message<?> originalMessage, Message<?> retrievedMessage,
MessageHeaderAccessor additionalHeaders) {
MessageHeaderAccessor headers = MessageHeaderAccessor.getMutableAccessor(originalMessage);
private Message<?> outputMessage(Message<?> originalMessage,
Message<?> retrievedMessage, MessageHeaderAccessor additionalHeaders) {
MessageHeaderAccessor headers = MessageHeaderAccessor
.getMutableAccessor(originalMessage);
if (originalMessage instanceof ErrorMessage) {
ErrorMessage errorMessage = (ErrorMessage) originalMessage;
headers.copyHeaders(MessageHeaderPropagation.propagationHeaders(additionalHeaders.getMessageHeaders(),
headers.copyHeaders(MessageHeaderPropagation.propagationHeaders(
additionalHeaders.getMessageHeaders(),
this.tracing.propagation().keys()));
return new ErrorMessage(errorMessage.getPayload(), isWebSockets(headers) ? headers.getMessageHeaders()
: new MessageHeaders(headers.getMessageHeaders()), errorMessage.getOriginalMessage());
return new ErrorMessage(errorMessage.getPayload(),
isWebSockets(headers) ? headers.getMessageHeaders()
: new MessageHeaders(headers.getMessageHeaders()),
errorMessage.getOriginalMessage());
}
headers.copyHeaders(additionalHeaders.getMessageHeaders());
return new GenericMessage<>(retrievedMessage.getPayload(),
isWebSockets(headers) ? headers.getMessageHeaders() : new MessageHeaders(headers.getMessageHeaders()));
isWebSockets(headers) ? headers.getMessageHeaders()
: new MessageHeaders(headers.getMessageHeaders()));
}
private boolean isWebSockets(MessageHeaderAccessor headerAccessor) {
@@ -217,7 +227,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
private boolean isDirectChannel(MessageChannel channel) {
Class<?> targetClass = AopUtils.getTargetClass(channel);
boolean directChannel = this.hasDirectChannelClass && DirectChannel.class.isAssignableFrom(targetClass);
boolean directChannel = this.hasDirectChannelClass
&& DirectChannel.class.isAssignableFrom(targetClass);
if (!directChannel) {
return false;
}
@@ -232,7 +243,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
}
@Override
public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
public void afterSendCompletion(Message<?> message, MessageChannel channel,
boolean sent, Exception ex) {
if (emptyMessage(message)) {
return;
}
@@ -240,7 +252,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
afterMessageHandled(message, channel, null, ex);
}
if (log.isDebugEnabled()) {
log.debug("Will finish the current span after completion " + this.tracer.currentSpan());
log.debug("Will finish the current span after completion "
+ this.tracer.currentSpan());
}
finishSpan(ex);
}
@@ -257,7 +270,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
MessageHeaderAccessor headers = mutableHeaderAccessor(message);
TraceContextOrSamplingFlags extracted = this.extractor.extract(headers);
Span span = this.threadLocalSpan.next(extracted);
MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys());
MessageHeaderPropagation.removeAnyTraceHeaders(headers,
this.tracing.propagation().keys());
this.injector.inject(span.context(), headers);
if (!span.isNoop()) {
span.kind(Span.Kind.CONSUMER).name("receive").start();
@@ -270,19 +284,21 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
headers.setImmutable();
if (message instanceof ErrorMessage) {
ErrorMessage errorMessage = (ErrorMessage) message;
return new ErrorMessage(errorMessage.getPayload(), headers.getMessageHeaders(),
errorMessage.getOriginalMessage());
return new ErrorMessage(errorMessage.getPayload(),
headers.getMessageHeaders(), errorMessage.getOriginalMessage());
}
return new GenericMessage<>(message.getPayload(), headers.getMessageHeaders());
}
@Override
public void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex) {
public void afterReceiveCompletion(Message<?> message, MessageChannel channel,
Exception ex) {
if (emptyMessage(message)) {
return;
}
if (log.isDebugEnabled()) {
log.debug("Will finish the current span after receive completion " + this.tracer.currentSpan());
log.debug("Will finish the current span after receive completion "
+ this.tracer.currentSpan());
}
finishSpan(ex);
}
@@ -292,7 +308,8 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
* context. It then creates a span for the handler, placing it in scope.
*/
@Override
public Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler) {
public Message<?> beforeHandle(Message<?> message, MessageChannel channel,
MessageHandler handler) {
if (emptyMessage(message)) {
return message;
}
@@ -307,28 +324,34 @@ public final class TracingChannelInterceptor extends ChannelInterceptorAdapter i
consumerSpan.finish();
}
// create and scope a span for the message processor
this.threadLocalSpan.next(TraceContextOrSamplingFlags.create(consumerSpan.context())).name("handle").start();
this.threadLocalSpan
.next(TraceContextOrSamplingFlags.create(consumerSpan.context()))
.name("handle").start();
// remove any trace headers, but don't re-inject as we are synchronously
// processing the
// message and can rely on scoping to access this span later.
MessageHeaderPropagation.removeAnyTraceHeaders(headers, this.tracing.propagation().keys());
MessageHeaderPropagation.removeAnyTraceHeaders(headers,
this.tracing.propagation().keys());
if (log.isDebugEnabled()) {
log.debug("Created a new span in before handle" + consumerSpan);
}
if (message instanceof ErrorMessage) {
return new ErrorMessage((Throwable) message.getPayload(), headers.getMessageHeaders());
return new ErrorMessage((Throwable) message.getPayload(),
headers.getMessageHeaders());
}
headers.setImmutable();
return new GenericMessage<>(message.getPayload(), headers.getMessageHeaders());
}
@Override
public void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler, Exception ex) {
public void afterMessageHandled(Message<?> message, MessageChannel channel,
MessageHandler handler, Exception ex) {
if (emptyMessage(message)) {
return;
}
if (log.isDebugEnabled()) {
log.debug("Will finish the current span after message handled " + this.tracer.currentSpan());
log.debug("Will finish the current span after message handled "
+ this.tracer.currentSpan());
}
finishSpan(ex);
}

165
spring-cloud-sleuth-core/src/test/java/org/springframework/cloud/sleuth/instrument/messaging/TracingChannelInterceptorTest.java
View File

@@ -53,11 +53,10 @@ public class TracingChannelInterceptorTest {
List<Span> spans = new ArrayList<>();
ChannelInterceptor interceptor = TracingChannelInterceptor
.create(Tracing.newBuilder()
.currentTraceContext(ThreadLocalCurrentTraceContext.newBuilder()
.addScopeDecorator(StrictScopeDecorator.create()).build())
.spanReporter(this.spans::add).build());
ChannelInterceptor interceptor = TracingChannelInterceptor.create(Tracing.newBuilder()
.currentTraceContext(ThreadLocalCurrentTraceContext.newBuilder()
.addScopeDecorator(StrictScopeDecorator.create()).build())
.spanReporter(this.spans::add).build());
QueueChannel channel = new QueueChannel();
@@ -86,9 +85,10 @@ public class TracingChannelInterceptorTest {
this.channel.send(MessageBuilder.withPayload("foo").build());
assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
"nativeHeaders");
assertThat(this.spans).hasSize(1).flatExtracting(Span::kind).containsExactly(Span.Kind.PRODUCER);
assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId",
"X-B3-SpanId", "X-B3-Sampled", "nativeHeaders");
assertThat(this.spans).hasSize(1).flatExtracting(Span::kind)
.containsExactly(Span.Kind.PRODUCER);
}
@Test
@@ -98,9 +98,10 @@ public class TracingChannelInterceptorTest {
this.directChannel.send(MessageBuilder.withPayload("foo").build());
assertThat(this.message).isNotNull();
assertThat(this.message.getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
"nativeHeaders");
assertThat(this.spans).flatExtracting(Span::kind).contains(Span.Kind.CONSUMER, Span.Kind.PRODUCER);
assertThat(this.message.getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId",
"X-B3-Sampled", "nativeHeaders");
assertThat(this.spans).flatExtracting(Span::kind).contains(Span.Kind.CONSUMER,
Span.Kind.PRODUCER);
}
@Test
@@ -109,8 +110,9 @@ public class TracingChannelInterceptorTest {
this.channel.send(MessageBuilder.withPayload("foo").build());
assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsOnlyKeys("X-B3-TraceId",
"X-B3-SpanId", "X-B3-Sampled", "spanTraceId", "spanId", "spanSampled");
assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS))
.containsOnlyKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
"spanTraceId", "spanId", "spanSampled");
}
/**
@@ -122,11 +124,13 @@ public class TracingChannelInterceptorTest {
public void producerConsidersOldSpanIds() {
this.channel.addInterceptor(producerSideOnly(this.interceptor));
this.channel.send(MessageBuilder.withPayload("foo").setHeader("X-B3-TraceId", "000000000000000a")
.setHeader("X-B3-ParentSpanId", "000000000000000a").setHeader("X-B3-SpanId", "000000000000000b")
.build());
this.channel.send(MessageBuilder.withPayload("foo")
.setHeader("X-B3-TraceId", "000000000000000a")
.setHeader("X-B3-ParentSpanId", "000000000000000a")
.setHeader("X-B3-SpanId", "000000000000000b").build());
assertThat(this.channel.receive().getHeaders()).containsEntry("X-B3-ParentSpanId", "000000000000000b");
assertThat(this.channel.receive().getHeaders()).containsEntry("X-B3-ParentSpanId",
"000000000000000b");
}
@Test
@@ -140,10 +144,12 @@ public class TracingChannelInterceptorTest {
accessor.setNativeHeader("X-B3-ParentSpanId", "000000000000000a");
accessor.setNativeHeader("X-B3-SpanId", "000000000000000b");
this.channel.send(MessageBuilder.withPayload("foo").copyHeaders(accessor.toMessageHeaders()).build());
this.channel.send(MessageBuilder.withPayload("foo")
.copyHeaders(accessor.toMessageHeaders()).build());
assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsEntry("X-B3-ParentSpanId",
Collections.singletonList("000000000000000b"));
assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS))
.containsEntry("X-B3-ParentSpanId",
Collections.singletonList("000000000000000b"));
}
/**
@@ -156,9 +162,10 @@ public class TracingChannelInterceptorTest {
this.channel.send(MessageBuilder.withPayload("foo").build());
assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
"nativeHeaders");
assertThat(this.spans).hasSize(1).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER);
assertThat(this.channel.receive().getHeaders()).containsKeys("X-B3-TraceId",
"X-B3-SpanId", "X-B3-Sampled", "nativeHeaders");
assertThat(this.spans).hasSize(1).flatExtracting(Span::kind)
.containsExactly(Span.Kind.CONSUMER);
}
@Test
@@ -167,8 +174,9 @@ public class TracingChannelInterceptorTest {
this.channel.send(MessageBuilder.withPayload("foo").build());
assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS)).containsOnlyKeys("X-B3-TraceId",
"X-B3-SpanId", "X-B3-Sampled", "spanTraceId", "spanId", "spanSampled");
assertThat((Map) this.channel.receive().getHeaders().get(NATIVE_HEADERS))
.containsOnlyKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
"spanTraceId", "spanId", "spanSampled");
}
@Test
@@ -180,9 +188,10 @@ public class TracingChannelInterceptorTest {
channel.send(MessageBuilder.withPayload("foo").build());
assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled",
"nativeHeaders");
assertThat(this.spans).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER, null);
assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId",
"X-B3-SpanId", "X-B3-Sampled", "nativeHeaders");
assertThat(this.spans).flatExtracting(Span::kind)
.containsExactly(Span.Kind.CONSUMER, null);
}
/**
@@ -199,7 +208,8 @@ public class TracingChannelInterceptorTest {
channel.send(MessageBuilder.withPayload("foo").build());
assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled");
assertThat(messages.get(0).getHeaders()).doesNotContainKeys("X-B3-TraceId",
"X-B3-SpanId", "X-B3-Sampled");
}
@Test
@@ -211,8 +221,8 @@ public class TracingChannelInterceptorTest {
channel.send(MessageBuilder.withPayload("foo").build());
assertThat((Map) messages.get(0).getHeaders().get(NATIVE_HEADERS)).doesNotContainKeys("X-B3-TraceId",
"X-B3-SpanId", "X-B3-Sampled");
assertThat((Map) messages.get(0).getHeaders().get(NATIVE_HEADERS))
.doesNotContainKeys("X-B3-TraceId", "X-B3-SpanId", "X-B3-Sampled");
}
@Test
@@ -222,8 +232,8 @@ public class TracingChannelInterceptorTest {
this.channel.send(MessageBuilder.withPayload("foo").build());
this.channel.receive();
assertThat(this.spans).flatExtracting(Span::kind).containsExactlyInAnyOrder(Span.Kind.CONSUMER,
Span.Kind.PRODUCER);
assertThat(this.spans).flatExtracting(Span::kind)
.containsExactlyInAnyOrder(Span.Kind.CONSUMER, Span.Kind.PRODUCER);
}
@Test
@@ -235,7 +245,8 @@ public class TracingChannelInterceptorTest {
channel.send(MessageBuilder.withPayload("foo").build());
assertThat(this.spans).flatExtracting(Span::kind).containsExactly(Span.Kind.CONSUMER, null, Span.Kind.PRODUCER);
assertThat(this.spans).flatExtracting(Span::kind)
.containsExactly(Span.Kind.CONSUMER, null, Span.Kind.PRODUCER);
}
@Test
@@ -246,41 +257,54 @@ public class TracingChannelInterceptorTest {
Map<String, Object> errorChannelHeaders = new HashMap<>();
errorChannelHeaders.put(MessageHeaders.REPLY_CHANNEL, errorsReplyChannel);
errorChannelHeaders.put(MessageHeaders.ERROR_CHANNEL, errorsReplyChannel);
this.channel.send(new ErrorMessage(
new MessagingException(MessageBuilder.withPayload("hi")
.setHeader(TraceMessageHeaders.TRACE_ID_NAME, "000000000000000a")
.setHeader(TraceMessageHeaders.SPAN_ID_NAME, "000000000000000a")
.setReplyChannel(deadReplyChannel).setErrorChannel(deadReplyChannel).build()),
errorChannelHeaders));
this.channel
.send(new ErrorMessage(
new MessagingException(MessageBuilder.withPayload("hi")
.setHeader(TraceMessageHeaders.TRACE_ID_NAME,
"000000000000000a")
.setHeader(TraceMessageHeaders.SPAN_ID_NAME,
"000000000000000a")
.setReplyChannel(deadReplyChannel)
.setErrorChannel(deadReplyChannel).build()),
errorChannelHeaders));
this.message = this.channel.receive();
assertThat(this.message).isNotNull();
String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class);
String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME,
String.class);
assertThat(spanId).isNotNull();
String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class);
String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME,
String.class);
assertThat(traceId).isEqualTo("000000000000000a");
assertThat(spanId).isNotEqualTo("000000000000000a");
assertThat(this.spans).hasSize(2);
assertThat(this.message.getHeaders().getReplyChannel()).isSameAs(errorsReplyChannel);
assertThat(this.message.getHeaders().getErrorChannel()).isSameAs(errorsReplyChannel);
assertThat(this.message.getHeaders().getReplyChannel())
.isSameAs(errorsReplyChannel);
assertThat(this.message.getHeaders().getErrorChannel())
.isSameAs(errorsReplyChannel);
}
@Test
public void errorMessageOriginalMessageRetained() {
this.channel.addInterceptor(this.interceptor);
Message<?> originalMessage = MessageBuilder.withPayload("Hello").setHeader("header", "value").build();
Message<?> failedMessage = MessageBuilder.fromMessage(originalMessage).removeHeader("header").build();
this.channel.send(
new ErrorMessage(new MessagingException(failedMessage), originalMessage.getHeaders(), originalMessage));
Message<?> originalMessage = MessageBuilder.withPayload("Hello")
.setHeader("header", "value").build();
Message<?> failedMessage = MessageBuilder.fromMessage(originalMessage)
.removeHeader("header").build();
this.channel.send(new ErrorMessage(new MessagingException(failedMessage),
originalMessage.getHeaders(), originalMessage));
this.message = this.channel.receive();
assertThat(this.message).isNotNull();
assertThat(this.message).isInstanceOfSatisfying(ErrorMessage.class, errorMessage -> {
assertThat(errorMessage.getOriginalMessage()).isSameAs(originalMessage);
assertThat(errorMessage.getHeaders().get("header")).isEqualTo("value");
});
assertThat(this.message).isInstanceOfSatisfying(ErrorMessage.class,
errorMessage -> {
assertThat(errorMessage.getOriginalMessage())
.isSameAs(originalMessage);
assertThat(errorMessage.getHeaders().get("header"))
.isEqualTo("value");
});
}
@Test
@@ -289,14 +313,17 @@ public class TracingChannelInterceptorTest {
Map<String, Object> errorChannelHeaders = new HashMap<>();
errorChannelHeaders.put(TraceMessageHeaders.TRACE_ID_NAME, "000000000000000a");
errorChannelHeaders.put(TraceMessageHeaders.SPAN_ID_NAME, "000000000000000a");
this.channel.send(new ErrorMessage(new MessagingException("exception"), errorChannelHeaders));
this.channel.send(new ErrorMessage(new MessagingException("exception"),
errorChannelHeaders));
this.message = this.channel.receive();
assertThat(this.message).isNotNull();
String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME, String.class);
String spanId = this.message.getHeaders().get(TraceMessageHeaders.SPAN_ID_NAME,
String.class);
assertThat(spanId).isNotNull();
String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME, String.class);
String traceId = this.message.getHeaders().get(TraceMessageHeaders.TRACE_ID_NAME,
String.class);
assertThat(traceId).isEqualTo("000000000000000a");
assertThat(spanId).isNotEqualTo("000000000000000a");
assertThat(this.spans).hasSize(2);
@@ -327,7 +354,8 @@ public class TracingChannelInterceptorTest {
headers.put(AmqpHeaders.RECEIVED_ROUTING_KEY, "hello");
channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers)));
assertThat(this.spans).flatExtracting(Span::remoteServiceName).contains("rabbitmq");
assertThat(this.spans).flatExtracting(Span::remoteServiceName)
.contains("rabbitmq");
}
@Test
@@ -340,7 +368,8 @@ public class TracingChannelInterceptorTest {
Map<String, Object> headers = new HashMap<>();
channel.send(MessageBuilder.createMessage("foo", new MessageHeaders(headers)));
assertThat(this.spans).flatExtracting(Span::remoteServiceName).containsOnly("broker", null);
assertThat(this.spans).flatExtracting(Span::remoteServiceName)
.containsOnly("broker", null);
}
ChannelInterceptor producerSideOnly(ChannelInterceptor delegate) {
@@ -351,7 +380,8 @@ public class TracingChannelInterceptorTest {
}
@Override
public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
public void afterSendCompletion(Message<?> message, MessageChannel channel,
boolean sent, Exception ex) {
delegate.afterSendCompletion(message, channel, sent, ex);
}
};
@@ -365,24 +395,29 @@ public class TracingChannelInterceptorTest {
}
@Override
public void afterReceiveCompletion(Message<?> message, MessageChannel channel, Exception ex) {
public void afterReceiveCompletion(Message<?> message, MessageChannel channel,
Exception ex) {
delegate.afterReceiveCompletion(message, channel, ex);
}
};
}
ExecutorChannelInterceptor executorSideOnly(ChannelInterceptor delegate) {
class ExecutorSideOnly extends ChannelInterceptorAdapter implements ExecutorChannelInterceptor {
class ExecutorSideOnly extends ChannelInterceptorAdapter
implements ExecutorChannelInterceptor {
@Override
public Message<?> beforeHandle(Message<?> message, MessageChannel channel, MessageHandler handler) {
return ((ExecutorChannelInterceptor) delegate).beforeHandle(message, channel, handler);
public Message<?> beforeHandle(Message<?> message, MessageChannel channel,
MessageHandler handler) {
return ((ExecutorChannelInterceptor) delegate).beforeHandle(message,
channel, handler);
}
@Override
public void afterMessageHandled(Message<?> message, MessageChannel channel, MessageHandler handler,
Exception ex) {
((ExecutorChannelInterceptor) delegate).afterMessageHandled(message, channel, handler, ex);
public void afterMessageHandled(Message<?> message, MessageChannel channel,
MessageHandler handler, Exception ex) {
((ExecutorChannelInterceptor) delegate).afterMessageHandled(message,
channel, handler, ex);
}
}