//设置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 com.ovoenergy.comms.templates

import cats.Id
import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.services.s3.AmazonS3Client
import com.ovoenergy.comms.model.CommManifest
import com.ovoenergy.comms.templates.cache.CachingStrategy
import com.ovoenergy.comms.templates.model.HandlebarsTemplate
import com.ovoenergy.comms.templates.model.template.processed.CommTemplate
import com.ovoenergy.comms.templates.parsing.Parsing
import com.ovoenergy.comms.templates.parsing.handlebars.HandlebarsParsing
import com.ovoenergy.comms.templates.retriever.{PartialsS3Retriever, TemplatesRetriever, TemplatesS3Retriever}
import com.ovoenergy.comms.templates.s3.AmazonS3ClientWrapper

object TemplatesContext {

  
  def customCachingContext(
      credentialsProvider: AWSCredentialsProvider,
      cachingStrategy: CachingStrategy[CommManifest, ErrorsOr[CommTemplate[Id]]]): TemplatesContext = {
    val s3Client = new AmazonS3ClientWrapper(new AmazonS3Client(credentialsProvider))
    TemplatesContext(
      templatesRetriever = new TemplatesS3Retriever(s3Client),
      parser = new HandlebarsParsing(new PartialsS3Retriever(s3Client)),
      cachingStrategy = cachingStrategy
    )
  }
}

case class TemplatesContext(
    templatesRetriever: TemplatesRetriever,
    parser: Parsing[HandlebarsTemplate],
    cachingStrategy: CachingStrategy[CommManifest, ErrorsOr[CommTemplate[Id]]]
) 

//设置package包名称以及导入依赖的类
package me.snov.akka.sqs.client

import com.amazonaws.ClientConfiguration
import com.amazonaws.auth.AWSCredentialsProvider
import com.typesafe.config.ConfigFactory
import org.scalatest.{FlatSpec, Matchers}
import org.scalatest.mockito.MockitoSugar._

class SqsClientSettingsSpec extends FlatSpec with Matchers {

  it should "parse configuration" in {
    val conf = ConfigFactory.parseString(
      """
        reactive-sqs {
          endpoint = "http://localhost:9324/"
          queue-url = "http://localhost:9324/queue/queue1"
          max-number-of-messages = 10
          visibility-timeout = 60
          wait-time-seconds = 5
        }
      """)
      .getConfig("reactive-sqs")

    val settings = SqsSettings(
      conf,
      Some(mock[AWSCredentialsProvider]),
      Some(mock[ClientConfiguration])
    )

    settings.endpoint shouldBe Some("http://localhost:9324/")
    settings.queueUrl shouldBe "http://localhost:9324/queue/queue1"
    settings.maxNumberOfMessages shouldBe 10
    settings.visibilityTimeout shouldBe Some(60)
    settings.waitTimeSeconds shouldBe 5
  }

  it should "support optional parameters" in {
    val conf = ConfigFactory.parseString(
      """
        reactive-sqs {
          queue-url = "http://localhost:9324/queue/queue1"
          wait-time-seconds = 5
        }
      """)
      .getConfig("reactive-sqs")

    val settings = SqsSettings(
      conf,
      Some(mock[AWSCredentialsProvider]),
      Some(mock[ClientConfiguration])
    )

    settings.endpoint shouldBe None
    settings.queueUrl shouldBe "http://localhost:9324/queue/queue1"
    settings.maxNumberOfMessages shouldBe 10
    settings.visibilityTimeout shouldBe None
    settings.waitTimeSeconds shouldBe 5
  }
} 

//设置package包名称以及导入依赖的类
package actors.workflow.tasks

import actors.DeploymentSupervisor.{StackAndAppVersion, AppVersion, Version}
import actors.workflow.AWSRestartableActor
import akka.actor.Props
import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.services.cloudformation.model.{Capability, CreateStackRequest, Parameter, Tag}
import play.api.libs.json.JsValue
import utils.{AmazonCloudFormationService, PropFactory}

class StackCreator(credentials: AWSCredentialsProvider) extends AWSRestartableActor with AmazonCloudFormationService {

  import actors.workflow.tasks.StackCreator._

  override def receive: Receive = {
    case launchCommand: StackCreateCommand =>
      val appVersionTag = new Tag()
                          .withKey("ApplicationVersion")
                          .withValue(launchCommand.version.appVersion)

      val tags = launchCommand.version match {
        case x:AppVersion => Seq(appVersionTag)
        case x:StackAndAppVersion => Seq(appVersionTag, new Tag().withKey("StackVersion").withValue(x.stackVersion))
      }

      val params = Seq(
        new Parameter()
        .withParameterKey("ImageId")
        .withParameterValue(launchCommand.imageId),
        new Parameter()
        .withParameterKey("ApplicationVersion")
        .withParameterValue(launchCommand.version.appVersion)
      )

      val createStackRequest = new CreateStackRequest()
                               .withTemplateBody(launchCommand.stackData.toString())
                               .withStackName(launchCommand.stackName)
                               .withTags(tags.toArray: _*)
                               .withParameters(params.toArray: _*)
                               .withCapabilities(Capability.CAPABILITY_IAM)

      val awsClient = cloudFormationClient(credentials)
      awsClient.createStack(createStackRequest)

      context.parent ! StackCreateRequestCompleted
  }
}

object StackCreator extends PropFactory {
  case class StackCreateCommand(stackName: String, imageId: String, version: Version, stackData: JsValue)
  case object StackCreateRequestCompleted

  override def props(args: Any*): Props = Props(classOf[StackCreator], args: _*)
} 

//设置package包名称以及导入依赖的类
package actors.workflow.tasks

import actors.workflow.AWSRestartableActor
import akka.actor.Props
import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.services.autoscaling.model.SuspendProcessesRequest
import utils.{AmazonAutoScalingService, PropFactory}

import scala.collection.JavaConverters._

class FreezeASG(credentials: AWSCredentialsProvider) extends AWSRestartableActor with AmazonAutoScalingService {

  import actors.workflow.tasks.FreezeASG._

  override def receive: Receive = {
    case query: FreezeASGCommand =>

      val suspendProcessesRequest = new SuspendProcessesRequest()
                                    .withAutoScalingGroupName(query.asgName)
                                    .withScalingProcesses(Seq("AlarmNotification", "ScheduledActions").asJava)

      val awsClient = autoScalingClient(credentials)
      awsClient.suspendProcesses(suspendProcessesRequest)
      context.parent ! FreezeASGCompleted(query.asgName)
  }
}

object FreezeASG extends PropFactory {
  case class FreezeASGCommand(asgName: String)
  case class FreezeASGCompleted(asgName: String)

  override def props(args: Any*): Props = Props(classOf[FreezeASG], args: _*)
} 

//设置package包名称以及导入依赖的类
package actors.workflow.tasks

import actors.workflow.AWSRestartableActor
import akka.actor.Props
import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.services.autoscaling.model.DescribeAutoScalingGroupsRequest
import utils.{AmazonAutoScalingService, PropFactory}

import scala.collection.JavaConverters._

class ASGInfo(credentials: AWSCredentialsProvider) extends AWSRestartableActor with AmazonAutoScalingService {

  import actors.workflow.tasks.ASGInfo._

  override def receive: Receive = {
    case msg: ASGInServiceInstancesAndELBSQuery =>

      val request = new DescribeAutoScalingGroupsRequest()
                    .withAutoScalingGroupNames(msg.asgName)

      val awsClient = autoScalingClient(credentials)
      val asg = awsClient.describeAutoScalingGroups(request).getAutoScalingGroups.asScala.toSeq(0)

      val instanceIds = asg.getInstances.asScala.toSeq.foldLeft(Seq.empty[String]) {
        (sum, i) => i.getLifecycleState match {
          case "InService" => sum :+ i.getInstanceId
          case _ => sum
        }
      }

      val elbNames = asg.getLoadBalancerNames.asScala.toSeq

      context.parent ! ASGInServiceInstancesAndELBSResult(elbNames, instanceIds)

    case m: Any =>
      log.debug(s"unhandled message: ${m.toString}")
  }
}

object ASGInfo extends PropFactory {
  case class ASGInServiceInstancesAndELBSQuery(asgName: String)
  case class ASGInServiceInstancesAndELBSResult(elbNames: Seq[String], instanceIds: Seq[String])

  override def props(args: Any*): Props = Props(classOf[ASGInfo], args: _*)
} 

//设置package包名称以及导入依赖的类
package actors.workflow.tasks

import actors.workflow.AWSRestartableActor
import akka.actor.Props
import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.services.cloudformation.model.DescribeStacksRequest
import utils.{AmazonCloudFormationService, PropFactory}

import scala.collection.JavaConverters._

class StackInfo(credentials: AWSCredentialsProvider) extends AWSRestartableActor with AmazonCloudFormationService {

  import actors.workflow.tasks.StackInfo._

  override def receive: Receive = {
    case query: StackASGNameQuery =>
      val describeStackRequest = new DescribeStacksRequest()
                                 .withStackName(query.stackName)

      val awsClient = cloudFormationClient(credentials)
      val stacksResults = awsClient.describeStacks(describeStackRequest).getStacks.asScala.toSeq

      stacksResults.length match {
        case 1 =>
          val stackOutputs = stacksResults.seq(0).getOutputs.asScala.toSeq
          val asgOutput = stackOutputs.filter(p => p.getOutputKey.equals("ChadashASG"))
          asgOutput.length match {
            case 1 => context.parent ! StackASGNameResponse(asgOutput(0).getOutputValue)
            case _ => throw new UnsupportedOperationException("missing ChadashASG output")
          }
        case _ => throw new UnsupportedOperationException("expected only one stack!")
      }

    case msg: StackIdQuery =>
      val describeStacksRequest = new DescribeStacksRequest()
                                  .withStackName(msg.stackName)

      val awsClient = cloudFormationClient(credentials)
      val stacksResults = awsClient.describeStacks(describeStacksRequest).getStacks.asScala.toSeq

      stacksResults.length match {
        case 1 =>
          val stackId = stacksResults.seq(0).getStackId
          context.parent ! StackIdResponse(stackId)
        case _ => throw new UnsupportedOperationException("expected only one stack!")
      }
  }
}

object StackInfo extends PropFactory {
  case class StackASGNameQuery(stackName: String)
  case class StackASGNameResponse(asgName: String)
  case class StackIdQuery(stackName: String)
  case class StackIdResponse(stackId: String)

  override def props(args: Any*): Props = Props(classOf[StackInfo], args: _*)
} 

//设置package包名称以及导入依赖的类
package actors.workflow.tasks

import actors.workflow.AWSRestartableActor
import akka.actor.Props
import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.services.cloudformation.AmazonCloudFormation
import com.amazonaws.services.cloudformation.model.{StackSummary, ListStacksRequest}
import com.amazonaws.services.cloudformation.model.StackStatus._
import utils.{AmazonCloudFormationService, PropFactory}

import scala.annotation.tailrec
import scala.collection.JavaConverters._

class StackList(credentials: AWSCredentialsProvider) extends AWSRestartableActor with AmazonCloudFormationService {

  import actors.workflow.tasks.StackList._

  override def receive: Receive = {
    case query: ListNonDeletedStacksStartingWithName =>

      val listStackRequest = new ListStacksRequest()
                              .withStackStatusFilters(stackStatusFilters.toArray: _*)

      val awsClient = cloudFormationClient(credentials)
      val results = getStackSummaries(awsClient, listStackRequest)
      val filteredResults = results.filter(p => p.getStackName.startsWith(query.stackName))
      val filteredStackNames = filteredResults.foldLeft(Seq.empty[String])((sum, i) => sum :+ i.getStackName)

      context.parent ! FilteredStacks(filteredStackNames)
  }

  def getStackSummaries(awsClient: AmazonCloudFormation, listStacksRequest: ListStacksRequest): Seq[StackSummary] = {
    @tailrec
    def impl(awsClient: AmazonCloudFormation, listStacksRequest: ListStacksRequest, sum: Seq[StackSummary]): Seq[StackSummary] = {
      val results = awsClient.listStacks(listStacksRequest)
      val pageSummaries = results.getStackSummaries.asScala.toSeq
      val totalSummaries = sum ++ pageSummaries
      if (results.getNextToken == null) {
        totalSummaries
      } else {
        val nextPageRequest = listStacksRequest.withNextToken(results.getNextToken)
        impl(awsClient, nextPageRequest, totalSummaries)
      }
    }
    impl(awsClient, listStacksRequest, Seq.empty[StackSummary])
  }
}

object StackList extends PropFactory {
  //only consider stacks that are not in the set of: delete_complete, delete_failed
  val stackStatusFilters = Seq(CREATE_IN_PROGRESS, CREATE_COMPLETE, CREATE_FAILED, ROLLBACK_IN_PROGRESS, ROLLBACK_FAILED, ROLLBACK_COMPLETE,
    DELETE_IN_PROGRESS, UPDATE_COMPLETE_CLEANUP_IN_PROGRESS, UPDATE_IN_PROGRESS, UPDATE_COMPLETE, UPDATE_ROLLBACK_COMPLETE, UPDATE_ROLLBACK_COMPLETE_CLEANUP_IN_PROGRESS,
    UPDATE_ROLLBACK_FAILED, UPDATE_ROLLBACK_IN_PROGRESS)

  case class ListNonDeletedStacksStartingWithName(stackName: String)
  case class FilteredStacks(stackList: Seq[String])

  override def props(args: Any*): Props = Props(classOf[StackList], args: _*)
} 

//设置package包名称以及导入依赖的类
package actors.workflow.tasks

import actors.workflow.AWSRestartableActor
import akka.actor.Props
import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.services.elasticloadbalancing.model.{DescribeInstanceHealthRequest, Instance}
import utils.{AmazonElasticLoadBalancingService, PropFactory}

import scala.collection.JavaConverters._

class ELBHealthyInstanceChecker(credentials: AWSCredentialsProvider) extends AWSRestartableActor
                                                                     with AmazonElasticLoadBalancingService {

  import actors.workflow.tasks.ELBHealthyInstanceChecker._

  override def receive: Receive = {
    case msg: ELBIsInstanceListHealthy =>

      val elbInstances: Seq[Instance] = msg.instances.foldLeft(Seq.empty[Instance])((sum, i) => sum :+ new Instance(i))
      val instanceHealthRequest = new DescribeInstanceHealthRequest()
                                  .withInstances(elbInstances.asJava)
                                  .withLoadBalancerName(msg.elbName)

      val awsClient = elasticLoadBalancingClient(credentials)
      val instanceStates = awsClient.describeInstanceHealth(instanceHealthRequest).getInstanceStates.asScala.toSeq
      val unhealthyInstances = instanceStates.filter(p => p.getState != "InService")

      unhealthyInstances.size match {
        case i if i > 0 =>
          context.parent ! ELBInstanceListNotHealthy(i, msg.elbName)
        case i if i == 0 =>
          context.parent ! ELBInstanceListAllHealthy(msg.elbName)
      }
  }
}

object ELBHealthyInstanceChecker extends PropFactory {
  case class ELBIsInstanceListHealthy(elbName: String, instances: Seq[String])
  case class ELBInstanceListNotHealthy(unhealthyInstances: Int, elbName: String)
  case class ELBInstanceListAllHealthy(elbName: String)

  override def props(args: Any*): Props = Props(classOf[ELBHealthyInstanceChecker], args: _*)
} 

//设置package包名称以及导入依赖的类
package actors.workflow.tasks

import actors.workflow.AWSRestartableActor
import akka.actor.Props
import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.services.autoscaling.model.{DescribeAutoScalingGroupsRequest, SetDesiredCapacityRequest}
import utils.{AmazonAutoScalingService, PropFactory}

import scala.collection.JavaConverters._

class ASGSize(credentials: AWSCredentialsProvider) extends AWSRestartableActor with AmazonAutoScalingService {

  import actors.workflow.tasks.ASGSize._

  override def receive: Receive = {
    case msg: ASGDesiredSizeQuery =>
      val asgFilter = new DescribeAutoScalingGroupsRequest()
                      .withAutoScalingGroupNames(msg.asgName)

      val awsClient = autoScalingClient(credentials)
      val result = awsClient.describeAutoScalingGroups(asgFilter).getAutoScalingGroups.asScala.toSeq
      context.parent ! ASGDesiredSizeResult(result(0).getDesiredCapacity)

    case msg: ASGSetDesiredSizeCommand =>
      val desiredCapRequest = new SetDesiredCapacityRequest()
                              .withDesiredCapacity(msg.size)
                              .withAutoScalingGroupName(msg.asgName)

      val awsClient = autoScalingClient(credentials)
      awsClient.setDesiredCapacity(desiredCapRequest)
      context.parent ! ASGSetDesiredSizeRequested
  }
}

object ASGSize extends PropFactory {
  case class ASGDesiredSizeQuery(asgName: String)
  case class ASGDesiredSizeResult(size: Int)
  case class ASGSetDesiredSizeCommand(asgName: String, size: Int)
  case object ASGSetDesiredSizeRequested

  override def props(args: Any*): Props = Props(classOf[ASGSize], args: _*)
} 

//设置package包名称以及导入依赖的类
package actors.workflow.tasks

import actors.workflow.AWSRestartableActor
import akka.actor.Props
import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.services.cloudformation.model.DeleteStackRequest
import utils.{AmazonCloudFormationService, PropFactory}

class DeleteStack(credentials: AWSCredentialsProvider) extends AWSRestartableActor with AmazonCloudFormationService {

  import actors.workflow.tasks.DeleteStack._

  override def receive: Receive = {
    case msg: DeleteStackCommand =>
      val delStackReq = new DeleteStackRequest()
                        .withStackName(msg.stackName)

      val awsClient = cloudFormationClient(credentials)
      awsClient.deleteStack(delStackReq)

      context.parent ! StackDeleteRequested
  }
}

object DeleteStack extends PropFactory {
  case class DeleteStackCommand(stackName: String)
  case object StackDeleteRequested

  override def props(args: Any*): Props = Props(classOf[DeleteStack], args: _*)
} 

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

import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.services.autoscaling.{AmazonAutoScaling, AmazonAutoScalingClient}
import com.amazonaws.services.cloudformation.{AmazonCloudFormation, AmazonCloudFormationClient}
import com.amazonaws.services.elasticloadbalancing.{AmazonElasticLoadBalancing, AmazonElasticLoadBalancingClient}
import com.amazonaws.services.s3.{AmazonS3, AmazonS3Client}

trait AmazonAutoScalingService {
  def autoScalingClient(credentials: AWSCredentialsProvider): AmazonAutoScaling = new AmazonAutoScalingClient(credentials)
}

trait AmazonCloudFormationService {
  def cloudFormationClient(credentials: AWSCredentialsProvider): AmazonCloudFormation = new AmazonCloudFormationClient(credentials)
}

trait AmazonElasticLoadBalancingService {
  def elasticLoadBalancingClient(credentials: AWSCredentialsProvider): AmazonElasticLoadBalancing = new AmazonElasticLoadBalancingClient(credentials)
}

trait AmazonS3Service {
  def s3Client(credentials: AWSCredentialsProvider): AmazonS3 = new AmazonS3Client(credentials)
} 

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

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.Http.HostConnectionPool
import akka.http.scaladsl.model.{HttpRequest, HttpResponse}
import akka.stream.Materializer
import akka.stream.scaladsl.Flow
import com.amazonaws.auth.AWSCredentialsProvider

import scala.concurrent.Future
import scala.util.Try

case class HttpConnectionFlow(connectionSettings: ConnectionSettings, service: AWSService)(implicit system: ActorSystem, materializer: Materializer) extends ConnectionFlow[HttpRequest, HttpResponse]{

  val connectionFlow: Flow[(HttpRequest, Int), (Try[HttpResponse], Int), HostConnectionPool] = {
    if (connectionSettings.scheme == "http")
      Http()(system).cachedHostConnectionPool[Int](connectionSettings.host, connectionSettings.port)
    else
      Http()(system).cachedHostConnectionPoolHttps[Int](connectionSettings.host, connectionSettings.port)
  }

  val doubleUrlEncoding: Boolean = connectionSettings.doubleUrlEncoding

  val credentialsProvider: AWSCredentialsProvider = connectionSettings.credentialsProvider

  val endpoint: String = s"${connectionSettings.scheme}://${connectionSettings.host}:${connectionSettings.port}"
} 

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

import java.net.InetSocketAddress

import akka.event.LoggingAdapter
import akka.http.scaladsl.settings.ClientConnectionSettings
import com.amazonaws.auth.AWSCredentialsProvider

case class ConnectionSettings(scheme: String = "https",
                              host: String,
                              port: Int = 443,
                              credentialsProvider: AWSCredentialsProvider,
                              localAddress: Option[InetSocketAddress] = None,
                              settings: Option[ClientConnectionSettings] = None,
                              log: Option[LoggingAdapter] = None,
                              doubleUrlEncoding: Boolean = true) {
  require(scheme == "https" || scheme == "http")
} 

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

import com.amazonaws.auth.AWSCredentialsProvider
import scala.concurrent.duration._

case class KinesisStreamReaderConfig(
    streamName: String,
    app: String,
    stage: String,
    mode: String,
    suffix: Option[String],
    kinesisCredentialsProvider: AWSCredentialsProvider,
    dynamoCredentialsProvider: AWSCredentialsProvider,
    awsRegion: String,
    checkpointInterval: Duration = 30.second,
    maxCheckpointBatchSize: Int = 20,
    maxRecords: Int = 10000,
    idleTimeBetweenReadsInMillis: Long = 2000L

) {

  lazy val applicationName: String = s"${streamName}_${app}-${mode}-${stage.toUpperCase}${suffix.getOrElse("")}"

} 

//设置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 com.gu.cm

import com.amazonaws.auth.{AWSCredentialsProvider, DefaultAWSCredentialsProviderChain}
import com.amazonaws.regions.RegionUtils
import com.amazonaws.regions.ServiceAbbreviations.S3
import com.amazonaws.services.s3.AmazonS3Client
import com.amazonaws.services.s3.model.GetObjectRequest
import com.typesafe.config.{Config, ConfigFactory}

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

class S3ConfigurationSource(s3: AmazonS3Client, identity: Identity, bucket: String, version: Option[Int]) extends ConfigurationSource {

  override def load: Config = {
    val configPath = versionedFilePath(identity, version)
    val request = new GetObjectRequest(bucket, configPath)
    val config = for {
      result <- Try(s3.getObject(request))
      item <- Try(scala.io.Source.fromInputStream(result.getObjectContent, "UTF-8").mkString)
    } yield {
      ConfigFactory.parseString(item)
    }

    config match {
      case Success(theConfig) => theConfig
      case Failure(theFailure) => ConfigFactory.empty(s"no s3 config (or failed to load) for bucket=$bucket path=$configPath, exception=[$theFailure]")
    }
  }

  def versionedFilePath(identity: Identity, version:Option[Int]): String = {
    val fileVersion = version.map(".v" + _).getOrElse("")
    s"config/${identity.region}-${identity.stack}$fileVersion.conf"
  }
}

object S3ConfigurationSource {
  def apply(identity: Identity, bucket: String, credentials: AWSCredentialsProvider = new DefaultAWSCredentialsProviderChain(), version: Option[Int] = None): S3ConfigurationSource = {
    val s3 = {
      val client = new AmazonS3Client(credentials)
      client.setRegion(RegionUtils.getRegion(identity.region))
      client.setEndpoint(RegionUtils.getRegion(identity.region).getServiceEndpoint(S3))
      client
    }
    new S3ConfigurationSource(s3, identity, bucket, version)
  }
} 

//设置package包名称以及导入依赖的类
import java.net.InetAddress
import java.util
import java.util.UUID

import com.amazonaws.auth.{AWSCredentialsProvider, BasicAWSCredentials}
import com.amazonaws.internal.StaticCredentialsProvider
import com.amazonaws.services.kinesis.clientlibrary.interfaces.{IRecordProcessor, IRecordProcessorCheckpointer, IRecordProcessorFactory}
import com.amazonaws.services.kinesis.clientlibrary.lib.worker.{InitialPositionInStream, KinesisClientLibConfiguration, Worker}
import com.amazonaws.services.kinesis.clientlibrary.types.ShutdownReason
import com.amazonaws.services.kinesis.model.Record

object TailMain {
  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"
  val idleTimeBetweenReadsInMillis = 3000

  def main(args: Array[String]): Unit = {
    val workerId = InetAddress.getLocalHost.getCanonicalHostName + ":" + UUID.randomUUID
    val credentialsProvider: AWSCredentialsProvider = new StaticCredentialsProvider(new BasicAWSCredentials(accessKeyId, secretAccessKey))

    val kclConf = new KinesisClientLibConfiguration(appName, streamName, credentialsProvider, workerId)
      .withInitialPositionInStream(InitialPositionInStream.valueOf(initialPosition))
      .withRegionName(region)
      .withIdleTimeBetweenReadsInMillis(idleTimeBetweenReadsInMillis)

    println(s"worker start. name:$appName stream:$streamName workerId:$workerId")
    val tailWorker = new Worker(StreamTailProcessor.processorFactory, kclConf)
    tailWorker.run()
  }
}

class StreamTailProcessor extends IRecordProcessor{
  override def shutdown(checkpointer: IRecordProcessorCheckpointer, reason: ShutdownReason): Unit = {
    println(s"Shutting down record processor")
  }

  override def initialize(shardId: String): Unit = {
    println(s"Initialising record processor for shard: $shardId")
  }

  override def processRecords(records: util.List[Record], checkpointer: IRecordProcessorCheckpointer): Unit = {
    import scala.collection.JavaConversions._
    records foreach { r =>
      val line = new String(r.getData.array)
      println(s"[stream-tail] $line")
    }
  }
}

object StreamTailProcessor {
  def processorFactory = new IRecordProcessorFactory {
    def createProcessor(): IRecordProcessor = new StreamTailProcessor
  }
} 

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

    }
  }
}