@SuppressWarnings("FutureReturnValueIgnored")
public void success(final T result) {
  for (final SuccessCallback<T> callback : successCallbacks) {
    context.submit(
        () -> {
          Span childSpan =
              tracer
                  .buildSpan("success")
                  .addReference(References.FOLLOWS_FROM, parentSpan.context())
                  .withTag(Tags.COMPONENT.getKey(), "success")
                  .start();
          try (Scope childScope = tracer.activateSpan(childSpan)) {
            callback.accept(result);
          } finally {
            childSpan.finish();
          }
          context.getPhaser().arriveAndAwaitAdvance(); // trace reported
        });
  }
}
 

public Future<Object> send(final Object message) {
  final SpanContext parentSpanContext = tracer.activeSpan().context();

  return executor.submit(
      () -> {
        logger.info("Child thread with message '{}' started", message);

        Span span =
            tracer
                .buildSpan("subtask")
                .addReference(References.FOLLOWS_FROM, parentSpanContext)
                .start();
        try (Scope subtaskScope = tracer.activateSpan(span)) {
          // Simulate work - make sure we finish *after* the parent Span.
          parentDoneLatch.await();
        } finally {
          span.finish();
        }

        logger.info("Child thread with message '{}' finished", message);
        return message + "::response";
      });
}
 

public Future<?> tell(final String message) {
  final Span parent = tracer.scopeManager().activeSpan();
  phaser.register();
  return executor.submit(
      () -> {
        Span child =
            tracer
                .buildSpan("received")
                .addReference(References.FOLLOWS_FROM, parent.context())
                .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_CONSUMER)
                .start();
        try (Scope scope = tracer.activateSpan(child)) {
          phaser.arriveAndAwaitAdvance(); // child tracer started
          child.log("received " + message);
          phaser.arriveAndAwaitAdvance(); // assert size
        } finally {
          child.finish();
        }

        phaser.arriveAndAwaitAdvance(); // child tracer finished
        phaser.arriveAndAwaitAdvance(); // assert size
      });
}
 

public Future<String> ask(final String message) {
  final Span parent = tracer.scopeManager().activeSpan();
  phaser.register();
  Future<String> future =
      executor.submit(
          () -> {
            Span span =
                tracer
                    .buildSpan("received")
                    .addReference(References.FOLLOWS_FROM, parent.context())
                    .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_CONSUMER)
                    .start();
            try {
              phaser.arriveAndAwaitAdvance(); // child tracer started
              phaser.arriveAndAwaitAdvance(); // assert size
              return "received " + message;
            } finally {
              span.finish();

              phaser.arriveAndAwaitAdvance(); // child tracer finished
              phaser.arriveAndAwaitAdvance(); // assert size
            }
          });
  return future;
}
 

public static void onMessageEnter(final Object record) {
  if (LocalSpanContext.get(COMPONENT_NAME) != null) {
    LocalSpanContext.get(COMPONENT_NAME).increment();
    return;
  }

  final Tracer tracer = GlobalTracer.get();
  final SpanBuilder builder = tracer
    .buildSpan("onMessage")
    .withTag(Tags.COMPONENT, COMPONENT_NAME)
    .withTag(Tags.SPAN_KIND, Tags.SPAN_KIND_CONSUMER);

  if (record instanceof ConsumerRecord) {
    final ConsumerRecord<?,?> consumerRecord = (ConsumerRecord<?,?>)record;
    final SpanContext spanContext = TracingKafkaUtils.extractSpanContext(consumerRecord.headers(), tracer);
    if (spanContext != null)
      builder.addReference(References.FOLLOWS_FROM, spanContext);
  }

  final Span span = builder.start();
  LocalSpanContext.set(COMPONENT_NAME, span, tracer.activateSpan(span));
}
 

public static void onMessageEnter(final Object msg) {
  if (LocalSpanContext.get(COMPONENT_NAME) != null) {
    LocalSpanContext.get(COMPONENT_NAME).increment();
    return;
  }

  final Tracer tracer = GlobalTracer.get();
  final SpanBuilder builder = tracer
    .buildSpan("onMessage")
    .withTag(Tags.COMPONENT, COMPONENT_NAME)
    .withTag(Tags.SPAN_KIND, Tags.SPAN_KIND_CONSUMER);

  final Message message = (Message)msg;
  if (message.getMessageProperties() != null) {
    final Map<String,Object> headers = message.getMessageProperties().getHeaders();
    final SpanContext spanContext = tracer.extract(Builtin.TEXT_MAP, new HeadersMapExtractAdapter(headers));
    if (spanContext != null)
      builder.addReference(References.FOLLOWS_FROM, spanContext);
  }

  final Span span = builder.start();
  LocalSpanContext.set(COMPONENT_NAME, span, tracer.activateSpan(span));
}
 

public static void handleMessageEnter(final Object function, final Object contextArg, final Object arg0) {
  final Tracer tracer = GlobalTracer.get();
  final SpanBuilder spanBuilder = tracer
    .buildSpan(getFunctionName(function, contextArg))
    .withTag(Tags.COMPONENT, COMPONENT_NAME)
    .withTag(Tags.SPAN_KIND, Tags.SPAN_KIND_SERVER);

  if (arg0 != null) {
    final Record<?> record = (Record<?>)arg0;
    final SpanContext spanContext = tracer.extract(Builtin.TEXT_MAP, new TextMapExtractAdapter(record.getProperties()));
    if (spanContext != null)
      spanBuilder.addReference(References.FOLLOWS_FROM, spanContext);
  }

  final Span span = spanBuilder.start();
  final Scope scope = tracer.activateSpan(span);

  LocalSpanContext.set(COMPONENT_NAME, span, scope);
}
 

@Override
public V call() throws Exception {
  final Tracer tracer = GlobalTracer.get();
  if (verbose) {
    final Span span = tracer
      .buildSpan("callable")
      .withTag(Tags.COMPONENT, "java-concurrent")
      .addReference(References.FOLLOWS_FROM, parent.context())
      .start();
    try (final Scope scope = tracer.activateSpan(span)) {
      return delegate.call();
    }
    finally {
      span.finish();
    }
  }

  try (final Scope scope = tracer.activateSpan(parent)) {
    return delegate.call();
  }
}
 

@Override
public void run() {
  final Tracer tracer = GlobalTracer.get();
  if (verbose) {
    final Span span = tracer
      .buildSpan("runnable")
      .withTag(Tags.COMPONENT, "java-concurrent")
      .addReference(References.FOLLOWS_FROM, parent.context())
      .start();
    try (final Scope scope = tracer.activateSpan(span)) {
      delegate.run();
    }
    finally {
      span.finish();
    }
  }
  else {
    try (final Scope scope = tracer.activateSpan(parent)) {
      delegate.run();
    }
  }
}
 

private static void buildConsumerSpan(final Consumer<?> consumer, final Message<?> message) {
  final Tracer tracer = GlobalTracer.get();
  final SpanContext parentContext = tracer.extract(Builtin.TEXT_MAP, new TextMapAdapter(message.getProperties()));

  final SpanBuilder spanBuilder = tracer
    .buildSpan("receive")
    .withTag(Tags.COMPONENT, COMPONENT_NAME)
    .withTag(Tags.SPAN_KIND, Tags.SPAN_KIND_CONSUMER)
    .withTag("topic", consumer.getTopic())
    .withTag("subscription", consumer.getSubscription())
    .withTag(Tags.PEER_SERVICE, "pulsar");

  if (parentContext != null)
    spanBuilder.addReference(References.FOLLOWS_FROM, parentContext);

  spanBuilder.start().finish();
}
 

public static void onMessageEnter(final Object msg) {
  if (LocalSpanContext.get(COMPONENT_NAME) != null) {
    LocalSpanContext.get(COMPONENT_NAME).increment();
    return;
  }

  final Tracer tracer = GlobalTracer.get();
  final SpanBuilder builder = tracer
    .buildSpan("onMessage")
    .withTag(Tags.COMPONENT, COMPONENT_NAME)
    .withTag(Tags.SPAN_KIND, Tags.SPAN_KIND_CONSUMER);

  SpanContext spanContext = null;
  if (msg instanceof SpanContextContainer)
    spanContext = ((SpanContextContainer)msg).getSpanContext();

  if (spanContext == null)
    spanContext = TracingMessageUtils.extract((Message)msg, tracer);

  if (spanContext != null)
    builder.addReference(References.FOLLOWS_FROM, spanContext);

  final Span span = builder.start();
  LocalSpanContext.set(COMPONENT_NAME, span, tracer.activateSpan(span));
}
 

@Override
public Tracer.SpanBuilder addReference(String referenceType, SpanContext referencedContext) {
    if (referencedContext == null) {
        return this;
    }
    if (!(referencedContext instanceof SofaTracerSpanContext)) {
        return this;
    }
    if (!References.CHILD_OF.equals(referenceType)
        && !References.FOLLOWS_FROM.equals(referenceType)) {
        return this;
    }
    if (references.isEmpty()) {
        // Optimization for 99% situations, when there is only one parent
        references = Collections.singletonList(new SofaTracerSpanReferenceRelationship(
            (SofaTracerSpanContext) referencedContext, referenceType));
    } else {
        if (references.size() == 1) {
            //To ensure order
            references = new ArrayList<>(references);
        }
        references.add(new SofaTracerSpanReferenceRelationship(
            (SofaTracerSpanContext) referencedContext, referenceType));
    }
    return this;
}
 

@Override
public void run() {
	SpanBuilder spanBuilder = GlobalTracer.get().buildSpan("inProduction");
	spanBuilder.addReference(References.FOLLOWS_FROM, parent.context());
	Span span = spanBuilder.start();
	span.setTag(Robot.KEY_SERIAL_NUMBER, String.valueOf(serialNumber));
	try (Scope scope = GlobalTracer.get().scopeManager().activate(span, false)) {
		Robot chassis = createChassis(serialNumber, robotTypeId, scope.span().context());
		// Takes some time to roll the robot over to the painting
		Utils.sleep(10);
		Robot paintedRobot = paintRobot(chassis, paint, scope.span().context());
		completedRobots.put(paintedRobot.getSerialNumber(), paintedRobot);
		jobsInProduction.remove(serialNumber);
	} catch (Throwable t) {
		span.log(OpenTracingUtil.getSpanLogMap(t));
		throw t;
	} finally {
		span.finish();
		parent.finish();
	}
}
 

@Override
public void run() {
	SpanBuilder spanBuilder = GlobalTracer.get().buildSpan("orderJob");
	spanBuilder.withTag(RobotOrder.KEY_ORDER_ID, String.valueOf(order.getOrderId()));
	spanBuilder.addReference(References.FOLLOWS_FROM, parent.context());
	Span span = spanBuilder.start();
	try (Scope scope = GlobalTracer.get().scopeManager().activate(span, false)) {
		Customer customer = validateUser(order.getCustomerId(), scope.span().context());
		Collection<CompletableFuture<Robot>> robots = dispatch(order.getLineItems(), scope.span().context());
		CompletableFuture<Void> allOf = CompletableFuture.allOf(robots.toArray(new CompletableFuture[0]));
		allOf.get();
		List<Robot> collect = robots.stream().map((robot) -> get(robot)).collect(Collectors.toList());

		// TODO verify that all list items got realized - otherwise add errors for the ones missing etc
		completedOrders.put(order.getOrderId(),
				new RealizedOrder(order, customer, collect.toArray(new Robot[0]), null));
	} catch (Throwable t) {
		span.log(OpenTracingUtil.getSpanLogMap(t));
		completedOrders.put(order.getOrderId(), new RealizedOrder(order, null, null, t));
	} finally {
		span.finish();
		parent.finish();
	}
	orderQueue.remove(order.getOrderId());
}
 

/**
 * Verifies that the resource uses the SpanContext extracted from a CoAP request
 * as the parent of the newly created Span.
 */
@Test
public void testHandleRequestExtractsParentTraceContext() {

    final SpanContext extractedContext = mock(SpanContext.class);
    when(tracer.extract(eq(Format.Builtin.BINARY), any(Binary.class))).thenReturn(extractedContext);

    final Request request = new Request(Code.POST);
    request.getOptions().addOption(new Option(CoapOptionInjectExtractAdapter.OPTION_TRACE_CONTEXT));
    final Exchange exchange = new Exchange(request, Origin.REMOTE, mock(Executor.class));
    resource.handleRequest(exchange);

    verify(tracer).buildSpan(eq(Code.POST.toString()));
    verify(spanBuilder).withTag(eq(Tags.SPAN_KIND.getKey()), eq(Tags.SPAN_KIND_SERVER.toString()));
    verify(spanBuilder).addReference(eq(References.CHILD_OF), eq(extractedContext));
}
 

@Test
public void testSingleExtractedFollowsFrom() {
    APMTracerTest.TestTraceRecorder testTraceReporter = new APMTracerTest.TestTraceRecorder();
    Tracer tracer = new APMTracer(testTraceReporter);

    SpanContext spanCtx = extractedTraceState(tracer, TEST_APM_ID1);

    Span span = tracer.buildSpan("root")
            .addReference(References.FOLLOWS_FROM, spanCtx)
            .start();
    span.finish();

    assertEquals(1, testTraceReporter.getTraces().size());

    Trace trace = testTraceReporter.getTraces().get(0);
    assertEquals(1, trace.getNodes().size());
    assertEquals(Consumer.class, trace.getNodes().get(0).getClass());
    assertEquals(((Consumer) trace.getNodes().get(0)).getCorrelationIds().get(0),
            new CorrelationIdentifier(Scope.Interaction, TEST_APM_ID1));
    assertEquals(0, ((Consumer) trace.getNodes().get(0)).getNodes().size());
}
 

@Test
public void testFindPrimaryReferenceSingleChildOfSpanContextWithOtherRefs() {
    Tracer tracer = new APMTracer();

    SpanContext spanCtx1 = extractSpanContext(tracer, TEST_APM_ID1);

    Reference ref1 = new Reference(References.CHILD_OF, spanCtx1);

    Span span2 = tracer.buildSpan("test2").start();
    Reference ref2 = new Reference(References.FOLLOWS_FROM, span2.context());

    Span span3 = tracer.buildSpan("test3").start();
    Reference ref3 = new Reference(References.CHILD_OF, span3.context());
    Span span4 = tracer.buildSpan("test4").start();
    Reference ref4 = new Reference(References.CHILD_OF, span4.context());

    assertEquals(ref1, APMSpan.findPrimaryReference(Arrays.asList(ref1, ref2, ref3, ref4)));
}
 

@Test
public void testFollowFromReference() {
    MockTracer tracer = new MockTracer(MockTracer.Propagator.TEXT_MAP);
    final MockSpan precedent = tracer.buildSpan("precedent").start();

    final MockSpan followingSpan = tracer.buildSpan("follows")
        .addReference(References.FOLLOWS_FROM, precedent.context())
        .start();

    assertEquals(precedent.context().spanId(), followingSpan.parentId());
    assertEquals(1, followingSpan.references().size());

    final MockSpan.Reference followsFromRef = followingSpan.references().get(0);

    assertEquals(new MockSpan.Reference(precedent.context(), References.FOLLOWS_FROM), followsFromRef);
}
 

@Test
public void testMultiReferences() {
    MockTracer tracer = new MockTracer(MockTracer.Propagator.TEXT_MAP);
    final MockSpan parent = tracer.buildSpan("parent").start();
    final MockSpan precedent = tracer.buildSpan("precedent").start();

    final MockSpan followingSpan = tracer.buildSpan("follows")
        .addReference(References.FOLLOWS_FROM, precedent.context())
        .asChildOf(parent.context())
        .start();

    assertEquals(parent.context().spanId(), followingSpan.parentId());
    assertEquals(2, followingSpan.references().size());

    final MockSpan.Reference followsFromRef = followingSpan.references().get(0);
    final MockSpan.Reference parentRef = followingSpan.references().get(1);

    assertEquals(new MockSpan.Reference(precedent.context(), References.FOLLOWS_FROM), followsFromRef);
    assertEquals(new MockSpan.Reference(parent.context(), References.CHILD_OF), parentRef);
}
 

@Test
public void testMultiReferencesBaggage() {
    MockTracer tracer = new MockTracer(MockTracer.Propagator.TEXT_MAP);
    final MockSpan parent = tracer.buildSpan("parent").start();
    parent.setBaggageItem("parent", "foo");
    final MockSpan precedent = tracer.buildSpan("precedent").start();
    precedent.setBaggageItem("precedent", "bar");

    final MockSpan followingSpan = tracer.buildSpan("follows")
        .addReference(References.FOLLOWS_FROM, precedent.context())
        .asChildOf(parent.context())
        .start();

    assertEquals("foo", followingSpan.getBaggageItem("parent"));
    assertEquals("bar", followingSpan.getBaggageItem("precedent"));
}
 

public void success(final T result) {
  for (final SuccessCallback<T> callback : successCallbacks) {
    context.submit(
        new Runnable() {
          @Override
          public void run() {
            Span childSpan = tracer
                .buildSpan("success")
                .addReference(References.FOLLOWS_FROM, parentSpan.context())
                .withTag(Tags.COMPONENT.getKey(), "success")
                .start();
            try (Scope childScope = tracer.activateSpan(childSpan)) {
              callback.accept(result);
            } finally {
              childSpan.finish();
            }
            context.getPhaser().arriveAndAwaitAdvance(); // trace reported
          }
        });
  }
}
 

public void tell(final String message) {
  final Span parent = tracer.scopeManager().activeSpan();
  phaser.register();
  executor.submit(
      new Runnable() {
        @Override
        public void run() {
          Span child = tracer
              .buildSpan("received")
              .addReference(References.FOLLOWS_FROM, parent.context())
              .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_CONSUMER)
              .start();
          try (Scope scope = tracer.activateSpan(child)) {
            phaser.arriveAndAwaitAdvance(); // child tracer started
            child.log("received " + message);
            phaser.arriveAndAwaitAdvance(); // assert size
          } finally {
            child.finish();
          }

          phaser.arriveAndAwaitAdvance(); // child tracer finished
          phaser.arriveAndAwaitAdvance(); // assert size
        }
      });
}
 

public static void dfs(Vertex node, Consumer<Vertex> vertexConsumer) {
  vertexConsumer.accept(node);
  Iterator<Edge> edges = node.edges(Direction.OUT, References.CHILD_OF);
  while (edges.hasNext()) {
    Edge edge = edges.next();
    dfs(edge.inVertex(), vertexConsumer);
  }
}
 

public static void dfs(Vertex node, BiConsumer<Vertex, Vertex> vertexConsumer) {
  Iterator<Edge> edges = node.edges(Direction.OUT, References.CHILD_OF);
  if (!edges.hasNext()) {
    vertexConsumer.accept(node, null);
  }
  while (edges.hasNext()) {
    Edge edge = edges.next();
    vertexConsumer.accept(node, edge.inVertex());
    dfs(edge.inVertex(), vertexConsumer);
  }
}
 

public static Vertex parent(Vertex vertex) {
  Iterator<Edge> edges = vertex.edges(Direction.IN, References.CHILD_OF);
  if (!edges.hasNext()) {
    return null;
  }
  return edges.next().outVertex();
}
 

public static List<Vertex> children(Vertex vertex) {
  Iterator<Edge> edges = vertex.edges(Direction.OUT, References.CHILD_OF);
  List<Vertex> vertices = new ArrayList<>();
  while (edges.hasNext()) {
    Edge edge = edges.next();
    vertices.add(edge.inVertex());
  }
  return Collections.unmodifiableList(vertices);
}
 

@Override
public ApmSpanBuilder addReference(String referenceType, SpanContext referencedContext) {
    if (References.CHILD_OF.equals(referenceType)) {
        asChildOf(referencedContext);
    }
    return this;
}
 

static Optional<Scope> buildReceiveSpan(MessageProperties messageProperties, Tracer tracer) {
  Optional<SpanContext> context = findParent(messageProperties, tracer);
  Tracer.SpanBuilder spanBuilder =
      tracer
          .buildSpan(RabbitMqTracingTags.SPAN_KIND_CONSUMER)
          .ignoreActiveSpan()
          .withTag(Tags.SPAN_KIND.getKey(), RabbitMqTracingTags.SPAN_KIND_CONSUMER);

  context.ifPresent(spanContext -> spanBuilder.addReference(References.FOLLOWS_FROM, spanContext));
  Scope scope = tracer.scopeManager().activate(spanBuilder.start());

  return Optional.of(scope);
}
 

@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
  log.trace("Processing message before sending it to the channel");
  boolean isConsumer = message.getHeaders().containsKey(Headers.MESSAGE_SENT_FROM_CLIENT);

  SpanBuilder spanBuilder = tracer.buildSpan(getOperationName(channel, isConsumer))
      .withTag(Tags.SPAN_KIND.getKey(), isConsumer ? Tags.SPAN_KIND_CONSUMER : Tags.SPAN_KIND_PRODUCER)
      .withTag(Tags.COMPONENT.getKey(), COMPONENT_NAME)
      .withTag(Tags.MESSAGE_BUS_DESTINATION.getKey(), getChannelName(channel));

  MessageTextMap<?> carrier = new MessageTextMap<>(message, isKafkaBinder);
  SpanContext extractedContext = tracer.extract(Format.Builtin.TEXT_MAP, carrier);
  if (isConsumer) {
    spanBuilder.addReference(References.FOLLOWS_FROM, extractedContext);
  } else if (tracer.activeSpan() == null) {
    // it's a client but active span is null
    // This is a fallback we try to add extractedContext in case there is something
    spanBuilder.asChildOf(extractedContext);
  }

  Span span = spanBuilder.startActive(true).span();

  if (isConsumer) {
    log.trace("Adding 'messageConsumed' header");
    carrier.put(Headers.MESSAGE_CONSUMED, "true");
    // TODO maybe we should remove Headers.MESSAGE_SENT_FROM_CLIENT header here?
  } else {
    log.trace("Adding 'messageSent' header");
    carrier.put(Headers.MESSAGE_SENT_FROM_CLIENT, "true");
  }

  tracer.inject(span.context(), Format.Builtin.TEXT_MAP, carrier);
  return carrier.getMessage();
}
 

public static Object aroundReceiveStart(final Object thiz, final Object message) {
  if (!(message instanceof TracedMessage) && LocalSpanContext.get(COMPONENT_NAME) != null) {
    LocalSpanContext.get(COMPONENT_NAME).increment();
    return message;
  }

  final Tracer tracer = GlobalTracer.get();
  final SpanBuilder spanBuilder = tracer
    .buildSpan("receive")
    .withTag(Tags.COMPONENT, COMPONENT_NAME)
    .withTag(Tags.SPAN_KIND, Tags.SPAN_KIND_CONSUMER);

  final TracedMessage<?> tracedMessage;
  if (message instanceof TracedMessage) {
    tracedMessage = (TracedMessage<?>)message;
    spanBuilder.addReference(References.FOLLOWS_FROM, tracedMessage.spanContext(tracer));
  }
  else {
    tracedMessage = null;
    spanBuilder.withTag(Tags.MESSAGE_BUS_DESTINATION, ((AbstractActor)thiz).getSelf().path().toString());
  }

  final Span span = spanBuilder.start();
  final Scope scope = tracer.activateSpan(span);

  LocalSpanContext.set(COMPONENT_NAME, span, scope);

  return tracedMessage != null ? tracedMessage.getMessage() : message;
}