package com.atlassian.messagequeue.internal.core;
   
   import com.atlassian.messagequeue.MessageRunnerKey;
   import com.atlassian.messagequeue.registry.MessageContext;
   import com.atlassian.messagequeue.registry.MessageRunner;
   import com.atlassian.tenant.impl.TenantIdSetter;
   import com.atlassian.workcontext.api.WorkContextDoorway;
   import org.slf4j.Logger;
   import org.slf4j.LoggerFactory;
  import org.slf4j.MDC;
  
  import java.util.Optional;
  import java.util.concurrent.TimeUnit;
  
  import static java.util.Objects.requireNonNull;
  
  /**
   * Consumes a {@code NestedMessage}.
   */
  public class NestedMessageConsumer {
      public static final String MDC_MESSAGE_RUNNER_KEY = "AMQ-messageRunnerKey";
  
      private static final Logger log = LoggerFactory.getLogger(NestedMessageConsumer.class);
  
      /**
       * The point in time that message processing completed (without error) expressed as the number of milliseconds since epoch.
       */
      private static final String MDC_MESSAGE_PROCESSING_TIMESTAMP = "AMQ-messageProcessingTimestamp";
      private static final String MDC_MESSAGE_PROCESSING_DURATION_MILLIS = "AMQ-messageProcessingDurationMillis";
  
      private final MessageRunnerRegistryHelper messageRunnerRegistryHelper;
      private final TenantIdSetter tenantIdSetter;
  
      /**
       * Constructs an instance of {@code NestedMessageConsumer}.
       *
       * @param messageRunnerRegistryHelper the message runner registry helper
       * @param tenantIdSetter the tenant ID setter
       */
      public NestedMessageConsumer(MessageRunnerRegistryHelper messageRunnerRegistryHelper, TenantIdSetter tenantIdSetter) {
          this.messageRunnerRegistryHelper = requireNonNull(messageRunnerRegistryHelper);
          this.tenantIdSetter = requireNonNull(tenantIdSetter);
      }
  
      /**
       * Consumes a {@link NestedMessage}.
       *
       * @param nestedMessage the nested message
       * @param messageContext the message context
       */
      public void consume(NestedMessage nestedMessage, MessageContext messageContext) {
          final String messageId = messageContext.getMessageId().orElse("");
          final String tenantId = nestedMessage.getAttribute(NestedMessageConstants.TENANT_ID_ATTRIBUTE_NAME);
          if (tenantId == null) {
              log.error("Message received that does not contain required '{}' attribute. Message will be acknowledged (messageID: {})", NestedMessageConstants.TENANT_ID_ATTRIBUTE_NAME, messageId);
              messageContext.acknowledge();
              return;
          }
  
          final String messageRunnerKey = nestedMessage.getAttribute(NestedMessageConstants.MESSAGE_RUNNER_KEY_ATTRIBUTE_NAME);
          if (messageRunnerKey == null) {
              log.error("Message received that does not contain required '{}' attribute. Message will be acknowledged (messageID: {})", NestedMessageConstants.MESSAGE_RUNNER_KEY_ATTRIBUTE_NAME, messageId);
              messageContext.acknowledge();
              return;
          }
  
          final Optional<MessageRunner> messageRunner = messageRunnerRegistryHelper.getMessageRunner(MessageRunnerKey.of(messageRunnerKey));
          if (!messageRunner.isPresent()) {
              log.error("No MessageRunner found to process message (messageID: {}, messageRunnerKey: {})", messageId, messageRunnerKey);
              return;
          }
  
          try (WorkContextDoorway workContextDoorway = new WorkContextDoorway().open();
               MDC.MDCCloseable mdcMessageRunnerKey = MDC.putCloseable(MDC_MESSAGE_RUNNER_KEY, messageRunnerKey)) {
              tenantIdSetter.setTenantId(tenantId);
  
              long startTimeNanos = 0;
              if (log.isInfoEnabled()) {
                  log.info("Message processing started (messageId: {}, messageRunnerKey: {})", messageId, messageRunnerKey);
                  startTimeNanos = System.nanoTime();
              }
  
              messageRunner.get().processMessage(messageContext);
  
              if (log.isInfoEnabled()) {
                  final long messageProcessingCompleteTimestamp = System.currentTimeMillis();
                  final long messageProcessingDurationMillis = TimeUnit.MILLISECONDS.convert(System.nanoTime() - startTimeNanos, TimeUnit.NANOSECONDS);
                  try (MDC.MDCCloseable mdcMessageProcessingTimestamp = MDC.putCloseable(MDC_MESSAGE_PROCESSING_TIMESTAMP, String.valueOf(messageProcessingCompleteTimestamp));
                       MDC.MDCCloseable mdcMessageProcessingDurationMillis = MDC.putCloseable(MDC_MESSAGE_PROCESSING_DURATION_MILLIS, String.valueOf(messageProcessingDurationMillis))) {
                      log.info("Message processing completed in {}ms (messageProcessingCompletedTimestamp: {}, messageId: {}, messageRunnerKey: {})",
                              messageProcessingDurationMillis, messageProcessingCompleteTimestamp, messageId, messageRunnerKey);
                  }
              }
  
              if (messageContext.shouldAutoAcknowledgeMessage()) {
                  messageContext.acknowledge();
              }
          } catch (Exception e) {
              log.error("Invoking MessageRunner for message failed with an exception (messageID: {}, messageRunnerKey: {})", messageId, messageRunnerKey, e);
         }
     }
 }