//设置package包名称以及导入依赖的类
package uk.ac.wellcome.finatra.modules

import javax.inject.Singleton

import com.amazonaws.auth.{AWSStaticCredentialsProvider, BasicAWSCredentials}
import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration
import com.amazonaws.services.sns._
import com.google.inject.Provides
import com.twitter.inject.TwitterModule
import uk.ac.wellcome.models.aws.AWSConfig

object SNSClientModule extends TwitterModule {
  val snsEndpoint = flag[String](
    "aws.sns.endpoint",
    "",
    "Endpoint of AWS SNS. The region will be used if the enpoint is not provided")
  @Singleton
  @Provides
  def providesSNSClient(awsConfig: AWSConfig): AmazonSNS = {
    val standardSnsClient = AmazonSNSClientBuilder.standard
    if (snsEndpoint().isEmpty)
      standardSnsClient
        .withRegion(awsConfig.region)
        .build()
    else
      standardSnsClient
        .withCredentials(
          new AWSStaticCredentialsProvider(
            new BasicAWSCredentials(awsConfig.accessKey.get,
                                    awsConfig.secretKey.get)))
        .withEndpointConfiguration(
          new EndpointConfiguration(snsEndpoint(), awsConfig.region))
        .build()
  }
} 

//设置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 akka.persistence.s3

import java.io.InputStream

import com.amazonaws.auth.{ BasicAWSCredentials, DefaultAWSCredentialsProviderChain }
import com.amazonaws.services.s3.{ S3ClientOptions, AmazonS3Client }
import com.amazonaws.services.s3.model._

import scala.concurrent.{ Future, ExecutionContext }

trait S3Client {
  val s3ClientConfig: S3ClientConfig

  lazy val client: AmazonS3Client = {
    val client =
      if (s3ClientConfig.awsUseDefaultCredentialsProviderChain)
        new AmazonS3Client(new DefaultAWSCredentialsProviderChain).withRegion(s3ClientConfig.region)
      else
        new AmazonS3Client(new BasicAWSCredentials(s3ClientConfig.awsKey, s3ClientConfig.awsSecret))

    s3ClientConfig.endpoint.foreach { endpoint =>
      client.withEndpoint(endpoint)
      ()
    }
    client.setS3ClientOptions(new S3ClientOptions()
      .withPathStyleAccess(s3ClientConfig.options.pathStyleAccess)
      .withChunkedEncodingDisabled(s3ClientConfig.options.chunkedEncodingDisabled))
    client
  }

  def createBucket(bucketName: String)(implicit ec: ExecutionContext): Future[Bucket] = Future {
    client.createBucket(bucketName)
  }

  def deleteBucket(bucketName: String)(implicit ec: ExecutionContext): Future[Unit] = Future {
    client.deleteBucket(bucketName)
  }

  def putObject(bucketName: String, key: String, input: InputStream, metadata: ObjectMetadata)(implicit ec: ExecutionContext): Future[PutObjectResult] = Future {
    client.putObject(new PutObjectRequest(bucketName, key, input, metadata))
  }

  def getObject(bucketName: String, key: String)(implicit ec: ExecutionContext): Future[S3Object] = Future {
    client.getObject(new GetObjectRequest(bucketName, key))
  }

  def listObjects(request: ListObjectsRequest)(implicit ec: ExecutionContext): Future[ObjectListing] = Future {
    client.listObjects(request)
  }

  def deleteObject(bucketName: String, key: String)(implicit ec: ExecutionContext): Future[Unit] = Future {
    client.deleteObject(bucketName, key)
  }

  def deleteObjects(request: DeleteObjectsRequest)(implicit ec: ExecutionContext): Future[Unit] = Future {
    client.deleteObjects(request)
  }
} 

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

import cats.data.Kleisli
import cats.~>
import com.amazonaws.auth.BasicAWSCredentials
import wookie.httpclient._
import wookie.result._
import wookie.service._
import wookie.signer._
import wookie.sqs.algebra.{ SQSIO, SQSOp }
import wookie.sqs.implicits._

import scala.concurrent.Future

case class SQS(props: Properties, client: HttpClient) extends Service {

  val endpoint = "https://sqs.amazonaws.com"

  val serviceName = "sqs"

  def credentials = new BasicAWSCredentials(
    props.accessKey,
    props.secretAccessKey
  )

  def run[A](op: SQSIO[A]): Future[A] = {
    val result = op foldMap sqsInterpreter

    result.run(Signer(endpoint, serviceName, credentials))
  }

  val sqsInterpreter = new (SQSOp ~> Result) {
    def apply[A](command: SQSOp[A]): Result[A] =
      Kleisli { signer: Signer ?
        client.exec(signer.sign(command.marshalledReq))(command.responseHandler, errorResponseHandler)
      }
  }

} 

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

import cats.data.Kleisli
import cats.~>
import com.amazonaws.auth.BasicAWSCredentials
import wookie.httpclient._
import wookie.result._
import wookie.s3.algebra._
import wookie.service._
import wookie.signer._

import scala.concurrent.Future

import implicits._

case class S3(props: Properties, client: HttpClient) extends Service {

  val endpoint = "https://s3.amazonaws.com"

  val serviceName = "s3"

  def credentials = new BasicAWSCredentials(
    props.accessKey,
    props.secretAccessKey
  )

  def run[A](op: S3IO[A]): Future[A] = {
    val result = op foldMap s3Interpreter

    result.run(Signer(endpoint, serviceName, credentials))
  }

  val s3Interpreter = new (S3Op ~> Result) {
    def apply[A](command: S3Op[A]): Result[A] =
      Kleisli { signer: Signer ?
        client.exec(signer.sign(command.marshalledReq))(command.responseHandler, errorResponseHandler)
      }
  }

} 

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

import cats.data.Kleisli
import cats.~>
import com.amazonaws.auth.BasicAWSCredentials
import wookie.dynamodb.algebra._
import wookie.httpclient._
import wookie.result._
import wookie.service._
import wookie.signer._

import scala.concurrent.Future
import scala.language._

import implicits._

case class DynamoDB(props: Properties, client: HttpClient) extends Service {

  def endpoint = "https://dynamodb.us-east-1.amazonaws.com"

  def serviceName = "dynamodb"

  def credentials = new BasicAWSCredentials(props.accessKey, props.secretAccessKey)

  def run[A](op: DynamoDBIO[A]): Future[A] = {
    val result = op foldMap dynamoDBInterpreter

    result.run(Signer(endpoint, serviceName, credentials))
  }

  val dynamoDBInterpreter = new (DynamoDBOp ~> Result) {
    def apply[A](command: DynamoDBOp[A]): Result[A] =
      Kleisli { signer: Signer ?
        client.exec(signer.sign(command.marshalledReq))(command.responseHandler, errorResponseHandler)
      }
  }
} 

//设置package包名称以及导入依赖的类
package akka.stream.alpakka.sqs.scaladsl

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import com.amazonaws.auth.{AWSCredentialsProvider, AWSStaticCredentialsProvider, BasicAWSCredentials}
import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration
import com.amazonaws.services.sqs.{AmazonSQSAsync, AmazonSQSAsyncClientBuilder}
import org.elasticmq.rest.sqs.{SQSRestServer, SQSRestServerBuilder}
import org.scalatest.{BeforeAndAfterAll, Suite, Tag}

import scala.concurrent.Await
import scala.concurrent.duration._
import scala.util.Random

trait DefaultTestContext extends BeforeAndAfterAll { this: Suite =>

  lazy val sqsServer: SQSRestServer = SQSRestServerBuilder.withDynamicPort().start()
  lazy val sqsAddress = sqsServer.waitUntilStarted().localAddress
  lazy val sqsPort = sqsAddress.getPort
  lazy val sqsEndpoint: String = {
    s"http://${sqsAddress.getHostName}:$sqsPort"
  }

  object Integration extends Tag("akka.stream.alpakka.sqs.scaladsl.Integration")

  //#init-mat
  implicit val system = ActorSystem()
  implicit val mat = ActorMaterializer()
  //#init-mat

  val credentialsProvider = new AWSStaticCredentialsProvider(new BasicAWSCredentials("x", "x"))

  implicit val sqsClient = createAsyncClient(sqsEndpoint, credentialsProvider)

  def randomQueueUrl(): String = sqsClient.createQueue(s"queue-${Random.nextInt}").getQueueUrl

  override protected def afterAll(): Unit = {
    super.afterAll()
    sqsServer.stopAndWait()
    Await.ready(system.terminate(), 5.seconds)
  }

  def createAsyncClient(sqsEndpoint: String, credentialsProvider: AWSCredentialsProvider): AmazonSQSAsync = {
    //#init-client
    val client: AmazonSQSAsync = AmazonSQSAsyncClientBuilder
      .standard()
      .withCredentials(credentialsProvider)
      .withEndpointConfiguration(new EndpointConfiguration(sqsEndpoint, "eu-central-1"))
      .build()
    //#init-client
    client
  }

} 

//设置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 eu.shiftforward.apso.io

import com.amazonaws.auth.BasicAWSCredentials
import eu.shiftforward.apso.CustomMatchers
import eu.shiftforward.apso.aws.S3Bucket
import org.specs2.mutable.Specification

import scala.util.Try

class FileDescriptorSpec extends Specification with CustomMatchers {

  "A FileDescriptor" should {
    val fdConfig = config.Credentials(
      s3 = config.Credentials.S3(
        ids = Map("test" -> config.Credentials.S3.Entry("a", "b"))),
      sftp = config.Credentials.Sftp(
        default = Some(config.Credentials.Sftp.Entry.Basic("foo", "bar"))))

    "correctly be initialized given a URI with protocol" in {
      FileDescriptor("file:///tmp/folder") mustEqual LocalFileDescriptor("/tmp/folder")
      FileDescriptor("s3://tmp/path") mustEqual S3FileDescriptor("tmp/path")
      FileDescriptor("sftp://localhost/tmp/path", fdConfig) mustEqual SftpFileDescriptor("localhost/tmp/path", fdConfig.sftp)
      FileDescriptor("sftp://valid-host.com/tmp/path", fdConfig) mustEqual SftpFileDescriptor("valid-host.com/tmp/path", fdConfig.sftp)
    }

    "be serializable" in {
      FileDescriptor("file:///tmp/folder") must beSerializable
      FileDescriptor("s3://tmp/path") must beSerializable
      FileDescriptor("sftp://localhost/tmp/path", fdConfig) must beSerializable
    }

    "fail when initializing with an unsupported protocol" in {
      Try(FileDescriptor("wrongprotocol:///tmp")) must beAFailedTry
    }

    "fail when initializing without a protocol" in {
      Try(FileDescriptor("tmp")) must beAFailedTry
    }

    "be initialized with credentials when given a config" in {
      FileDescriptor("s3://test/path/path", fdConfig) match {
        case s3: S3FileDescriptor =>
          s3 must beSerializable
          s3.bucket must beEqualTo(new S3Bucket("test", () => new BasicAWSCredentials("a", "b")))
      }
    }
  }
} 

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

import com.amazonaws.auth.{AWSCredentials, BasicAWSCredentials}
import com.amazonaws.regions.Regions
import play.api.Configuration

class Config(orig: Configuration) {
  lazy val googleMapsKey: Option[String] = orig.getString("google.maps.key")
  lazy val amazon = orig.getConfig("amazon").map(new AmazonConfig(_))
  lazy val mail: Option[String] = orig.getString("management.mail")
}

class AmazonConfig(config: Configuration) {
  val regionRaw: Option[String] = config.getString("region")
  val access: Option[String] = config.getString("access_key")
  val secret: Option[String] = config.getString("secret_key")

  lazy val credentials: Option[AWSCredentials] = for {
    a <- access
    s <- secret
  } yield new BasicAWSCredentials(a, s)

  lazy val region: Option[Regions] = regionRaw.map(Regions.fromName)

  def ses: Option[MyAmazonSES] = for {
    c <- credentials
    r <- region
  } yield new MyAmazonSES(c, r)
} 

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

import javax.inject.Inject

import com.amazonaws.auth.BasicAWSCredentials
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsyncClient
import com.github.dwhjames.awswrap.dynamodb.{AmazonDynamoDBScalaClient, AmazonDynamoDBScalaMapper}
import com.google.inject.ImplementedBy
import play.api.Configuration
import play.api.libs.concurrent.Execution.Implicits._

@ImplementedBy(classOf[AmazonDynamoDbConnection])
trait DynamoDbConnection {
  val db: AmazonDynamoDBScalaMapper
}

class AmazonDynamoDbConnection @Inject() (configuration: Configuration) extends DynamoDbConnection {
  private val credentials = new BasicAWSCredentials(
    configuration.getString("aws.accessKey").get,
    configuration.getString("aws.secretKey").get
  )
  private val sdkClient = new AmazonDynamoDBAsyncClient(credentials)
  private val client = new AmazonDynamoDBScalaClient(sdkClient)

  val db = AmazonDynamoDBScalaMapper(client)
}

class InMemoryDynamoDbConnection extends DynamoDbConnection {
  private val sdkClient = new AmazonDynamoDBAsyncClient(new BasicAWSCredentials("FAKE_ACCESS_KEY", "FAKE_SECRET_KEY"))
  sdkClient.setEndpoint("http://localhost:8000")
  private val client = new AmazonDynamoDBScalaClient(sdkClient)

  val db = AmazonDynamoDBScalaMapper(client)
} 

//设置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 me.davidvuong.http_api.utils

import com.amazonaws.AmazonWebServiceRequest
import com.amazonaws.auth.{AWSStaticCredentialsProvider, BasicAWSCredentials}
import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration
import com.amazonaws.handlers.AsyncHandler
import com.amazonaws.services.sqs.model.{SendMessageRequest, SendMessageResult}
import com.amazonaws.services.sqs.{AmazonSQSAsync, AmazonSQSAsyncClientBuilder}

import scalaz._
import scalaz.concurrent.Task
import me.davidvuong.http_api.config.SqsConfig

case class SqsQueueService(config: SqsConfig) {
  val client: AmazonSQSAsync = AmazonSQSAsyncClientBuilder
    .standard
    .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials(config.accessKey, config.secretKey)))
    .withEndpointConfiguration(new EndpointConfiguration(config.url, config.region))
    .build()

  def send(message: String): Task[SendMessageResult] = {
    Task.async[SendMessageResult] { k =>
      val request = new SendMessageRequest(config.url, message)
      client.sendMessageAsync(request, handler[SendMessageRequest, SendMessageResult](k))
    }
  }

  private def handler[E <: AmazonWebServiceRequest, A](k: (Throwable \/ A) => Unit) = new AsyncHandler[E, A] {
    override def onError(exception: Exception):    Unit = k(-\/(exception))
    override def onSuccess(request: E, result: A): Unit = k(\/-(result))
  }
} 

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

import com.amazonaws.services.sqs.AmazonSQSAsyncClient
import com.amazonaws.services.sqs.model._
import fs2.Stream
import java.util.concurrent.Executors

import cats.effect.IO
import com.amazonaws.auth.BasicAWSCredentials
import com.imageintelligence.fs2sqs.FS2SQS
import fs2._

import scala.concurrent.ExecutionContext

object PublisherExample {

  def loggingSink[F[_], A]: Sink[F, A] = { s =>
    s.map { i =>
      println(i)
    }
  }

  def main(args: Array[String]): Unit = {
    val tp = Executors.newFixedThreadPool(4)
    implicit val ec = ExecutionContext.fromExecutorService(tp)
    val credentials = new BasicAWSCredentials(sys.env("AWS_ACCESS_KEY"), sys.env("AWS_SECRET_KEY"))
    val client = new AmazonSQSAsyncClient(credentials)
    val queueUrl = "https://sqs.ap-southeast-2.amazonaws.com/1234/example"

    // Construct an infinite Stream SendMessageRequest's, with the same body "123"
    val messageRequestsStream: Stream[IO, SendMessageRequest] =
      Stream.constant(new SendMessageRequest(queueUrl, "123")).repeat

    // Construct a Publish pipe that can turn SendMessageRequest's into SendMessageResult's
    val publishPipe: Pipe[IO, SendMessageRequest, SendMessageResult] = FS2SQS.publishPipe(client)

    // Compose our stream and pipe.
    val effect = messageRequestsStream
      .through(publishPipe)
      .to(loggingSink)
      .onError(e => Stream.emit(println("Error: " + e.getMessage)))

    // Lift our effect into a Task, and run it.
    effect.run.unsafeRunSync()
  }
} 

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

import com.amazonaws.services.dynamodbv2.model.DeleteTableRequest
import com.amazonaws.services.dynamodbv2.util.TableUtils
import com.amazonaws.services.dynamodbv2.document.DynamoDB
import org.scalatest._
import scala.concurrent.duration._
import com.amazonaws.auth.{ BasicAWSCredentials, AWSStaticCredentialsProvider }

trait DynamoTestClient {
  val dynamoPortKey = "dynamodb.local.port"

  val dynamoPort = sys.props.get(dynamoPortKey).getOrElse {
    throw new Exception(s"Failed to find $dynamoPortKey")
  }

  val credentials = new AWSStaticCredentialsProvider(new BasicAWSCredentials("", ""))
  lazy val client = Repl.dynamoClient(Some(s"http://127.0.0.1:$dynamoPort"), Some(credentials))
}

trait DynamoSpec
    extends BeforeAndAfterAll
    with BeforeAndAfterEach
    with DynamoTestClient { self: Suite =>

  def tableNames: Seq[String]

  lazy val dynamo = new DynamoDB(client)

  override def afterEach() = {
    super.afterEach()
    tableNames.foreach { tableName =>
      TableUtils.deleteTableIfExists(client, new DeleteTableRequest().withTableName(tableName))
    }
  }

  override def afterAll() = {
    super.afterAll()
    client.shutdown()
  }

} 

//设置package包名称以及导入依赖的类
package uk.co.telegraph.cloud.aws

import com.amazonaws.auth.profile.ProfileCredentialsProvider
import com.amazonaws.auth.{AWSCredentialsProvider, AWSStaticCredentialsProvider, BasicAWSCredentials, EnvironmentVariableCredentialsProvider}
import uk.co.telegraph.cloud.{AuthCredentials, AuthEnvVars, AuthProfile, AuthToken}

package object auth {

  implicit object AuthTokenProvider extends AuthProvider[AuthToken]{
    override def authenticate(authToken: AuthToken): AWSCredentialsProvider = {
      val basicAuth = new BasicAWSCredentials(authToken.accessToken, authToken.secretToken)
      new AWSStaticCredentialsProvider(basicAuth)
    }
  }

  implicit object AuthProfileProvider extends AuthProvider[AuthProfile]{
    override def authenticate(authProfile: AuthProfile): AWSCredentialsProvider = {
      new ProfileCredentialsProvider(authProfile.profileName.orNull)
    }
  }

  implicit object AuthEnvVarsProvider extends AuthProvider[AuthEnvVars]{
    override def authenticate(x: AuthEnvVars): AWSCredentialsProvider = {
      new EnvironmentVariableCredentialsProvider()
    }
  }

  private def doAuthenticate[A <: AuthCredentials:AuthProvider](thing:A) = AuthProvider[A].authenticate(thing)

  implicit class AuthenticationOper[A <: AuthCredentials]( authentication: A){
    def toProvider: AWSCredentialsProvider = {
      //TODO: There must be a way to remove this thing!
      authentication match {
        case auth:AuthProfile => doAuthenticate( auth )
        case auth:AuthToken   => doAuthenticate( auth )
        case auth:AuthEnvVars => doAuthenticate( auth )
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package org.idark.spark.ssh

import java.io.File
import java.util

import com.amazonaws.auth.{BasicAWSCredentials}
import com.amazonaws.services.elasticmapreduce.model.{HadoopJarStepConfig, StepConfig, AddJobFlowStepsRequest}
import com.amazonaws.services.elasticmapreduce.util.StepFactory
import com.amazonaws.services.elasticmapreduce.{AmazonElasticMapReduceClient}

object AwsEmrExecutor {

  def run(uberJarFile: File, className: String, accessKey: String, secretKey: String) = {
    val credentials = new BasicAWSCredentials(accessKey, secretKey)
    val emr = new AmazonElasticMapReduceClient(credentials)

    val stepFactory = new StepFactory()
    val emrClient = new AmazonElasticMapReduceClient(credentials)
    val req = new AddJobFlowStepsRequest()
    req.withJobFlowId("j-1K48XXXXXXHCB")

    val stepConfigs = new util.ArrayList[StepConfig]()

    val sparkStepConf = new HadoopJarStepConfig()
      .withJar("command-runner.jar")
      .withArgs("spark-submit","--executor-memory","1g","--class",className,uberJarFile.getAbsolutePath,"10")

    val sparkStep = new StepConfig()
      .withName("Spark Step")
      .withActionOnFailure("CONTINUE")
      .withHadoopJarStep(sparkStepConf)

    stepConfigs.add(sparkStep)
    req.withSteps(stepConfigs)
    val result = emrClient.addJobFlowSteps(req)
    result
  }

} 

//设置package包名称以及导入依赖的类
package uk.gov.homeoffice.aws.sqs

import java.net.URL
import java.util.UUID
import com.amazonaws.auth.BasicAWSCredentials
import org.elasticmq.rest.sqs.SQSRestServerBuilder
import org.specs2.execute.{AsResult, Result}
import org.specs2.matcher.Scope
import de.flapdoodle.embed.process.runtime.Network._
import uk.gov.homeoffice.specs2.ComposableAround

trait SQSServerEmbedded extends SQSServer with QueueCreation with Scope with ComposableAround {
  val sqsHost = new URL(s"http://127.0.0.1:$getFreeServerPort")

  val sqsServer = SQSRestServerBuilder withInterface sqsHost.getHost withPort sqsHost.getPort start()

  implicit val sqsClient = new SQSClient(new URL(s"$sqsHost/queue"), new BasicAWSCredentials("x", "x"))

  val createMessage: String => Message =
    message => {
      val queue = create(new Queue(UUID.randomUUID().toString))
      val sqs = new SQS(queue)

      sqs publish message
      sqs.receive.head
    }

  override def around[R: AsResult](r: => R): Result = try {
    sqsServer waitUntilStarted()
    info(s"Started SQS $sqsHost")
    super.around(r)
  } finally {
    info(s"Stopping SQS $sqsHost")
    sqsServer stopAndWait()
  }
} 

//设置package包名称以及导入依赖的类
package uk.ac.wellcome.finatra.modules

import javax.inject.Singleton

import com.amazonaws.auth.{AWSStaticCredentialsProvider, BasicAWSCredentials}
import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration
import com.amazonaws.services.sqs._
import com.google.inject.Provides
import com.twitter.inject.TwitterModule
import uk.ac.wellcome.models.aws.AWSConfig

object SQSClientModule extends TwitterModule {
  override val modules = Seq(SQSConfigModule)

  val sqsEndpoint = flag[String](
    "aws.sqs.endpoint",
    "",
    "Endpoint for AWS SQS Service. If not provided, the region will be used instead")

  @Singleton
  @Provides
  def providesSQSClient(awsConfig: AWSConfig): AmazonSQS = {
    val sqsClientBuilder = AmazonSQSClientBuilder.standard
    if (sqsEndpoint().isEmpty)
      sqsClientBuilder
        .withRegion(awsConfig.region)
        .build()
    else
      sqsClientBuilder
        .withEndpointConfiguration(
          new EndpointConfiguration(sqsEndpoint(), awsConfig.region))
        .withCredentials(
          new AWSStaticCredentialsProvider(
            new BasicAWSCredentials(awsConfig.accessKey.get,
                                    awsConfig.secretKey.get)))
        .build()
  }

} 

//设置package包名称以及导入依赖的类
package uk.ac.wellcome.test.utils

import com.amazonaws.auth.{AWSStaticCredentialsProvider, BasicAWSCredentials}
import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration
import com.amazonaws.services.sqs.model.PurgeQueueRequest
import com.amazonaws.services.sqs.{AmazonSQS, AmazonSQSClientBuilder}
import org.scalatest.concurrent.Eventually
import org.scalatest.{BeforeAndAfterEach, Suite}

trait SQSLocal
    extends BeforeAndAfterEach
    with Eventually
    with ExtendedPatience { this: Suite =>

  private val sqsEndpointUrl = s"http://localhost:9324"
  private val accessKey = "access"
  private val secretKey = "secret"

  val sqsLocalFlags = Map(
    "aws.sqs.endpoint" -> sqsEndpointUrl,
    "aws.accessKey" -> accessKey,
    "aws.secretKey" -> secretKey,
    "aws.region" -> "localhost"
  )

  val sqsClient: AmazonSQS = AmazonSQSClientBuilder
    .standard()
    .withCredentials(new AWSStaticCredentialsProvider(
      new BasicAWSCredentials(accessKey, secretKey)))
    .withEndpointConfiguration(
      new EndpointConfiguration(sqsEndpointUrl, "localhost"))
    .build()

  private var queueUrls: List[String] = Nil

  def createQueueAndReturnUrl(queueName: String): String = {
    // Use eventually to allow some time for the local SQS to start up.
    // If it is not started all suites using this will crash at start up time.
    val queueUrl = eventually {
      sqsClient.createQueue(queueName).getQueueUrl
    }
    queueUrls = queueUrl :: queueUrls
    queueUrl
  }

  override def beforeEach(): Unit = {
    super.beforeEach()
    queueUrls.foreach(queueUrl =>
      sqsClient.purgeQueue(new PurgeQueueRequest().withQueueUrl(queueUrl)))
  }
}