//设置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 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 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 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 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)))
  }
} 

//设置package包名称以及导入依赖的类
package io.flow.event.v2

import com.amazonaws.ClientConfiguration
import com.amazonaws.auth.{AWSCredentials, AWSCredentialsProvider, AWSStaticCredentialsProvider}
import com.amazonaws.services.kinesis.{AmazonKinesis, AmazonKinesisClientBuilder}
import io.flow.event.Naming

case class StreamConfig(
  awsCredentials: AWSCredentials,
  appName: String,
  streamName: String,
  maxRecords: Int = 1000,   // number of records in each fetch
  idleTimeBetweenReadsInMillis: Int = 1000
) {

  val awSCredentialsProvider: AWSCredentialsProvider = new AWSStaticCredentialsProvider(awsCredentials)

  def kinesisClient: AmazonKinesis = {
    AmazonKinesisClientBuilder.standard().
      withCredentials(awSCredentialsProvider).
      withClientConfiguration(
        new ClientConfiguration()
          .withMaxErrorRetry(5)
          .withThrottledRetries(true)
          .withConnectionTTL(60000)
      ).
      build()
  }

  def dynamoTableName: String = {
    Naming.dynamoKinesisTableName(
      streamName = streamName,
      appName = appName
    )
  }

} 

//设置package包名称以及导入依赖的类
package org.alcaudon.runtime

import java.net.URL
import java.util.Date

import com.amazonaws.HttpMethod
import com.amazonaws.auth.AWSStaticCredentialsProvider
import com.amazonaws.services.s3.AmazonS3ClientBuilder
import com.amazonaws.services.s3.model.GeneratePresignedUrlRequest
import org.alcaudon.runtime.BlobLocation.AWSInformation

import scala.concurrent.duration._

object ObjectStorageUtils {

  def sign(bucketName: String, objectKey: String)(
      implicit awsInfo: AWSInformation): URL = {

    val s3client = AmazonS3ClientBuilder
      .standard()
      .withCredentials(new AWSStaticCredentialsProvider(awsInfo.credentials))
      .withRegion(awsInfo.region)
      .build()

    val expiration = new Date
    var milliSeconds = expiration.getTime
    milliSeconds += 1.hour.toMillis

    expiration.setTime(milliSeconds)

    val generatePresignedUrlRequest =
      new GeneratePresignedUrlRequest(bucketName, objectKey)
    generatePresignedUrlRequest.setMethod(HttpMethod.PUT)
    generatePresignedUrlRequest.setExpiration(expiration)

    s3client.generatePresignedUrl(generatePresignedUrlRequest)
  }
}