content: /*
   * Copyright The OpenTelemetry Authors
   * SPDX-License-Identifier: Apache-2.0
   */
  
  package io.opentelemetry.javaagent.instrumentation.jms.v1_1;
  
  import static io.opentelemetry.api.trace.SpanKind.CONSUMER;
  import static io.opentelemetry.api.trace.SpanKind.PRODUCER;
  import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.equalTo;
  import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.satisfies;
  import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_NAME;
  import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_DESTINATION_TEMPORARY;
  import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_MESSAGE_ID;
  import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_OPERATION;
  import static io.opentelemetry.semconv.incubating.MessagingIncubatingAttributes.MESSAGING_SYSTEM;
  import static org.assertj.core.api.Assertions.assertThat;
  import static org.junit.jupiter.params.provider.Arguments.arguments;
  
  import io.opentelemetry.instrumentation.testing.internal.AutoCleanupExtension;
  import io.opentelemetry.instrumentation.testing.junit.AgentInstrumentationExtension;
  import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;
  import io.opentelemetry.sdk.testing.assertj.AttributeAssertion;
  import io.opentelemetry.sdk.trace.data.LinkData;
  import io.opentelemetry.sdk.trace.data.SpanData;
  import java.io.File;
  import java.nio.file.Files;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicReference;
  import java.util.stream.Stream;
  import javax.jms.Connection;
  import javax.jms.Destination;
  import javax.jms.JMSException;
  import javax.jms.Message;
  import javax.jms.MessageConsumer;
  import javax.jms.MessageProducer;
  import javax.jms.Session;
  import javax.jms.TextMessage;
  import org.assertj.core.api.AbstractAssert;
  import org.hornetq.api.core.TransportConfiguration;
  import org.hornetq.api.core.client.ClientSession;
  import org.hornetq.api.core.client.ClientSessionFactory;
  import org.hornetq.api.core.client.HornetQClient;
  import org.hornetq.api.core.client.ServerLocator;
  import org.hornetq.api.jms.HornetQJMSClient;
  import org.hornetq.api.jms.JMSFactoryType;
  import org.hornetq.core.config.Configuration;
  import org.hornetq.core.config.CoreQueueConfiguration;
  import org.hornetq.core.config.impl.ConfigurationImpl;
  import org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory;
  import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory;
  import org.hornetq.core.server.HornetQServer;
  import org.hornetq.core.server.HornetQServers;
  import org.hornetq.jms.client.HornetQConnectionFactory;
  import org.junit.jupiter.api.AfterAll;
  import org.junit.jupiter.api.BeforeAll;
  import org.junit.jupiter.api.extension.RegisterExtension;
  import org.junit.jupiter.params.ParameterizedTest;
  import org.junit.jupiter.params.provider.Arguments;
  import org.junit.jupiter.params.provider.MethodSource;
  
  @SuppressWarnings("deprecation") // using deprecated semconv
  public class Jms2InstrumentationTest {
  
    @RegisterExtension
    static final InstrumentationExtension testing = AgentInstrumentationExtension.create();
  
    @RegisterExtension static final AutoCleanupExtension cleanup = AutoCleanupExtension.create();
  
    static HornetQServer server;
    static HornetQConnectionFactory connectionFactory;
    static Session session;
    static Connection connection;
  
    @BeforeAll
    static void setUp() throws Exception {
      File tempDir = Files.createTempDirectory("jmsTempDir").toFile();
      tempDir.deleteOnExit();
  
      Configuration config = new ConfigurationImpl();
      config.setBindingsDirectory(tempDir.getPath());
      config.setJournalDirectory(tempDir.getPath());
      config.setCreateBindingsDir(false);
      config.setCreateJournalDir(false);
      config.setSecurityEnabled(false);
      config.setPersistenceEnabled(false);
      config.setQueueConfigurations(
          Collections.singletonList(
              new CoreQueueConfiguration("someQueue", "someQueue", null, true)));
      config.setAcceptorConfigurations(
          new HashSet<>(
              Collections.singletonList(
                  new TransportConfiguration(InVMAcceptorFactory.class.getName()))));
  
      server = HornetQServers.newHornetQServer(config);
      server.start();
  
      ServerLocator serverLocator =
          HornetQClient.createServerLocatorWithoutHA(
              new TransportConfiguration(InVMConnectorFactory.class.getName()));
      ClientSessionFactory sf = serverLocator.createSessionFactory();
      ClientSession clientSession = sf.createSession(false, false, false);
      clientSession.createQueue("jms.queue.someQueue", "jms.queue.someQueue", true);
      clientSession.createQueue("jms.topic.someTopic", "jms.topic.someTopic", true);
      clientSession.close();
      sf.close();
      serverLocator.close();
  
      connectionFactory =
          HornetQJMSClient.createConnectionFactoryWithoutHA(
              JMSFactoryType.CF, new TransportConfiguration(InVMConnectorFactory.class.getName()));
      connection = connectionFactory.createConnection();
      connection.start();
      session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      session.run();
    }
  
    @AfterAll
    static void tearDown() throws Exception {
      if (session != null) {
        session.close();
      }
      if (connection != null) {
        connection.close();
      }
      if (connectionFactory != null) {
        connectionFactory.close();
      }
      if (server != null) {
        server.stop();
      }
    }
  
    @MethodSource("destinationArguments")
    @ParameterizedTest
    void testMessageConsumer(
        DestinationFactory destinationFactory, String destinationName, boolean isTemporary)
        throws JMSException {
  
      // given
      Destination destination = destinationFactory.create(session);
      TextMessage sentMessage = session.createTextMessage("a message");
  
      MessageProducer producer = session.createProducer(destination);
      cleanup.deferCleanup(producer);
      MessageConsumer consumer = session.createConsumer(destination);
      cleanup.deferCleanup(consumer);
  
      // when
      testing.runWithSpan("producer parent", () -> producer.send(sentMessage));
  
      TextMessage receivedMessage =
          testing.runWithSpan("consumer parent", () -> (TextMessage) consumer.receive());
  
      // then
      assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
  
      String messageId = receivedMessage.getJMSMessageID();
  
      AtomicReference<SpanData> producerSpan = new AtomicReference<>();
      testing.waitAndAssertTraces(
          trace -> {
            trace.hasSpansSatisfyingExactly(
                span -> span.hasName("producer parent").hasNoParent(),
                span ->
                    span.hasName(destinationName + " publish")
                        .hasKind(PRODUCER)
                        .hasParent(trace.getSpan(0))
                        .hasAttributesSatisfyingExactly(
                            equalTo(MESSAGING_SYSTEM, "jms"),
                            equalTo(MESSAGING_DESTINATION_NAME, destinationName),
                            equalTo(MESSAGING_OPERATION, "publish"),
                            equalTo(MESSAGING_MESSAGE_ID, messageId),
                            messagingTempDestination(isTemporary)));
  
            producerSpan.set(trace.getSpan(1));
          },
          trace ->
              trace.hasSpansSatisfyingExactly(
                  span -> span.hasName("consumer parent").hasNoParent(),
                  span ->
                      span.hasName(destinationName + " receive")
                          .hasKind(CONSUMER)
                          .hasParent(trace.getSpan(0))
                          .hasLinks(LinkData.create(producerSpan.get().getSpanContext()))
                          .hasAttributesSatisfyingExactly(
                              equalTo(MESSAGING_SYSTEM, "jms"),
                              equalTo(MESSAGING_DESTINATION_NAME, destinationName),
                              equalTo(MESSAGING_OPERATION, "receive"),
                              equalTo(MESSAGING_MESSAGE_ID, messageId),
                              messagingTempDestination(isTemporary))));
    }
  
    @MethodSource("destinationArguments")
    @ParameterizedTest
    void testMessageListener(
        DestinationFactory destinationFactory, String destinationName, boolean isTemporary)
        throws Exception {
  
      // given
      Destination destination = destinationFactory.create(session);
      TextMessage sentMessage = session.createTextMessage("a message");
  
      MessageProducer producer = session.createProducer(null);
      cleanup.deferCleanup(producer);
      MessageConsumer consumer = session.createConsumer(destination);
      cleanup.deferCleanup(consumer);
  
      CompletableFuture<TextMessage> receivedMessageFuture = new CompletableFuture<>();
      consumer.setMessageListener(
          message ->
              testing.runWithSpan(
                  "consumer", () -> receivedMessageFuture.complete((TextMessage) message)));
  
      // when
      testing.runWithSpan("producer parent", () -> producer.send(destination, sentMessage));
  
      // then
      TextMessage receivedMessage = receivedMessageFuture.get(10, TimeUnit.SECONDS);
      assertThat(receivedMessage.getText()).isEqualTo(sentMessage.getText());
  
      String messageId = receivedMessage.getJMSMessageID();
  
      testing.waitAndAssertTraces(
          trace ->
              trace.hasSpansSatisfyingExactly(
                  span -> span.hasName("producer parent").hasNoParent(),
                  span ->
                      span.hasName(destinationName + " publish")
                          .hasKind(PRODUCER)
                          .hasParent(trace.getSpan(0))
                          .hasAttributesSatisfyingExactly(
                              equalTo(MESSAGING_SYSTEM, "jms"),
                              equalTo(MESSAGING_DESTINATION_NAME, destinationName),
                              equalTo(MESSAGING_OPERATION, "publish"),
                              equalTo(MESSAGING_MESSAGE_ID, messageId),
                              messagingTempDestination(isTemporary)),
                  span ->
                      span.hasName(destinationName + " process")
                          .hasKind(CONSUMER)
                          .hasParent(trace.getSpan(1))
                          .hasAttributesSatisfyingExactly(
                              equalTo(MESSAGING_SYSTEM, "jms"),
                              equalTo(MESSAGING_DESTINATION_NAME, destinationName),
                              equalTo(MESSAGING_OPERATION, "process"),
                              equalTo(MESSAGING_MESSAGE_ID, messageId),
                              messagingTempDestination(isTemporary)),
                  span -> span.hasName("consumer").hasParent(trace.getSpan(2))));
    }
  
    @MethodSource("emptyReceiveArguments")
    @ParameterizedTest
    void shouldNotEmitTelemetryOnEmptyReceive(
        DestinationFactory destinationFactory, MessageReceiver receiver) throws JMSException {
  
      // given
      Destination destination = destinationFactory.create(session);
  
      MessageConsumer consumer = session.createConsumer(destination);
      cleanup.deferCleanup(consumer);
  
      // when
      Message message = receiver.receive(consumer);
  
      // then
      assertThat(message).isNull();
  
      testing.waitForTraces(0);
    }
  
    private static AttributeAssertion messagingTempDestination(boolean isTemporary) {
      return isTemporary
          ? equalTo(MESSAGING_DESTINATION_TEMPORARY, true)
          : satisfies(MESSAGING_DESTINATION_TEMPORARY, AbstractAssert::isNull);
    }
  
    private static Stream<Arguments> emptyReceiveArguments() {
      DestinationFactory topic = session -> session.createTopic("someTopic");
      DestinationFactory queue = session -> session.createQueue("someQueue");
      MessageReceiver receive = consumer -> consumer.receive(100);
      MessageReceiver receiveNoWait = MessageConsumer::receiveNoWait;
  
      return Stream.of(
          arguments(topic, receive),
          arguments(queue, receive),
          arguments(topic, receiveNoWait),
          arguments(queue, receiveNoWait));
    }
  
    private static Stream<Arguments> destinationArguments() {
      DestinationFactory topic = session -> session.createTopic("someTopic");
      DestinationFactory queue = session -> session.createQueue("someQueue");
      DestinationFactory tempTopic = Session::createTemporaryTopic;
      DestinationFactory tempQueue = Session::createTemporaryQueue;
  
      return Stream.of(
          arguments(topic, "someTopic", false),
          arguments(queue, "someQueue", false),
          arguments(tempTopic, "(temporary)", true),
          arguments(tempQueue, "(temporary)", true));
    }
  
    @FunctionalInterface
    interface DestinationFactory {
  
      Destination create(Session session) throws JMSException;
    }
  
    @FunctionalInterface
    interface MessageReceiver {
  
      Message receive(MessageConsumer consumer) throws JMSException;
    }
  }
  