TracingMessagePostProcessor implements MessagePostProcessor { @Override public Message postProcessMessage(Message message) { MessageProducerRequest request = new MessageProducerRequest(message); TraceContext maybeParent = currentTraceContext.get(); Span span; if (maybeParent == null) { TraceContextOrSamplingFlags extracted = springRabbitTracing.extractAndClearTraceIdHeaders(extractor, request, message); span = springRabbitTracing.nextMessagingSpan(sampler, request, extracted); } else { span = tracer.newChild(maybeParent); } if (!span.isNoop()) { span.kind(PRODUCER).name("publish"); if (remoteServiceName != null) span.remoteServiceName(remoteServiceName); long timestamp = tracing.clock(span.context()).currentTimeMicroseconds(); span.start(timestamp).finish(timestamp); } injector.inject(span.context(), request); return message; } TracingMessagePostProcessor(SpringRabbitTracing springRabbitTracing); @Override Message postProcessMessage(Message message);  }
@Test public void should_resume_headers() { Message message = MessageBuilder.withBody(new byte[0]).build(); message.getMessageProperties().setHeader("b3", B3SingleFormat.writeB3SingleFormat(parent)); Message postProcessMessage = tracingMessagePostProcessor.postProcessMessage(message); assertThat(spans.get(0).parentId()).isEqualTo(parent.spanIdString()); Map<String, Object> headers = postProcessMessage.getMessageProperties().getHeaders(); assertThat(headers.get("b3").toString()).endsWith("-" + spans.get(0).id() + "-1"); }
@Test public void should_retain_baggage() { Message message = MessageBuilder.withBody(new byte[0]).build(); message.getMessageProperties().setHeader("b3", B3SingleFormat.writeB3SingleFormat(parent)); message.getMessageProperties().setHeader(BAGGAGE_FIELD_KEY, ""); Message postProcessMessage = tracingMessagePostProcessor.postProcessMessage(message); assertThat(spans.get(0).parentId()).isEqualTo(parent.spanIdString()); Map<String, Object> headers = postProcessMessage.getMessageProperties().getHeaders(); assertThat(headers.get("b3").toString()).endsWith("-" + spans.get(0).id() + "-1"); assertThat(headers.get(BAGGAGE_FIELD_KEY).toString()).isEmpty(); }
@Test public void should_prefer_current_span() { Message message = MessageBuilder.withBody(new byte[0]).build(); message.getMessageProperties().setHeader("b3", B3SingleFormat.writeB3SingleFormat(grandparent)); Message postProcessMessage; try (Scope scope = tracing.currentTraceContext().newScope(parent)) { postProcessMessage = tracingMessagePostProcessor.postProcessMessage(message); } assertThat(spans.get(0).parentId()).isEqualTo(parent.spanIdString()); Map<String, Object> headers = postProcessMessage.getMessageProperties().getHeaders(); assertThat(headers.get("b3").toString()).endsWith("-" + spans.get(0).id() + "-1"); }
@Test public void should_add_b3_single_header_to_message() { Message message = MessageBuilder.withBody(new byte[0]).build(); Message postProcessMessage = tracingMessagePostProcessor.postProcessMessage(message); assertThat(postProcessMessage.getMessageProperties().getHeaders()) .containsOnlyKeys("b3"); assertThat(postProcessMessage.getMessageProperties().getHeaders().get("b3").toString()) .matches("^[0-9a-f]{16}-[0-9a-f]{16}-1$"); }
@Test public void should_report_span() { Message message = MessageBuilder.withBody(new byte[0]).build(); tracingMessagePostProcessor.postProcessMessage(message); assertThat(spans).hasSize(1); }
@Test public void should_set_remote_service() { Message message = MessageBuilder.withBody(new byte[0]).build(); tracingMessagePostProcessor.postProcessMessage(message); assertThat(spans.get(0).remoteServiceName()) .isEqualTo("my-exchange"); }