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

import akka.actor.{Actor, ActorLogging, ActorSystem, Props}
import akka.stream.{ActorMaterializer, ActorMaterializerSettings}

class CrawlerActor(kafkaProducer: SimpleKafkaProducer) extends Actor with ActorLogging with AutoMarshaller {
  import CrawlerActor._
  import context.dispatcher

  implicit val system: ActorSystem = context.system
  final implicit val materializer: ActorMaterializer = ActorMaterializer(ActorMaterializerSettings(system))

  val statisticDataFetcher = new StatisticDataFetcher()
  val dataSender: DataSender = new DataSender(kafkaProducer)


  def receive = {
  	case FetchData =>
      statisticDataFetcher.getStatistics() map { statistics =>
        dataSender.send(statistics)
      }
  }
}

object CrawlerActor {
  val kafkaHelpers = new KafkaHelpers()
  val props = Props(classOf[CrawlerActor], new SimpleKafkaProducer(kafkaHelpers.kafkaSocket(), kafkaHelpers.topic()))
  case object FetchData
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp

import akka.actor.{ ActorSystem, Props }
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.Directives._
import akka.stream.{ ActorMaterializer, ActorMaterializerSettings, Supervision }
import com.typesafe.config.ConfigFactory

import net.ruippeixotog.scalafbp.http._
import net.ruippeixotog.scalafbp.protocol.MainProtocolActor
import net.ruippeixotog.scalafbp.runtime.{ ComponentLoader, ComponentRegistry, GraphStore }

object Server extends App with WsRuntimeHttpService with RegisterHttpService with RegistryHttpService
    with UiHttpService {

  implicit val system = ActorSystem()
  implicit val ec = system.dispatcher

  val decider: Supervision.Decider = { e =>
    log.error("Unhandled exception in stream", e)
    Supervision.Stop
  }

  implicit val materializer = ActorMaterializer(
    ActorMaterializerSettings(system).withSupervisionStrategy(decider))

  val config = ConfigFactory.load.getConfig("scalafbp")
  val registryConfig = config.getConfig("registry")
  val runtimeConfig = config.getConfig("runtime")

  val runtimeId = config.getString("runtime-id")
  val secret = config.getString("secret")

  val host = config.getString("host")
  val port = config.getInt("port")

  val disableUi = config.getBoolean("disable-ui")

  // the registry of components that will be made available to clients
  val compRegistry = system.actorOf(ComponentRegistry.props(ComponentLoader.allInClasspath))

  // an object responsible for storing and managing the graph definitions currently in the runtime
  val graphStore = system.actorOf(Props(new GraphStore))

  // actor that receives incoming messages (as `Message` objects) and translates them into actions using the above
  // constructs
  val protocolActor = system.actorOf(Props(
    new MainProtocolActor(runtimeId, secret, compRegistry, graphStore, runtimeConfig)))

  // all the routes offered by this server
  val routes = registrationRoutes ~ registryRoutes ~ wsRuntimeRoutes ~ uiRoutes

  Http().bindAndHandle(routes, host, port).foreach { binding =>
    log.info(s"Bound to ${binding.localAddress}")
    onBind(binding)
  }
} 

//设置package包名称以及导入依赖的类
package com.scalaio.http.client.actor

import akka.actor.{Actor, ActorLogging, ActorRef, Props}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model._
import akka.stream.{ActorMaterializer, ActorMaterializerSettings}
import akka.util.{ByteString, Timeout}
import play.api.libs.json.Json

import scala.concurrent.Future
import scala.concurrent.duration._

class HttpClientAsActor(notifier: ActorRef) extends Actor with ActorLogging {

  import akka.pattern.pipe
  import context.dispatcher
  implicit val timeout = Timeout(5 seconds)
  implicit val materializer = ActorMaterializer(ActorMaterializerSettings(context.system))

  val http = Http(context.system)

  override def preStart() = {
    http
      .singleRequest(HttpRequest(method = GET, uri = "https://jsonplaceholder.typicode.com/posts/1"))
      .pipeTo(self)
  }

  def receive = {
    case HttpResponse(StatusCodes.OK, headers, entity, _) =>
      val response: Future[ByteString] = entity.dataBytes.runFold(ByteString(""))(_ ++ _)
      log.info(s"got response $headers $entity")
      response pipeTo self
      context become handlingMessage

    case [email protected](code, _, _, _) =>
      log.warning("Request failed, response code: " + code)
      resp.discardEntityBytes()
  }

  def handlingMessage: Receive = {
    case content: ByteString =>
      log.info("Success was OK: " + content)
      val contentAsString = (Json.parse(content.utf8String) \ "title").as[String]
      notifier ! contentAsString
      context become receive
  }

}

object HttpClientAsActor {

  def props(notifier: ActorRef) = Props(classOf[HttpClientAsActor], notifier)

} 

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

import akka.actor.ActorSystem
import akka.stream.ActorMaterializerSettings

trait MaterializerSupport {
  mixin: {
    def system: ActorSystem
    def log: org.slf4j.Logger
  } =>

  lazy val decider: akka.stream.Supervision.Decider = {
    case ex: Throwable ?
      log.debug("Akka-Stream error:" + ex.getMessage)
      akka.stream.Supervision.Stop
  }

  lazy val settings = ActorMaterializerSettings.create(system)
    .withInputBuffer(32, 32)
    .withSupervisionStrategy(decider)
    .withDispatcher("akka.stream-dispatcher")

  implicit val Mat = akka.stream.ActorMaterializer(settings)(system)
  implicit val Ex = Mat.executionContext
} 

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

import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, ActorMaterializerSettings}
import akka.stream.scaladsl.{Sink, Source}
import akka.testkit.TestKit
import akka.util.ByteString
import org.scalatest.time.{Millis, Seconds, Span}
import org.scalatest.{BeforeAndAfterAll, FlatSpecLike, Matchers}
import org.scalatest.concurrent.ScalaFutures

class MemoryBufferSpec(_system: ActorSystem)
    extends TestKit(_system)
    with FlatSpecLike
    with Matchers
    with BeforeAndAfterAll
    with ScalaFutures {

  def this() = this(ActorSystem("MemoryBufferSpec"))

  implicit val defaultPatience =
    PatienceConfig(timeout = Span(5, Seconds), interval = Span(30, Millis))

  implicit val materializer = ActorMaterializer(ActorMaterializerSettings(system).withDebugLogging(true))

  "MemoryBuffer" should "emit a chunk on its output containg the concatenation of all input values" in {
    val result = Source(Vector(ByteString(1, 2, 3, 4, 5), ByteString(6, 7, 8, 9, 10, 11, 12), ByteString(13, 14)))
      .via(new MemoryBuffer(200))
      .runWith(Sink.seq)
      .futureValue

    result should have size (1)
    val chunk = result.head
    chunk.size should be(14)
    chunk.data.runWith(Sink.seq).futureValue should be(Seq(ByteString(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14)))
  }

  it should "fail if more than maxSize bytes are fed into it" in {
    whenReady(
      Source(Vector(ByteString(1, 2, 3, 4, 5), ByteString(6, 7, 8, 9, 10, 11, 12), ByteString(13, 14)))
        .via(new MemoryBuffer(10))
        .runWith(Sink.seq)
        .failed
    ) { e =>
      e shouldBe a[IllegalStateException]
    }
  }
} 

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

import akka.testkit.TestKit
import akka.stream.ActorMaterializerSettings
import org.scalatest.BeforeAndAfterAll
import org.scalatest.concurrent.ScalaFutures
import akka.stream.ActorMaterializer
import akka.actor.ActorSystem
import org.scalatest.Matchers
import org.scalatest.FlatSpecLike
import akka.stream.scaladsl.Source
import akka.stream.scaladsl.Flow
import akka.util.ByteString
import akka.stream.scaladsl.Sink
import org.scalatest.time.{Millis, Seconds, Span}
import scala.concurrent.duration._

class SplitAfterSizeSpec(_system: ActorSystem)
    extends TestKit(_system)
    with FlatSpecLike
    with Matchers
    with BeforeAndAfterAll
    with ScalaFutures {

  def this() = this(ActorSystem("SplitAfterSizeSpec"))
  implicit val defaultPatience =
    PatienceConfig(timeout = Span(5, Seconds), interval = Span(30, Millis))

  implicit val materializer = ActorMaterializer(ActorMaterializerSettings(system).withDebugLogging(true))

  "SplitAfterSize" should "yield a single empty substream on no input" in {
    Source
      .empty[ByteString]
      .via(
        SplitAfterSize(10)(Flow[ByteString]).concatSubstreams
      )
      .runWith(Sink.seq)
      .futureValue should be(Seq.empty)
  }

  it should "start a new stream after the element that makes it reach a maximum, but not split the element itself" in {
    Source(Vector(ByteString(1, 2, 3, 4, 5), ByteString(6, 7, 8, 9, 10, 11, 12), ByteString(13, 14)))
      .via(
        SplitAfterSize(10)(Flow[ByteString]).prefixAndTail(10).map { case (prefix, tail) => prefix }.concatSubstreams
      )
      .runWith(Sink.seq)
      .futureValue should be(
      Seq(
        Seq(ByteString(1, 2, 3, 4, 5), ByteString(6, 7, 8, 9, 10, 11, 12)),
        Seq(ByteString(13, 14))
      )
    )
  }

} 

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

import akka.Done
import akka.actor.ActorSystem
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ConsumerSettings, Subscriptions}
import akka.stream.scaladsl.Sink
import akka.stream.{ActorMaterializer, ActorMaterializerSettings}
import cats.data.Xor
import com.typesafe.config.ConfigFactory
import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.clients.consumer.internals.PartitionAssignor.Subscription
import org.apache.kafka.common.serialization.{ByteArrayDeserializer, StringDeserializer}
import io.circe._
import io.circe.generic.auto._
import cats.data.Xor.{Left, Right}
import model.Employee

import scala.concurrent.Future

object StreamConsumer  extends App{
  implicit val actorSystem = ActorSystem("consumer-actors", ConfigFactory.load())
  implicit val materializer = ActorMaterializer(ActorMaterializerSettings(actorSystem))

  lazy val consumerSettings = ConsumerSettings(actorSystem, new ByteArrayDeserializer, new StringDeserializer)
    .withBootstrapServers("localhost:9092")
    .withGroupId("group13")
    .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")//"latest")
    .withProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true")
    .withProperty(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000")

  lazy val subscription = Subscriptions.topics("raw-data-1")
  lazy val db = new Processor()
  Consumer.plainSource(consumerSettings, subscription)
      .mapAsync(4){
        db.processMessage
      }
    .runWith(Sink.ignore)

}

class Processor {
  def processMessage(record: ConsumerRecord[Array[Byte], String]): Future[Done] ={
    println(s"DB.save: ${record.value()}")

    Option(record.value()).foreach{ jsonString =>
      val mayBeEmp: Xor[Error, Employee] = jawn.decode[Employee](jsonString)
      mayBeEmp match {
        case Left(error) => println(error)
        case Right(emp) => println(s"employee name: ${emp.name}")
      }
    }
    Future.successful(Done)  }
} 

//设置package包名称以及导入依赖的类
import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Supervision}
import akka.testkit.{ImplicitSender, TestKit, TestKitBase}
import com.amazonaws.services.sqs.AmazonSQSAsync
import com.taxis99.amazon.sqs.SqsClientFactory
import com.typesafe.config.ConfigFactory
import org.scalatest._
import org.scalatest.concurrent.PatienceConfiguration
import org.scalatest.time._

import scala.concurrent.{ExecutionContext, Future}

package object test {
  trait BaseSpec extends FlatSpec with Matchers with OptionValues with PatienceConfiguration with RecoverMethods {
    implicit val defaultPatience =
      PatienceConfig(timeout =  Span(3, Seconds), interval = Span(5, Millis))
  }

  trait StreamSpec extends AsyncFlatSpec with Matchers with OptionValues with PatienceConfiguration
    with TestKitBase with ImplicitSender with BeforeAndAfterAll {

    implicit lazy val system: ActorSystem = ActorSystem("test", ConfigFactory.parseString("""
        akka.actor.deployment.default.dispatcher = "akka.test.calling-thread-dispatcher"
      """))

    override implicit def executionContext: ExecutionContext = system.dispatcher
    
    override implicit def patienceConfig = PatienceConfig(timeout =  Span(1, Minute), interval = Span(5, Millis))

    val decider: Supervision.Decider = {
      case _ => Supervision.Stop
    }
    val settings = ActorMaterializerSettings(system).withSupervisionStrategy(decider)

    implicit lazy val materializer = ActorMaterializer(settings)

    def withInMemoryQueue(testCode: (AmazonSQSAsync) => Future[Assertion]): Future[Assertion] = {
      val (server, aws) = SqsClientFactory.inMemory(Some(system))
      // "loan" the fixture to the test
      testCode(aws) andThen {
        case _ => server.stopAndWait()
      }
    }

    override def afterAll {
      TestKit.shutdownActorSystem(system)
    }
  }
} 

//设置package包名称以及导入依赖的类
import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Supervision}
import akka.testkit.{TestKit, TestKitBase}
import com.taxis99.amazon.sns.SnsClientFactory
import com.taxis99.amazon.sqs.SqsClientFactory
import com.typesafe.config.ConfigFactory
import org.scalatest._
import org.scalatest.concurrent.PatienceConfiguration
import org.scalatest.time.{Millis, Minute, Span}

import scala.concurrent.ExecutionContext

package object it {

  trait IntegrationSpec extends AsyncFlatSpec with Matchers with OptionValues with PatienceConfiguration
    with TestKitBase with BeforeAndAfterAll {

    implicit lazy val system: ActorSystem = ActorSystem("test", ConfigFactory.parseString("""
        akka.actor.deployment.default.dispatcher = "akka.test.calling-thread-dispatcher"
      """))

    override implicit def executionContext: ExecutionContext = system.dispatcher

    override implicit def patienceConfig = PatienceConfig(timeout =  Span(1, Minute), interval = Span(5, Millis))
    
    implicit lazy val amazonSqsConn = SqsClientFactory.atLocalhost(9324)
    implicit lazy val amazonSnsConn = SnsClientFactory.atLocalhost(9292)

    val decider: Supervision.Decider = {
      case _ => Supervision.Stop
    }
    val settings = ActorMaterializerSettings(system).withSupervisionStrategy(decider)

    implicit lazy val materializer = ActorMaterializer(settings)

    override def afterAll {
      TestKit.shutdownActorSystem(system)
    }
  }
} 

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

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.Supervision.Decider
import akka.stream.scaladsl.Source
import akka.stream.{ActorMaterializer, ActorMaterializerSettings}
import com.yannick_cw.elastic_indexer4s.elasticsearch.elasic_config.ElasticWriteConfig
import com.sksamuel.elastic4s.Indexable
import cats.instances.future.catsStdInstancesForFuture
import com.sksamuel.elastic4s.streams.RequestBuilder
import com.yannick_cw.elastic_indexer4s.elasticsearch.ElasticseachInterpreter
import com.yannick_cw.elastic_indexer4s.indexing_logic.IndexableStream
import com.sksamuel.elastic4s.ElasticDsl._

import scala.concurrent.{ExecutionContext, Future}

class ElasticIndexer4s(esConf: ElasticWriteConfig)(implicit system: ActorSystem,
                                                   materializer: ActorMaterializer,
                                                   ex: ExecutionContext) {

  
  def fromBuilder[Entity](source: Source[Entity, NotUsed])(
      implicit requestBuilder: RequestBuilder[Entity]): IndexableStream[Entity, Future] =
    new IndexableStream[Entity, Future](source, new ElasticseachInterpreter[Entity](esConf))(
      catsStdInstancesForFuture(ex))

  def withDecider(decider: Decider): ElasticIndexer4s = {
    val materializer = ActorMaterializer(
      ActorMaterializerSettings(system).withSupervisionStrategy(decider))
    new ElasticIndexer4s(esConf)(system, materializer, ex)
  }
}

object ElasticIndexer4s {
  def apply(esConf: ElasticWriteConfig,
            system: ActorSystem,
            materializer: ActorMaterializer,
            ex: ExecutionContext): ElasticIndexer4s =
    new ElasticIndexer4s(esConf)(system, materializer, ex)

  def apply(esConf: ElasticWriteConfig): ElasticIndexer4s = {
    implicit val system = ActorSystem()
    implicit val ex = system.dispatcher
    implicit val materializer = ActorMaterializer()
    new ElasticIndexer4s(esConf)
  }
} 

//设置package包名称以及导入依赖的类
package com.tpalanga.test.account.api.users

import akka.actor.ActorSystem
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.{ContentTypes, RequestEntity}
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Materializer}
import com.tpalanga.test.account.api.users.model.{NewUser, User, Users}
import com.tpalanga.test.config.TestConfig
import com.tpalanga.testlib.test.client.{NoEntity, Response, RestServiceClient}
import com.tpalanga.testlib.test.config.RestServiceConfig
import com.typesafe.scalalogging.LazyLogging

import scala.concurrent.{ExecutionContext, Future}

class AccountServiceRestClient(val restServiceConfig: RestServiceConfig)
                              (implicit val testConfig: TestConfig, val system: ActorSystem)
extends RestServiceClient with LazyLogging {
  import NoEntity.DataFormats._
  import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
  import com.tpalanga.test.account.api.users.model.UserJsonProtocol._

  logger.debug(s"AccountServiceRestServiceClient: $restServiceConfig")
  private implicit val materializer: Materializer = ActorMaterializer(ActorMaterializerSettings(system))

  def userRetrieve(id: String)(implicit ec: ExecutionContext): Future[Response[User]] =
    client.get(s"/data/users/$id").map { httpResponse =>
      Response[User](httpResponse)
    }

  def userCreate(user: NewUser)(implicit ec: ExecutionContext): Future[Response[User]] =
    for {
      entity <- Marshal(user).to[RequestEntity]
      httpResponse <- client.post(s"/data/users", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[User](httpResponse)

  def userUpdate(user: User)(implicit ec: ExecutionContext): Future[Response[User]] =
    for {
      entity <- Marshal(user).to[RequestEntity]
      httpResponse <- client.put(s"/data/users/${user.id}", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[User](httpResponse)

  def userDelete(id: String)(implicit ec: ExecutionContext): Future[Response[NoEntity]] =
    client.delete(s"/data/users/$id").map { httpResponse =>
      Response[NoEntity](httpResponse)
    }

  def userList()(implicit ec: ExecutionContext): Future[Response[Users]] =
    client.get(s"/data/users").map { httpResponse =>
      Response[Users](httpResponse)
    }
} 

//设置package包名称以及导入依赖的类
package com.tpalanga.testlib.test.client.impl

import akka.actor.ActorSystem
import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.{ContentTypes, RequestEntity}
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Materializer}
import com.tpalanga.testlib.test.client.{NoEntity, Response, RestServiceClient}
import com.tpalanga.testlib.test.config.RestServiceConfig
import com.typesafe.scalalogging.LazyLogging

import scala.concurrent.{ExecutionContext, Future}


object NewsletterServiceRestClient {
  type NewsletterServiceRestClientFactory = (RestServiceConfig, ActorSystem) => NewsletterServiceRestClient

  def defaultFactory: NewsletterServiceRestClientFactory =
    (config, system) => new NewsletterServiceRestClient(config)(system)

}

class NewsletterServiceRestClient(val restServiceConfig: RestServiceConfig)
                                 (implicit val system: ActorSystem)
  extends RestServiceClient with LazyLogging {
  import NoEntity.DataFormats._
  import SubscriberJsonProtocol._
  import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._

  logger.debug(s"NewsletterServiceRestServiceClient: $restServiceConfig")
  private implicit val materializer: Materializer = ActorMaterializer(ActorMaterializerSettings(system))

  def subscriberRetrieve(id: String)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    client.get(s"/data/subscribers/$id").map { httpResponse =>
      Response[Subscriber](httpResponse)
    }

  def subscriberCreate(subscriber: Subscriber)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    for {
      entity <- Marshal(subscriber).to[RequestEntity]
      httpResponse <- client.post(s"/data/subscribers", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[Subscriber](httpResponse)

  def subscriberUpdate(user: Subscriber)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    for {
      entity <- Marshal(user).to[RequestEntity]
      httpResponse <- client.put(s"/data/subscribers/${user.id}", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[Subscriber](httpResponse)

  def subscriberDelete(id: String)(implicit ec: ExecutionContext): Future[Response[NoEntity]] =
    client.delete(s"/data/subscribers/$id").map { httpResponse =>
      Response[NoEntity](httpResponse)
    }

  def subscriberList()(implicit ec: ExecutionContext): Future[Response[Subscribers]] =
    client.get(s"/data/subscribers").map { httpResponse =>
      Response[Subscribers](httpResponse)
    }

} 

//设置package包名称以及导入依赖的类
package com.tpalanga.testlib.test.client

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model._
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Materializer}
import com.tpalanga.testlib.test.config.RestServiceConfig
import com.typesafe.scalalogging.LazyLogging

import scala.collection.immutable.Seq
import scala.concurrent.Future

class RestClient(config: RestServiceConfig)(implicit system: ActorSystem) extends LazyLogging {

  private implicit val materializer: Materializer = ActorMaterializer(ActorMaterializerSettings(system))
  private val http = Http(system)

  protected def uriFor(path: String): Uri = {
    val portext =
      if ((config.port == 80 && config.protocol == "http")
        || (config.port == 443 && config.protocol == "https")) ""
      else s":${config.port}"
    Uri(s"${config.protocol}://${config.host}$portext$path")
  }

  protected def sendRequest(httpRequest: HttpRequest): Future[HttpResponse] = {
    import system.dispatcher
    logger.debug(s"Sending request: $httpRequest")
    http.singleRequest(httpRequest).map { httpResponse =>
      logger.debug(s"Received response: $httpResponse")
      httpResponse
    }
  }

  def get(path: String, headers: Seq[HttpHeader] = Nil): Future[HttpResponse] =
    sendRequest(HttpRequest(GET, uriFor(path), headers))

  def post(path: String, headers: Seq[HttpHeader] = Nil, entity: RequestEntity = HttpEntity.Empty): Future[HttpResponse] =
    sendRequest(HttpRequest(POST, uriFor(path), headers, entity))

  def put(path: String, headers: Seq[HttpHeader] = Nil, entity: RequestEntity = HttpEntity.Empty): Future[HttpResponse] =
    sendRequest(HttpRequest(PUT, uriFor(path), headers, entity))

  def delete(path: String, headers: Seq[HttpHeader] = Nil): Future[HttpResponse] =
    sendRequest(HttpRequest(DELETE, uriFor(path), headers))
} 

//设置package包名称以及导入依赖的类
package com.tpalanga.test.newsletter.api.subscriber

import akka.http.scaladsl.marshalling.Marshal
import akka.http.scaladsl.model.{ContentTypes, RequestEntity}
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Materializer}
import com.tpalanga.test.config.TestConfig
import com.tpalanga.test.newsletter.api.subscriber.model.{Subscriber, Subscribers}
import com.tpalanga.testlib.test.client.{NoEntity, Response, RestServiceClient}
import com.tpalanga.testlib.test.config.RestServiceConfig

import scala.concurrent.{ExecutionContext, Future}

trait NewsletterServiceRestServiceClient extends RestServiceClient {
  import NoEntity.DataFormats._
  import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
  import com.tpalanga.test.newsletter.api.subscriber.model.SubscriberJsonProtocol._

  val testConfig: TestConfig

  override val restServiceConfig: RestServiceConfig = testConfig.restServiceConfig
  private implicit val materializer: Materializer = ActorMaterializer(ActorMaterializerSettings(system))

  def subscriberRetrieve(id: String)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    client.get(s"/data/subscribers/$id").map { httpResponse =>
      Response[Subscriber](httpResponse)
    }

  def subscriberCreate(subscriber: Subscriber)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    for {
      entity <- Marshal(subscriber).to[RequestEntity]
      httpResponse <- client.post(s"/data/subscribers", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[Subscriber](httpResponse)

  def subscriberUpdate(user: Subscriber)(implicit ec: ExecutionContext): Future[Response[Subscriber]] =
    for {
      entity <- Marshal(user).to[RequestEntity]
      httpResponse <- client.put(s"/data/subscribers/${user.id}", Nil, entity.withContentType(ContentTypes.`application/json`))
    } yield Response[Subscriber](httpResponse)

  def subscriberDelete(id: String)(implicit ec: ExecutionContext): Future[Response[NoEntity]] =
    client.delete(s"/data/subscribers/$id").map { httpResponse =>
      Response[NoEntity](httpResponse)
    }

  def subscriberList()(implicit ec: ExecutionContext): Future[Response[Subscribers]] =
    client.get(s"/data/subscribers").map { httpResponse =>
      Response[Subscribers](httpResponse)
    }

} 

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

import akka.actor.{Actor, ActorLogging}
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpRequest
import akka.stream.{ActorMaterializer, ActorMaterializerSettings}

import scala.concurrent.Future
import scala.concurrent.duration._
import scala.language.postfixOps


trait GymClient extends Actor with ActorLogging with JsonSupport {

  import context.dispatcher
  import spray.json._

  val gymServer: GymServer
  val http = Http(context.system)
  val timeout = 5 seconds

  implicit val materializer: ActorMaterializer = ActorMaterializer(ActorMaterializerSettings(context.system))

  override def preStart() = {
    self ! Initialize
  }

  protected def sendAction(action: Int): Future[StepResponse] = http
    .singleRequest(HttpRequest(uri = gymServer.actEndpoint(action)))
    .flatMap(r => r.entity.toStrict(timeout))
    .map(_.data.decodeString("UTF-8").parseJson.convertTo[StepResponse])

  protected def initialize(): Future[StepResponse] = http
    .singleRequest(HttpRequest(uri = gymServer.resetEndpoint()))
    .flatMap(r => r.entity.toStrict(timeout))
    .map(_.data.decodeString("UTF-8").parseJson.convertTo[InitResponse])
    .map(initResponse => initResponse.toStepResponse)
} 

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

import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Supervision}
import akka.testkit.TestKit
import org.scalatest.{BeforeAndAfterAll, FlatSpecLike, Matchers}


class LongActorRefPublisherSpec extends TestKit(ActorSystem("test-system")) with FlatSpecLike with Matchers with BeforeAndAfterAll {

  override def afterAll(): Unit = {
    super.afterAll()
    TestKit.shutdownActorSystem(system)
  }

  val decider: Supervision.Decider = {
    case e  => {
      println(s"Stopping Stream.. ${e.getMessage}")
      Supervision.Stop
    }
  }

  implicit val materializer = ActorMaterializer.create(ActorMaterializerSettings.create(system)
    .withDebugLogging(true)
    .withSupervisionStrategy(decider)
    .withAutoFusing(true), system)

  "Advert ID Actor" should "work" in {

  }

} 

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

import akka.actor.ActorSystem
import akka.stream.{ActorAttributes, ActorMaterializer, ActorMaterializerSettings, Supervision}
import akka.stream.scaladsl._

object HandlingErrorsApplication extends App {

  implicit val actorSystem = ActorSystem("HandlingErrors")

  val streamDecider: Supervision.Decider = {
    case e: IndexOutOfBoundsException =>
      println("Dropping element because of IndexOufOfBoundException. Resuming.")
      Supervision.Resume
    case _ => Supervision.Stop
  }

  val flowDecider: Supervision.Decider = {
    case e: IllegalArgumentException =>
      println("Dropping element because of IllegalArgumentException. Restarting.")
      Supervision.Restart
    case _ => Supervision.Stop
  }

  val actorMaterializerSettings = ActorMaterializerSettings(actorSystem).withSupervisionStrategy(streamDecider)
  implicit val actorMaterializer = ActorMaterializer(actorMaterializerSettings)

  val words = List("Handling", "Errors", "In", "Akka", "Streams", "")

  val flow = Flow[String].map(word => {
    if(word.length == 0) throw new IllegalArgumentException("Empty words are not allowed")
    word
  }).withAttributes(ActorAttributes.supervisionStrategy(flowDecider))

  Source(words).via(flow).map(array => array(2)).to(Sink.foreach(println)).run()
} 

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

import java.security.{DigestInputStream, MessageDigest}

import akka.actor.ActorSystem
import akka.stream.scaladsl.{Source, StreamConverters}
import akka.stream.{ActorMaterializer, ActorMaterializerSettings}
import akka.testkit.TestKit
import akka.util.ByteString
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.time.{Millis, Seconds, Span}
import org.scalatest.{FlatSpecLike, Matchers}

import scala.concurrent.Future

class StreamUtilsSpec(_system: ActorSystem) extends TestKit(_system) with FlatSpecLike with Matchers with ScalaFutures {
  def this() = this(ActorSystem("StreamUtilsSpec"))

  implicit val materializer = ActorMaterializer(ActorMaterializerSettings(system).withDebugLogging(true))

  implicit val defaultPatience =
    PatienceConfig(timeout =  Span(5, Seconds), interval = Span(30, Millis))


  "digest" should "calculate the digest of a short string" in {
    val bytes: Array[Byte] = "abcdefghijklmnopqrstuvwxyz".getBytes()
    val flow: Future[ByteString] = Source.single(ByteString(bytes)).runWith(StreamUtils.digest())

    val testDigest = MessageDigest.getInstance("SHA-256").digest(bytes)
    whenReady(flow) { result =>
      result should contain theSameElementsInOrderAs testDigest
    }
  }

  it should "calculate the digest of a file" in {
    val input = StreamConverters.fromInputStream(() => getClass.getResourceAsStream("/testdata.txt"))
    val flow: Future[ByteString] = input.runWith(StreamUtils.digest())

    val testDigest = MessageDigest.getInstance("SHA-256")
    val dis: DigestInputStream = new DigestInputStream(getClass.getResourceAsStream("/testdata.txt"), testDigest)

    val buffer = new Array[Byte](1024)

    var bytesRead: Int = dis.read(buffer)
    while (bytesRead > -1) {
      bytesRead = dis.read(buffer)
    }

    whenReady(flow) { result =>
      result should contain theSameElementsInOrderAs dis.getMessageDigest.digest()
    }

  }

} 

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

import akka.NotUsed
import akka.actor.ActorSystem
import akka.stream.Supervision.Decider
import akka.stream.scaladsl._
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Supervision}
import com.typesafe.config.ConfigFactory
import org.scalatest.{AsyncFunSuite, BeforeAndAfterAll, Matchers}

import scala.concurrent.duration._
import scala.concurrent.{Await, Future}

class StreamTest extends AsyncFunSuite with BeforeAndAfterAll with Matchers {
  implicit val system = ActorSystem.create("stream", ConfigFactory.load("test.conf"))
  implicit val dispatcher = system.dispatcher
  val decider: Decider = Supervision.restartingDecider
  val settings = ActorMaterializerSettings(system).withSupervisionStrategy(decider)
  implicit val materializer = ActorMaterializer(settings)

  val source: Source[Int, NotUsed] = Source(1 to 10)
  val flow: Flow[Int, Int, NotUsed] = Flow[Int].filter(_ % 2 == 0).map(_ * 2)
  val sink: Sink[Int, Future[Int]] = Sink.fold[Int, Int](0)(_ + _)

  override protected def afterAll(): Unit = {
    Await.result(system.terminate(), 1 second)
  }

  test("source") {
    source.runFold(0)(_ + _) map { _ shouldBe 55 }
    source.runReduce(_ + _) map { _ shouldBe 55 }
  }

  test("source ~ sink") {
    source.toMat(sink)(Keep.right).run map { _ shouldBe 55 }
    source.runWith(sink) map { _ shouldBe 55 }
  }

  test("source ~ flow ~ sink") {
    source.via(flow).toMat(sink)(Keep.right).run map { _ shouldBe 60 }
    source.via(flow).runWith(sink) map { _ shouldBe 60 }
  }

  test("flow ~ source ~ sink") {
    flow.runWith(source, sink)._2 map { _ shouldBe 60 }
  }
} 

//设置package包名称以及导入依赖的类
package com.github.kliewkliew.cornucopia.kafka

import akka.actor.ActorSystem
import akka.kafka.scaladsl.{Producer, Consumer => ConsumerDSL}
import akka.kafka.{ConsumerSettings, ProducerSettings, Subscriptions}
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Supervision}
import com.typesafe.config.ConfigFactory
import org.apache.kafka.common.serialization.{StringDeserializer, StringSerializer}
import org.slf4j.LoggerFactory

import scala.concurrent.duration._

object Config {
  object Cornucopia {
    private val config = ConfigFactory.load().getConfig("cornucopia")
    val minReshardWait = config.getInt("reshard.interval").seconds
    val gracePeriod = config.getInt("grace.period") * 1000
    val refreshTimeout = config.getInt("refresh.timeout") * 1000
    val batchPeriod = config.getInt("batch.period").seconds
  }

  object Consumer {
    private val kafkaConfig = ConfigFactory.load().getConfig("kafka")
    private val kafkaServers = kafkaConfig.getString("bootstrap.servers")
    private val kafkaConsumerConfig = kafkaConfig.getConfig("consumer")
    private val topic = kafkaConsumerConfig.getString("topic")
    private val groupId = kafkaConsumerConfig.getString("group.id")

    implicit val actorSystem = ActorSystem()
    // Log failures and resume processing
    private val decider: Supervision.Decider = { e =>
      LoggerFactory.getLogger(this.getClass).error("Failed to process event", e)
      Supervision.Resume
    }
    private val materializerSettings = ActorMaterializerSettings(actorSystem).withSupervisionStrategy(decider)

    private val sourceSettings = ConsumerSettings(actorSystem, new StringDeserializer, new StringDeserializer)
      .withBootstrapServers(kafkaServers)
      .withGroupId(groupId)
    private val subscription = Subscriptions.topics(topic)

    private val