//设置package包名称以及导入依赖的类
package templates

import aws.Interpreter.ErrorsOr
import cats.data.Validated.{Invalid, Valid}
import cats.data.{NonEmptyList, Validated, ValidatedNel}
import com.amazonaws.regions.{Region, Regions}
import com.ovoenergy.comms.model.{Channel, CommManifest}

object AssetProcessing {

  private val assetTemplateReferenceRegex = "(?:'|\")(?: *)(assets)(?:/[^(\"')]+)(?: *)(?:'|\")".r

  case class ProcessedFiles(templateFiles: List[UploadedTemplateFile], assetFiles: List[UploadedTemplateFile])

  def processAssets(region: Regions,
                    assetsS3Bucket: String,
                    commManifest: CommManifest,
                    uploadedFiles: List[UploadedTemplateFile]): ErrorsOr[ProcessedFiles] = {
    import cats.syntax.traverse._
    import cats.instances.list._
    val (assetFiles, nonAssetFiles) = uploadedFiles.partition(_.fileType == Asset)
    val processedTemplateFiles: Validated[NonEmptyList[String], List[UploadedTemplateFile]] = nonAssetFiles
      .traverseU(templateFile => {
        replaceAssetReferences(region, assetsS3Bucket, templateFile.channel, commManifest, templateFile.contents)
          .map(contents => templateFile.copy(contents = contents))
      })
    processedTemplateFiles.map(ProcessedFiles(_, assetFiles)).toEither
  }

  private def replaceAssetReferences(region: Regions,
                                     assetsS3Bucket: String,
                                     channel: Channel,
                                     commManifest: CommManifest,
                                     contents: Array[Byte]): ValidatedNel[String, Array[Byte]] = {
    def replaceReferences(s3Endpoint: String, contentsString: String) = {
      val replacementAssetsPath = s"$s3Endpoint/assets"
      assetTemplateReferenceRegex
        .replaceAllIn(contentsString, m => m.group(0).replaceFirst(m.group(1), replacementAssetsPath))
        .getBytes
    }
    determineS3Endpoint(region, assetsS3Bucket, channel, commManifest).map(replaceReferences(_, new String(contents)))
  }

  private def determineS3Endpoint(region: Regions,
                                  assetsS3Bucket: String,
                                  channel: Channel,
                                  commManifest: CommManifest): ValidatedNel[String, String] = {
    if (!Region.getRegion(region).isServiceSupported("s3")) {
      Invalid(NonEmptyList.of("S3 not supported in region selected"))
    } else if (!Region.getRegion(region).hasHttpsEndpoint("s3")) {
      Invalid(NonEmptyList.of("No https support for s3 in region selected"))
    } else {
      val s3ServiceEndpoint = Region.getRegion(region).getServiceEndpoint("s3")
      Valid(
        s"https://$s3ServiceEndpoint/$assetsS3Bucket/${commManifest.commType.toString.toLowerCase}/${commManifest.name}/${commManifest.version}/${channel.toString.toLowerCase}")
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.hivehome.kafka.connect.sqs

import javax.jms.{JMSException, MessageConsumer, Session}

import com.amazon.sqs.javamessaging.SQSConnectionFactory
import com.amazonaws.auth.{AWSCredentialsProviderChain, BasicAWSCredentials, DefaultAWSCredentialsProviderChain}
import com.amazonaws.internal.StaticCredentialsProvider
import com.amazonaws.regions.{Region, Regions}

object SQSConsumer {
  def apply(conf: Conf): MessageConsumer = {
    val chain = buildCredentialsProviderChain(conf)
    createSQSConsumer(conf, chain)
  }

  @throws(classOf[JMSException])
  private def createSQSConsumer(conf: Conf, chain: AWSCredentialsProviderChain): MessageConsumer = {
    val region = Regions.fromName(conf.awsRegion)
    val connectionFactory = SQSConnectionFactory.builder
      .withRegion(Region.getRegion(region))
      .withAWSCredentialsProvider(chain)
      .build

    val connection = connectionFactory.createConnection
    val session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE)
    val queue = session.createQueue(conf.queueName.get)
    val consumer = session.createConsumer(queue)
    connection.start()
    consumer
  }

  private def buildCredentialsProviderChain(conf: Conf): AWSCredentialsProviderChain = {
    (conf.awsKey, conf.awsSecret) match {
      case (Some(key), Some(secret)) =>
        val credentials = new BasicAWSCredentials(key, secret)
        new AWSCredentialsProviderChain(new StaticCredentialsProvider(credentials), new DefaultAWSCredentialsProviderChain)
      case _ => new DefaultAWSCredentialsProviderChain
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.hivehome.kafka.connect.sqs

import com.amazonaws.regions.{Region, Regions}
import com.amazonaws.services.sqs.AmazonSQSClient
import com.amazonaws.services.sqs.model.{CreateQueueRequest, SendMessageRequest, SendMessageResult}
import org.scalacheck.Gen

import scala.collection.JavaConverters._

trait SQSSupport {
  val queueName = Gen.alphaStr
    .map(a => s"test-connect-${a.take(10)}")
    .sample.get
  var queueUrl: String = null

  val sqs = new AmazonSQSClient()
  sqs.setRegion(Region.getRegion(Regions.EU_WEST_1))

  def createQueue(): Unit = {
    val request = new CreateQueueRequest(queueName)
      .withAttributes(Map("VisibilityTimeout" -> "2").asJava)
    val result = sqs.createQueue(request)
    queueUrl = result.getQueueUrl
    println("Url for created Queue = " + queueUrl)
  }

  def deleteQueue(): Unit = {
    sqs.deleteQueue(queueUrl)
  }

  def sendMessage(msgText: String): SendMessageResult = {
    sqs.sendMessage(new SendMessageRequest()
      .withQueueUrl(queueUrl)
      .withMessageBody(msgText))
  }
} 

//设置package包名称以及导入依赖的类
package akka.persistence.s3

import com.amazonaws.regions.{ Regions, Region }
import com.typesafe.config.Config

private object AWSRegionNames {
  val GovCloud = Regions.GovCloud.getName
  val US_EAST_1 = Regions.US_EAST_1.getName
  val US_WEST_1 = Regions.US_WEST_1.getName
  val US_WEST_2 = Regions.US_WEST_2.getName
  val EU_WEST_1 = Regions.EU_WEST_1.getName
  val EU_CENTRAL_1 = Regions.EU_CENTRAL_1.getName
  val AP_SOUTHEAST_1 = Regions.AP_SOUTHEAST_1.getName
  val AP_SOUTHEAST_2 = Regions.AP_SOUTHEAST_2.getName
  val AP_NORTHEAST_1 = Regions.AP_NORTHEAST_1.getName
  val AP_NORTHEAST_2 = Regions.AP_NORTHEAST_2.getName
  val SA_EAST_1 = Regions.SA_EAST_1.getName
  val CN_NORTH_1 = Regions.CN_NORTH_1.getName
}

class S3ClientConfig(config: Config) {
  import AWSRegionNames._
  val awsKey = config getString "aws-access-key-id"
  val awsSecret = config getString "aws-secret-access-key"
  val awsUseDefaultCredentialsProviderChain = config getBoolean "aws-use-default-credentials-provider-chain"
  val region: Region = config getString "region" match {
    case GovCloud       => Region.getRegion(Regions.GovCloud)
    case US_EAST_1      => Region.getRegion(Regions.US_EAST_1)
    case US_WEST_1      => Region.getRegion(Regions.US_WEST_1)
    case US_WEST_2      => Region.getRegion(Regions.US_WEST_2)
    case EU_WEST_1      => Region.getRegion(Regions.EU_WEST_1)
    case EU_CENTRAL_1   => Region.getRegion(Regions.EU_CENTRAL_1)
    case AP_SOUTHEAST_1 => Region.getRegion(Regions.AP_SOUTHEAST_1)
    case AP_SOUTHEAST_2 => Region.getRegion(Regions.AP_SOUTHEAST_2)
    case AP_NORTHEAST_1 => Region.getRegion(Regions.AP_NORTHEAST_1)
    case AP_NORTHEAST_2 => Region.getRegion(Regions.AP_NORTHEAST_2)
    case SA_EAST_1      => Region.getRegion(Regions.SA_EAST_1)
    case CN_NORTH_1     => Region.getRegion(Regions.CN_NORTH_1)
  }
  val endpoint: Option[String] = {
    val e = config getString "endpoint"
    if (e == "default") None else Some(e)
  }
  object options {
    val pathStyleAccess = config getBoolean "options.path-style-access"
    val chunkedEncodingDisabled = config getBoolean "options.chunked-encoding-disabled"
  }
} 

//设置package包名称以及导入依赖的类
package io.tailrec.sbt.awsfun

import com.amazonaws.regions.Region
import com.amazonaws.services.identitymanagement.AmazonIdentityManagementClient
import com.amazonaws.services.identitymanagement.model.{CreateRoleRequest, CreateRoleResult, Role}

import scala.util.Try

class AwsIamService(region: Region) extends AwsService {

  val defaultLambdaRole = "lambda_basic_execution"

  private val client = new AmazonIdentityManagementClient(credentialsProvider)
  client.setRegion(region)

  def basicLambdaRole(): Option[Role] = {
    import scala.collection.JavaConverters._
    val existingRoles = client.listRoles().getRoles.asScala
    existingRoles.find(_.getRoleName == defaultLambdaRole)
  }

  def createBasicLambdaRole(): Try[Role] = Try {
    println(s"Creating a new IAM role:  ${defaultLambdaRole}")
    val policyDocument = """{"Version":"2012-10-17","Statement":[{"Sid":"","Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}]}"""
    val request = new CreateRoleRequest
    request.setRoleName(defaultLambdaRole)
    request.setAssumeRolePolicyDocument(policyDocument)
    client.createRole(request).getRole
  }

} 

//设置package包名称以及导入依赖的类
package com.example

import java.nio.ByteBuffer

import akka.actor.ActorSystem
import akka.event.Logging
import akka.http.scaladsl.Http
import akka.stream.ActorMaterializer
import akka.http.scaladsl.server._
import com.amazonaws.auth.{AWSCredentialsProvider, BasicAWSCredentials}
import com.amazonaws.internal.StaticCredentialsProvider
import com.amazonaws.regions.{Region, Regions}
import com.amazonaws.services.kinesis.{AmazonKinesis, AmazonKinesisClient}
import com.amazonaws.services.kinesis.model.{PutRecordRequest, PutRecordResult}
import com.typesafe.config.ConfigFactory

object AkkaHttp extends App with Kinesis {
  implicit val system = ActorSystem()
  implicit val executor = system.dispatcher
  implicit val materializer = ActorMaterializer()

  val config = ConfigFactory.load()
  val logger = Logging(system, getClass)

  import Directives._

  val route: Route =
    path("order" / IntNumber) { id =>
      complete(id.toString)
    } ~
    parameters('key, 'value) { (key, value) =>
      if(accessKeyId.isDefined && secretAccessKey.isDefined) {
        put(key, value)
      }
      complete(s"key: ${key}, value: ${value}")
    }

  val bindingFuture = Http().bindAndHandle(route, config.getString("http.interface"), config.getInt("http.port"))

  override def put(key: String, value: String): PutRecordResult = {
    val credentialsProvider: AWSCredentialsProvider = new StaticCredentialsProvider(new BasicAWSCredentials(accessKeyId.get, secretAccessKey.get))
    val kinesis: AmazonKinesis = new AmazonKinesisClient(credentialsProvider)
    kinesis.setRegion(Region.getRegion(Regions.AP_NORTHEAST_1))
    val request: PutRecordRequest = new PutRecordRequest()
    request.setStreamName(streamName)
    request.setData(ByteBuffer.wrap(value.getBytes("UTF-8")))
    request.setPartitionKey(key)
    val putRecord: PutRecordResult = kinesis.putRecord(request)
    println(s"key:${key} ,record:${value}, ${putRecord}")
    println("--------")
    kinesis.putRecord(request)
  }
} 

//设置package包名称以及导入依赖的类
package services.aws

import com.amazonaws.regions.Region
import com.amazonaws.regions.Regions
import com.amazonaws.regions.{Region, Regions}
import com.amazonaws.services.cloudwatch.AmazonCloudWatchAsyncClient
import com.amazonaws.services.cloudwatch.AmazonCloudWatchAsyncClient
import com.amazonaws.services.ec2.AmazonEC2Client
import com.amazonaws.services.ec2.AmazonEC2Client
import com.amazonaws.services.ec2.model.DescribeTagsRequest
import com.amazonaws.services.ec2.model.DescribeTagsRequest
import com.amazonaws.services.ec2.model.Filter
import com.amazonaws.services.ec2.model.Filter
import com.amazonaws.util.EC2MetadataUtils
import com.amazonaws.util.EC2MetadataUtils
import scala.collection.JavaConverters._


object AWS {

  lazy val region = Region getRegion Regions.EU_WEST_1
  lazy val EC2Client = region.createClient(classOf[AmazonEC2Client], null, null)
  lazy val CloudWatch = region.createClient(classOf[AmazonCloudWatchAsyncClient], null, null)

}

trait AwsInstanceTags {
  lazy val instanceId = Option(EC2MetadataUtils.getInstanceId)

  def readTag(tagName: String) = {
    instanceId.flatMap { id =>
      val tagsResult = AWS.EC2Client.describeTags(
        new DescribeTagsRequest().withFilters(
          new Filter("resource-type").withValues("instance"),
          new Filter("resource-id").withValues(id),
          new Filter("key").withValues(tagName)
        )
      )
      tagsResult.getTags.asScala.find(_.getKey == tagName).map(_.getValue)
    }
  }
} 

//设置package包名称以及导入依赖的类
package sbtecr

import java.util.Base64

import com.amazonaws.regions.Region
import com.amazonaws.services.ecr.AmazonECRClient
import com.amazonaws.services.ecr.model._
import sbt.Logger

import scala.collection.JavaConverters._

private[sbtecr] object Ecr extends Aws {

  def domain(region: Region, accountId: String) = s"${accountId}.dkr.ecr.${region}.amazonaws.com"

  def createRepository(region: Region, repositoryName: String)(implicit logger: Logger): Unit = {
    val request = new CreateRepositoryRequest()
    request.setRepositoryName(repositoryName)

    try {
      val result = ecr(region).createRepository(request)
      logger.info(s"Repository created in ${region}: arn=${result.getRepository.getRepositoryArn}")
    } catch {
      case e: RepositoryAlreadyExistsException =>
        logger.info(s"Repository exists: ${region}/${repositoryName}")
    }
  }

  def dockerCredentials(region: Region)(implicit logger: Logger): (String, String) = {
    val request = new GetAuthorizationTokenRequest()
    val response = ecr(region).getAuthorizationToken(request)

    response
      .getAuthorizationData
      .asScala
      .map(_.getAuthorizationToken)
      .map(Base64.getDecoder.decode(_))
      .map(new String(_, "UTF-8"))
      .map(_.split(":"))
      .headOption match {
        case Some(creds) if creds.size == 2 =>
          (creds(0), creds(1))
        case _ =>
          throw new IllegalStateException("Authorization token not found.")
      }
  }

  private def ecr(region: Region) = client(classOf[AmazonECRClient], region)
} 

//设置package包名称以及导入依赖的类
package sbtecr

import com.amazonaws.AmazonWebServiceClient
import com.amazonaws.auth._
import com.amazonaws.auth.profile.ProfileCredentialsProvider
import com.amazonaws.regions.Region

private[sbtecr] trait Aws {

  def client[T <: AmazonWebServiceClient](clientClass: Class[T], region: Region): T = {
    region.createClient(clientClass, credentialsProvider(), null)
  }

  private def credentialsProvider(): AWSCredentialsProvider =
    new AWSCredentialsProviderChain(
      new EnvironmentVariableCredentialsProvider(),
      new SystemPropertiesCredentialsProvider(),
      new ProfileCredentialsProvider(sys.env.getOrElse("AWS_DEFAULT_PROFILE", "default")),
      new InstanceProfileCredentialsProvider()
    )
} 

//设置package包名称以及导入依赖的类
package vitomatic

import com.amazonaws.auth.BasicAWSCredentials
import com.amazonaws.regions.{Region, Regions}
import com.amazonaws.services.lambda.runtime.Context
import com.amazonaws.services.sqs.AmazonSQSClient
import com.amazonaws.services.sqs.model.{CreateQueueRequest, GetQueueUrlRequest}
import com.fasterxml.jackson.databind.ObjectMapper
import com.typesafe.config.ConfigFactory
import vitomatic.model.SingleHitResult
import scala.util.Try


class LoadHitResultQueue {

  val objectMapper = new ObjectMapper()

  val config = ConfigFactory.load()

  lazy val awsCredentials = new BasicAWSCredentials(
    config.getString("aws.accessKeyId"),
    config.getString("aws.secretAccessKey"))

  val queueName = config.getString("aws.sqs.queueName")

  val sqs = new AmazonSQSClient(awsCredentials)
  sqs.setRegion(Region.getRegion(Regions.valueOf(config.getString("aws.region"))))

  def enqueue(data: SingleHitResult, ctx: Context): String = {
    val clientQueueName = s"${queueName}_${data.getClientId}"
    Try(sqs.createQueue(new CreateQueueRequest(clientQueueName)))
    val queueUrl = sqs.getQueueUrl(new GetQueueUrlRequest(clientQueueName)).getQueueUrl
    val json = objectMapper.writeValueAsString(data)
    val messageId = sqs.sendMessage(queueUrl, json).getMessageId
    messageId
  }

  def echo(data: String): String = {
    s"Hello ${data}"
  }

} 

//设置package包名称以及导入依赖的类
package tanukki.akka.cluster.autodown.aws.elasticloadbalancing

import akka.actor.ActorSystem
import com.amazonaws.regions.{Regions, Region}

case class ElasticLoadBalancingAwareDowningSettings(region: Region, loadBalancerName: String) {
  require(loadBalancerName.nonEmpty, "loadBalancerName should not be empty")
}

object ElasticLoadBalancingAwareDowningSettings {
  def apply(system: ActorSystem): ElasticLoadBalancingAwareDowningSettings = {
    val c = system.settings.config
    ElasticLoadBalancingAwareDowningSettings(
      Region.getRegion(Regions.fromName(c.getString("custom-downing.aws.elasticloadbalancing.leader-auto-downing-roles-if-unhealthy.region"))),
      c.getString("custom-downing.aws.elasticloadbalancing.leader-auto-downing-roles-if-unhealthy.loadbalancer-name")
    )
  }
} 

//设置package包名称以及导入依赖的类
package com.gu.cm

import com.amazonaws.regions.{Regions, Region}
import com.amazonaws.services.ec2.AmazonEC2Client
import com.amazonaws.services.ec2.model.{Filter, DescribeTagsRequest}
import com.amazonaws.util.EC2MetadataUtils
import scala.collection.JavaConverters._

import scala.util.{Failure, Success, Try}

trait AwsInstance {
  def tags: Map[String, String]
  def region: Option[Region]
}

class AwsInstanceImpl(logger: Logger) extends AwsInstance {
  private def safeAwsOperation[A](operation: => A, errorMessage: => String): Option[A] = Try(operation) match {
    case Success(value) => Some(value)
    case Failure(e) =>
      logger.error(errorMessage, e)
      None
  }

  lazy val region: Option[Region] = safeAwsOperation(Regions.getCurrentRegion, "Impossible to identify the regionName of the instance")
  lazy val instanceId: Option[String] = safeAwsOperation(EC2MetadataUtils.getInstanceId, "Impossible to identify the instanceId")
  lazy val ec2Client: Option[AmazonEC2Client] = region.flatMap { r =>
    safeAwsOperation(r.createClient(classOf[AmazonEC2Client], null, null), "Impossible to create the amazon ec2 client")
  }

  lazy val tags: Map[String, String] = {
    val allTags = for {
      theInstanceId <- instanceId
      theClient <- ec2Client
      tagsResult <- safeAwsOperation(theClient.describeTags(new DescribeTagsRequest().withFilters(
        new Filter("resource-type").withValues("instance"),
        new Filter("resource-id").withValues(theInstanceId)
      )), s"Impossible to describe the tags of the instance $theInstanceId")
    } yield {
      tagsResult.getTags.asScala.map{td => td.getKey -> td.getValue }.toMap
    }

    allTags.getOrElse(Map.empty)
  }
}

object AwsInstance {
  def apply(logger: Logger): AwsInstance = new AwsInstanceImpl(logger)
} 

//设置package包名称以及导入依赖的类
package com.gu.cm

import com.amazonaws.regions.{RegionUtils, Region}
import org.specs2.mutable.Specification
import org.specs2.specification.Scope

class IdentitySpec extends Specification {
  "InstanceDescriber" should {
    "describe a test instance" in new InstanceDescriberScope {
      override val mode = Mode.Test
      instanceDescriber.whoAmI shouldEqual LocalApplication("Test App")
    }
    "describe a dev instance" in new InstanceDescriberScope {
      override val mode = Mode.Dev
      instanceDescriber.whoAmI shouldEqual LocalApplication("Test App")
    }
    "fail to describe a prod instance without tags" in new InstanceDescriberScope {
      instanceDescriber.whoAmI should throwA[RuntimeException]
    }
    "fail to describe a prod instance without the right tags" in new InstanceDescriberScope {
      override val instanceTags = Map("bleurg" -> "ouch")
      instanceDescriber.whoAmI should throwA[RuntimeException]
    }
    "describe a prod instance with the right tags" in new InstanceDescriberScope {
      override val instanceTags = Map("App" -> "myApp", "Stage" -> "TEST", "Stack" -> "myStack")
      instanceDescriber.whoAmI shouldEqual AwsApplication("myStack", "myApp", "TEST", "eu-west-1")
    }
  }

  trait InstanceDescriberScope extends Scope {
    def mode: Mode = Mode.Prod

    def instanceTags = Map.empty[String, String]

    def awsInstance = new AwsInstance {
      override def region: Option[Region] = Some(RegionUtils.getRegion("eu-west-1"))
      override def tags: Map[String, String] = instanceTags
    }

    def instanceDescriber = new InstanceDescriber(
      defaultAppName = "Test App",
      mode = mode,
      awsInstance = awsInstance
    )
  }
} 

//设置package包名称以及导入依赖的类
import java.nio.ByteBuffer
import java.util.Calendar

import com.amazonaws.auth.{AWSCredentialsProvider, BasicAWSCredentials}
import com.amazonaws.internal.StaticCredentialsProvider
import com.amazonaws.regions.{Region, Regions}
import com.amazonaws.services.kinesis.model.{PutRecordRequest, PutRecordResult}
import com.amazonaws.services.kinesis.{AmazonKinesis, AmazonKinesisClient}
import org.apache.commons.lang.RandomStringUtils

object PutMain {
  val accessKeyId = System.getProperty("accessKeyId")
  val secretAccessKey = System.getProperty("secretAccessKey")

  val appName = "kinesis-test-app"
  val streamName = "kinesis-test-stream"

  val initialPosition = "LATEST"
  val region = "ap-northeast-1"

  def main(args: Array[String]): Unit = {
    val credentialsProvider: AWSCredentialsProvider = new StaticCredentialsProvider(new BasicAWSCredentials(accessKeyId, secretAccessKey))

    val kinesis: AmazonKinesis = new AmazonKinesisClient(credentialsProvider)
    kinesis.setRegion(Region.getRegion(Regions.AP_NORTHEAST_1))

    while (true) {
      val key = RandomStringUtils.randomAlphanumeric(10)
      val data = "KEY_" + Calendar.getInstance().getTime().getTime() + ":" + key

      val request: PutRecordRequest = new PutRecordRequest()
      request.setStreamName(streamName)
      request.setData(ByteBuffer.wrap(data.getBytes("UTF-8")))
      request.setPartitionKey(key)
      val putRecord: PutRecordResult = kinesis.putRecord(request)

      println("key:{} ,record:{}", key, data, putRecord)
      println("--------")
      Thread.sleep(5000)

    }
  }
} 

//设置package包名称以及导入依赖的类
package uk.gov.homeoffice.mercury.boot.configuration

import java.net.URL

import com.amazonaws.ClientConfiguration
import com.amazonaws.auth.BasicAWSCredentials
import com.amazonaws.regions.{Region, Regions}
import com.amazonaws.services.s3.S3ClientOptions
import com.amazonaws.services.s3.model.{CryptoConfiguration, KMSEncryptionMaterialsProvider}
import uk.gov.homeoffice.aws.s3.S3EncryptionClient
import uk.gov.homeoffice.configuration.HasConfig

import scala.util.Try

object S3 extends HasConfig {
  def apply(clientConfiguration: ClientConfiguration = new ClientConfiguration) = {
    val s3Host = new URL(config.getString("aws.s3.uri"))

    val accessKey = config.getString("aws.s3.credentials.access-key")
    val secretKey = config.getString("aws.s3.credentials.secret-key")

    val regions = Regions.fromName(config.getString("aws.s3.region"))

    implicit val s3Client = new S3EncryptionClient(s3Host, new BasicAWSCredentials(accessKey, secretKey),
      new KMSEncryptionMaterialsProvider(config.getString("aws.s3.kms-key")),
      new CryptoConfiguration().withKmsRegion(regions))(clientConfiguration)
    s3Client.setS3ClientOptions(S3ClientOptions.builder().setPathStyleAccess(true).build())

    s3Client.setRegion(Region.getRegion(regions))

    val mercuryBucket = config.getString("aws.s3.buckets.mercury")

    Try {
      // We should be able to ignore any error message if this fails as the bucket should have already been created.
      s3Client.createBucket(mercuryBucket)
    }

    new uk.gov.homeoffice.aws.s3.S3(mercuryBucket)
  }
} 

//设置package包名称以及导入依赖的类
package co.verdigris.spark.connector.ssl

import co.verdigris.ssl.S3JKSSSLOptions
import com.amazonaws.regions.Region
import com.datastax.spark.connector.cql.CassandraConnectorConf.CassandraSSLConf

class AwsS3SSLOptions(
      sslConf: CassandraSSLConf,
      awsRegion: Option[String] = None)
    extends S3JKSSSLOptions(
      sslConf.trustStorePath,
      sslConf.trustStorePassword,
      Some(sslConf.trustStoreType),
      sslConf.enabledAlgorithms,
      awsRegion)


    def build(): AwsS3SSLOptions =
      new AwsS3SSLOptions(
        this.cassandraSSLConf.getOrElse(CassandraSSLConf()),
        this.awsRegion)
  }

  def builder(): Builder = new Builder
} 

//设置package包名称以及导入依赖的类
package services

import javax.inject._

import com.amazonaws.auth.{AWSCredentialsProviderChain, InstanceProfileCredentialsProvider, STSAssumeRoleSessionCredentialsProvider}
import com.amazonaws.auth.profile.ProfileCredentialsProvider
import com.amazonaws.regions.{Region, Regions}
import com.gu.contentapi.firehose.ContentApiFirehoseConsumer
import com.gu.contentapi.firehose.kinesis.KinesisStreamReaderConfig

trait FirehoseService { }

@Singleton
class CapiFirehoseService @Inject()(playconfig: play.Configuration, consumerLogic: ConsumerLogicService) extends FirehoseService {

  val appName = playconfig.getString("appName")
  val roleArn = playconfig.getString("roleArn")
  val streamName = playconfig.getString("streamName")
  val regionName = playconfig.getString("regionName")
  val profileName = playconfig.getString("profileName")

  val dynamoChain = new AWSCredentialsProviderChain(
    new ProfileCredentialsProvider(profileName),
    new InstanceProfileCredentialsProvider()
  )

  val kinesisChain = new AWSCredentialsProviderChain(
    new ProfileCredentialsProvider(profileName),
    new STSAssumeRoleSessionCredentialsProvider.Builder(
      roleArn,
      profileName
    ).build()
  )

  val region = Region.getRegion(
    Regions.fromName(regionName)
  )

  val kinesisStreamReaderConfig = KinesisStreamReaderConfig(
    streamName = streamName,
    app = appName,
    stage = "PROD",
    mode = "live",
    suffix = None,
    kinesisCredentialsProvider = kinesisChain,
    dynamoCredentialsProvider = dynamoChain,
    awsRegion = regionName
  )

  val client: ContentApiFirehoseConsumer = new ContentApiFirehoseConsumer(
    kinesisStreamReaderConfig,
    consumerLogic
  )

  client.start()

}