/*
 * Copyright 2018 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License")
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.netflix.spinnaker.echo.pubsub.aws;

import com.amazonaws.ClientConfiguration;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.services.sns.AmazonSNSClientBuilder;
import com.amazonaws.services.sqs.AmazonSQSClientBuilder;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
import com.netflix.spectator.api.Registry;
import com.netflix.spinnaker.echo.artifacts.MessageArtifactTranslator;
import com.netflix.spinnaker.echo.config.AmazonPubsubProperties;
import com.netflix.spinnaker.echo.discovery.DiscoveryActivated;
import com.netflix.spinnaker.echo.pubsub.PubsubEventCreator;
import com.netflix.spinnaker.echo.pubsub.PubsubMessageHandler;
import com.netflix.spinnaker.echo.pubsub.PubsubSubscribers;
import com.netflix.spinnaker.echo.pubsub.model.EventCreator;
import com.netflix.spinnaker.echo.pubsub.model.PubsubSubscriber;
import com.netflix.spinnaker.kork.aws.ARN;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import javax.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Component;

/** * Starts the individual SQS workers (one for each subscription) */
@Component
@ConditionalOnExpression("${pubsub.enabled:false} && ${pubsub.amazon.enabled:false}")
public class SQSSubscriberProvider implements DiscoveryActivated {
  private static final Logger log = LoggerFactory.getLogger(SQSSubscriberProvider.class);

  private final ObjectMapper objectMapper;
  private final AWSCredentialsProvider awsCredentialsProvider;
  private final AmazonPubsubProperties properties;
  private final PubsubSubscribers pubsubSubscribers;
  private final PubsubMessageHandler.Factory pubsubMessageHandlerFactory;
  private final Registry registry;
  private final MessageArtifactTranslator.Factory messageArtifactTranslatorFactory;

  @Autowired
  SQSSubscriberProvider(
      ObjectMapper objectMapper,
      AWSCredentialsProvider awsCredentialsProvider,
      AmazonPubsubProperties properties,
      PubsubSubscribers pubsubSubscribers,
      PubsubMessageHandler.Factory pubsubMessageHandlerFactory,
      Registry registry,
      MessageArtifactTranslator.Factory messageArtifactTranslatorFactory) {
    this.objectMapper = objectMapper;
    this.awsCredentialsProvider = awsCredentialsProvider;
    this.properties = properties;
    this.pubsubSubscribers = pubsubSubscribers;
    this.pubsubMessageHandlerFactory = pubsubMessageHandlerFactory;
    this.registry = registry;
    this.messageArtifactTranslatorFactory = messageArtifactTranslatorFactory;
  }

  @PostConstruct
  public void start() {
    Preconditions.checkNotNull(
        properties, "Can't initialize SQSSubscriberProvider with null properties");

    ExecutorService executorService =
        Executors.newFixedThreadPool(properties.getSubscriptions().size());

    List<PubsubSubscriber> subscribers = new ArrayList<>();

    properties
        .getSubscriptions()
        .forEach(
            (AmazonPubsubProperties.AmazonPubsubSubscription subscription) -> {
              log.info("Bootstrapping SQS for SNS topic: {}", subscription.getTopicARN());
              if (subscription.getTemplatePath() != null
                  && !subscription.getTemplatePath().equals("")) {
                log.info(
                    "Using template: {} for subscription: {}",
                    subscription.getTemplatePath(),
                    subscription.getName());
              }

              ARN queueArn = new ARN(subscription.getQueueARN());

              Optional<MessageArtifactTranslator> messageArtifactTranslator = Optional.empty();
              if (subscription.getMessageFormat() != AmazonPubsubProperties.MessageFormat.NONE) {
                messageArtifactTranslator =
                    Optional.ofNullable(subscription.readTemplatePath())
                        .map(messageArtifactTranslatorFactory::createJinja);
              }
              EventCreator eventCreator = new PubsubEventCreator(messageArtifactTranslator);

              SQSSubscriber worker =
                  new SQSSubscriber(
                      objectMapper,
                      subscription,
                      pubsubMessageHandlerFactory.create(eventCreator),
                      AmazonSNSClientBuilder.standard()
                          .withCredentials(awsCredentialsProvider)
                          .withClientConfiguration(new ClientConfiguration())
                          .withRegion(queueArn.getRegion())
                          .build(),
                      AmazonSQSClientBuilder.standard()
                          .withCredentials(awsCredentialsProvider)
                          .withClientConfiguration(new ClientConfiguration())
                          .withRegion(queueArn.getRegion())
                          .build(),
                      enabled::get,
                      registry);

              try {
                executorService.submit(worker);
                subscribers.add(worker);
                log.debug("Created worker for subscription: {}", subscription.getName());
              } catch (RejectedExecutionException e) {
                log.error("Could not start " + worker.getWorkerName(), e);
              }
            });
    pubsubSubscribers.putAll(subscribers);
  }
}

@@ -1537,16 +1537,86 @@
 ws.ARN;%0A
+import com.netflix.spinnaker.kork.dynamicconfig.DynamicConfigService;%0A
 import j
@@ -1631,24 +1631,24 @@
 .ArrayList;%0A
-
 import java.
@@ -1821,24 +1821,60 @@
 nException;%0A
+import java.util.function.Supplier;%0A
 import javax
@@ -2879,16 +2879,75 @@
 Factory;
+%0A  private final DynamicConfigService dynamicConfigService;
 %0A%0A  @Aut
@@ -3309,16 +3309,65 @@
 rFactory
+,%0A      DynamicConfigService dynamicConfigService
 ) %7B%0A    
@@ -3716,16 +3716,70 @@
 actory;%0A
+    this.dynamicConfigService = dynamicConfigService;%0A
   %7D%0A%0A  @
@@ -6071,20 +6071,27 @@
     
-e
+isE
 nabled
-::get
+Supplier()
 ,%0A  
@@ -6483,16 +6483,16 @@
     %7D);%0A
-
     pubs
@@ -6529,11 +6529,155 @@
 rs);%0A  %7D
+%0A%0A  private Supplier%3CBoolean%3E isEnabledSupplier() %7B%0A    return () -%3E enabled.get() && dynamicConfigService.isEnabled(%22pubsub.amazon%22, true);%0A  %7D
 %0A%7D%0A

Support dynamically enabling pubsub subscriber (#903)