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

import scala.concurrent.duration.Duration
import java.util.concurrent.{ CyclicBarrier, TimeUnit, TimeoutException }
import akka.actor.ActorSystem
import scala.concurrent.duration.FiniteDuration

class TestBarrierTimeoutException(message: String) extends RuntimeException(message)


object TestBarrier {
  val DefaultTimeout = Duration(5, TimeUnit.SECONDS)

  def apply(count: Int) = new TestBarrier(count)
}

class TestBarrier(count: Int) {
  private val barrier = new CyclicBarrier(count)

  def await()(implicit system: ActorSystem): Unit = await(TestBarrier.DefaultTimeout)

  def await(timeout: FiniteDuration)(implicit system: ActorSystem) {
    try {
      barrier.await(timeout.dilated.toNanos, TimeUnit.NANOSECONDS)
    } catch {
      case e: TimeoutException ?
        throw new TestBarrierTimeoutException("Timeout of %s and time factor of %s"
          format (timeout.toString, TestKitExtension(system).TestTimeFactor))
    }
  }

  def reset(): Unit = barrier.reset()
} 

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

import scala.concurrent.duration.Duration
import akka.actor.ActorSystem
import scala.concurrent.{ Await, CanAwait, Awaitable }
import java.util.concurrent.{ TimeoutException, CountDownLatch, TimeUnit }
import scala.concurrent.duration.FiniteDuration


object TestLatch {
  val DefaultTimeout = Duration(5, TimeUnit.SECONDS)

  def apply(count: Int = 1)(implicit system: ActorSystem) = new TestLatch(count)
}

class TestLatch(count: Int = 1)(implicit system: ActorSystem) extends Awaitable[Unit] {
  private var latch = new CountDownLatch(count)

  def countDown() = latch.countDown()
  def isOpen: Boolean = latch.getCount == 0
  def open() = while (!isOpen) countDown()
  def reset() = latch = new CountDownLatch(count)

  @throws(classOf[TimeoutException])
  def ready(atMost: Duration)(implicit permit: CanAwait) = {
    val waitTime = atMost match {
      case f: FiniteDuration ? f
      case _                 ? throw new IllegalArgumentException("TestLatch does not support waiting for " + atMost)
    }
    val opened = latch.await(waitTime.dilated.toNanos, TimeUnit.NANOSECONDS)
    if (!opened) throw new TimeoutException(
      "Timeout of %s with time factor of %s" format (atMost.toString, TestKitExtension(system).TestTimeFactor))
    this
  }
  @throws(classOf[Exception])
  def result(atMost: Duration)(implicit permit: CanAwait): Unit = {
    ready(atMost)
  }
} 

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

import java.util.concurrent.TimeoutException

import forms._
import models.{DAO, DAOComponent}
import play.api._
import play.api.libs.concurrent.Execution.Implicits._
import play.api.mvc._
import views._

import scala.concurrent.Future


  def delete(id: Long): Action[AnyContent] = Action.async { implicit request =>
    val futureEmpDel = dao.delete(id)
    futureEmpDel.map { result => Home.flashing("success" -> "Employee has been deleted") }.recover {
      case ex: TimeoutException =>
        Logger.error("Problem found in employee delete process")
        InternalServerError(ex.getMessage)
    }
  }

}

object Application extends Application(DAO) 

//设置package包名称以及导入依赖的类
package com.galacticfog.gestalt.lambda.impl

import java.util.concurrent.TimeoutException

import org.apache.mesos.Protos
import org.joda.time.DateTime
import scala.collection.mutable
import scala.concurrent.{Await, Future}
import scala.concurrent.duration._

case class EnvironmentCacheEntry( lambdaId : String, env : Protos.Environment, queuedTime : DateTime = DateTime.now )
class EnvironmentCache {

  val cache : mutable.Map[String, EnvironmentCacheEntry] = mutable.Map[String, EnvironmentCacheEntry]()
  val EXPIRATION_SECONDS = sys.env.getOrElse( "ENV_CACHE_EXPIRATION_SECONDS", "900" ).toInt

  def getEnvironment( lambdaId : String, env : Future[Map[String,String]] ) : Protos.Environment = {

    val cacheEntry = cache.get( lambdaId )

    if( !cacheEntry.isDefined || cacheEntry.get.queuedTime.plusSeconds( EXPIRATION_SECONDS ).isBeforeNow ) {
      //wait for the future
      try {
        val result = Await.result( env, 5 seconds )
        val builder = Protos.Environment.newBuilder
        result.foreach{ entry =>
          builder.addVariables( Protos.Environment.Variable.newBuilder
            .setName( entry._1 )
            .setValue( entry._2 )
          )
        }

        val newEnv =  builder.build
        cache( lambdaId ) = new EnvironmentCacheEntry( lambdaId, newEnv )
        newEnv
      }
      catch {
        case ex : TimeoutException => {
          println( "TIMEOUT" )
          Protos.Environment.newBuilder.build
        }
      }
    }
    else {
      cache( lambdaId ).env
    }
  }
} 

//设置package包名称以及导入依赖的类
package org.argus.jawa.core.util

import scala.concurrent.duration.FiniteDuration
import java.util.concurrent.TimeoutException

class MyTimeout(time: FiniteDuration) {
  private final var startTime: Long = System.currentTimeMillis()
  def refresh(): Unit = this.startTime = System.currentTimeMillis()
  def isTimeout: Boolean = {
    val currentTime = System.currentTimeMillis()
    (currentTime - startTime) >= time.toMillis
  }
  def isTimeoutThrow(): Unit = {
    if(isTimeout) throw new TimeoutException("Timeout after " + time.toMinutes + " minutes.")
  }
} 

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

import play.api._
import play.api.mvc._

import scala.concurrent.{ Await, Future }
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global
import java.util.concurrent.TimeoutException

object Application extends Controller {

  def foodtrucks = Action {
    try {
      val fut: Future[List[Truck]] = Future { Truck.listAll() }
      val response = Await.result(fut, 2500 milliseconds).asInstanceOf[List[Truck]]
      Ok(response.mkString("\n"))
    } catch {
      case e: TimeoutException => InternalServerError("Timeout exceeded")
      case e: Exception => InternalServerError("Unknown server error occured")
    }
  }

} 

//设置package包名称以及导入依赖的类
package swave.core.util

import java.util.concurrent.TimeoutException
import scala.concurrent.duration._
import scala.concurrent.{Await, Future, Promise}
import swave.core.StreamEnv

final class RichFuture[T](val underlying: Future[T]) extends AnyVal {

  def await(timeout: FiniteDuration = 1.second): T =
    underlying.value match {
      case Some(t)                          ? t.get
      case None if timeout == Duration.Zero ? throw new TimeoutException(s"Future was not completed")
      case _                                ? Await.result(underlying, timeout)
    }

  def delay(duration: FiniteDuration)(implicit env: StreamEnv): Future[T] = {
    import env.defaultDispatcher
    val promise = Promise[T]()
    underlying.onComplete { value ?
      env.scheduler.scheduleOnce(duration) { promise.complete(value); () }
    }
    promise.future
  }
} 

//设置package包名称以及导入依赖的类
package it.agilelab.bigdata.wasp.core.cluster

import java.util.concurrent.TimeoutException
import akka.actor._
import akka.pattern.gracefulStop
import akka.util.Timeout
import scala.concurrent.duration._
import it.agilelab.bigdata.wasp.core.WaspEvent.OutputStreamInitialized
import it.agilelab.bigdata.wasp.core.WaspEvent.NodeInitialized

abstract class ClusterAwareNodeGuardian extends ClusterAware {

  import akka.actor.SupervisorStrategy._

  // customize
  override val supervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1.minute) {
      case _: ActorInitializationException => Stop
      case _: IllegalArgumentException => Stop
      case _: IllegalStateException => Restart
      case _: TimeoutException => Escalate
      case _: Exception => Escalate
    }

  override def preStart(): Unit = {
    super.preStart()
    log.info("Starting at {}", cluster.selfAddress)
  }

  override def postStop(): Unit = {
    super.postStop()
    log.info("Node {} shutting down.", cluster.selfAddress)
    cluster.leave(self.path.address)
    gracefulShutdown()
  }

  override def receive: Actor.Receive = uninitialized orElse initialized orElse super.receive


  def uninitialized: Actor.Receive = {
    case OutputStreamInitialized => initialize()
  }

  def initialize(): Unit = {
    log.info(s"Node is transitioning from 'uninitialized' to 'initialized'")
    context.system.eventStream.publish(NodeInitialized)
  }

  def initialized: Actor.Receive

  def gracefulShutdown(): Unit = {
    val timeout = Timeout(5.seconds)
    context.children foreach (gracefulStop(_, timeout.duration))
    log.info(s"Graceful shutdown completed.")
  }
} 

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

import java.util.concurrent.TimeoutException

import akka.actor.Status.Failure
import akka.actor.{Actor, ActorRef, Props}
import akka.util.Timeout
import com.dbaktor.{ArticleBody, HttpResponse, ParseArticle, ParseHtmlArticle}
import com.dbaktor.messages.{GetRequest, SetRequest}

class TellDemoArticleParser(cacheActorPath: String,
                            httpClientActorPath: String,
                            acticleParserActorPath: String,
                            implicit val timeout: Timeout
                           ) extends Actor {
  val cacheActor = context.actorSelection(cacheActorPath)
  val httpClientActor = context.actorSelection(httpClientActorPath)
  val articleParserActor = context.actorSelection(acticleParserActorPath)

  implicit val ec = context.dispatcher

  
  private def buildExtraActor(senderRef: ActorRef, uri: String): ActorRef = {
    return context.actorOf(Props(new Actor{
      override def receive = {
        case "timeout" => //if we get timeout, then fail
          senderRef ! Failure(new TimeoutException("timeout!"))
          context.stop(self)

        case HttpResponse(body) => //If we get the http response first, we pass it to be parsed.
          articleParserActor ! ParseHtmlArticle(uri, body)

        case body: String => //If we get the cache response first, then we handle it and shut down.
          //The cache response will come back before the HTTP response so we never parse in this case.
          senderRef ! body
          context.stop(self)

        case ArticleBody(uri, body) => //If we get the parsed article back, then we've just parsed it
          cacheActor ! SetRequest(uri, body) //Cache it as we just parsed it
          senderRef ! body
          context.stop(self)

        case t => //We can get a cache miss
          println("ignoring msg: " + t.getClass)
      }
    }))
  }

} 

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

import java.util.concurrent.TimeoutException

import akka.actor.Status.Failure
import akka.actor.{Actor, ActorRef, Props}
import akka.util.Timeout
import com.dbaktor.messages._

class TellDemoArticleParser(cacheActorPath: String,
                            httpClientActorPath: String,
                            acticleParserActorPath: String,
                            implicit val timeout: Timeout
                           ) extends Actor {
  val cacheActor = context.actorSelection(cacheActorPath)
  val httpClientActor = context.actorSelection(httpClientActorPath)
  val articleParserActor = context.actorSelection(acticleParserActorPath)

  implicit val ec = context.dispatcher

  
  private def buildExtraActor(senderRef: ActorRef, uri: String): ActorRef = {
    return context.actorOf(Props(new Actor{
      override def receive = {
        case "timeout" => //if we get timeout, then fail
          senderRef ! Failure(new TimeoutException("timeout!"))
          context.stop(self)

        case HttpResponse(body) => //If we get the http response first, we pass it to be parsed.
          articleParserActor ! ParseHtmlArticle(uri, body)

        case body: String => //If we get the cache response first, then we handle it and shut down.
          //The cache response will come back before the HTTP response so we never parse in this case.
          senderRef ! body
          context.stop(self)

        case ArticleBody(uri, body) => //If we get the parsed article back, then we've just parsed it
          cacheActor ! SetRequest(uri, body) //Cache it as we just parsed it
          senderRef ! body
          context.stop(self)

        case t => //We can get a cache miss
          println("ignoring msg: " + t.getClass)
      }
    }))
  }

} 

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

import java.util.concurrent.TimeoutException
import javax.inject.Inject

import config.WeatherSerivceUrls
import models.{FiveDayReportRoot, Root}
import play.api.libs.ws.WSClient

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future


sealed trait MetOfficeResponse
case class AllLocationsSuccessResponse(locations: Root) extends MetOfficeResponse
case class FiveDayForecastSuccessResponse(report: FiveDayReportRoot) extends MetOfficeResponse
case class NotFoundResponse(code: Int) extends MetOfficeResponse
case class ExampleTimeOut(error: TimeoutException) extends MetOfficeResponse

class MetOfficeService @Inject() (http: WSClient) {

  def getLocations: Future[MetOfficeResponse] = {

//  Call 1: Making a call to ws.url so need to mock this call in testing -> Returns a WSRequest
    val r = http.url(WeatherSerivceUrls.listOfLocationsUrl)

//  Call 2: Making a call to .get -> Returns a WSResponse
    r.get map {
      response =>
        response.status match {
          case 200 => AllLocationsSuccessResponse(response.json.as[Root])
          case 404 => NotFoundResponse(response.status)
        }
    } recover {
      case t: TimeoutException => ExampleTimeOut(t)
    }
  }

  def getFiveDayForecast(id: String): Future[MetOfficeResponse] = {

    http.url(WeatherSerivceUrls.fiveDayForecast(id)).get map {
      fiveDayForecast =>
        fiveDayForecast.status match {
          case 200 => FiveDayForecastSuccessResponse(fiveDayForecast.json.as[FiveDayReportRoot])
          case 404 => NotFoundResponse(fiveDayForecast.status)
        }
    } recover {
      case t: TimeoutException => ExampleTimeOut(t)
    }
  }


} 

//设置package包名称以及导入依赖的类
package com.github.cupenya.auth.service.health

import java.util.concurrent.TimeoutException

import akka.actor.ActorSystem
import akka.http.scaladsl.model.DateTime

import scala.concurrent.duration._
import scala.concurrent.{ ExecutionContext, Future }
import scala.language.postfixOps

trait HealthCheckService {
  def checks: List[HealthCheck]

  def runChecks()(implicit ec: ExecutionContext, actorSystem: ActorSystem): Future[List[HealthCheckResult]] =
    Future.sequence(checks.map(_.checkWithRecovery))

  implicit class RichHealthCheck(check: HealthCheck) {
    def checkWithRecovery()(implicit ec: ExecutionContext, actorSystem: ActorSystem): Future[HealthCheckResult] =
      check
        .runCheck()
        .timeoutAfter(5 seconds)
        .recover {
          case e => HealthCheckResult(
            name = check.name,
            status = HealthCheckStatus.Critical,
            timestamp = DateTime.now.clicks,
            message = Some(s"Error while executing health check: ${e.getMessage}")
          )
        }
  }

  implicit class FutureExtensions[T](f: Future[T]) {
    import akka.pattern._

    def timeoutAfter(d: FiniteDuration)(implicit ec: ExecutionContext, sys: ActorSystem): Future[T] = {
      val eventualTimeout = after(d, sys.scheduler)(Future.failed(new TimeoutException(s"Timed out after ${d.toMillis} ms")))
      Future firstCompletedOf (f :: eventualTimeout :: Nil)
    }

  }
} 

//设置package包名称以及导入依赖的类
package com.github.cupenya.service.discovery.health

import java.util.concurrent.TimeoutException

import akka.actor.ActorSystem
import akka.http.scaladsl.model.DateTime

import scala.concurrent.{ ExecutionContext, Future }
import scala.concurrent.duration._
import scala.language.postfixOps

trait HealthCheckService {
  def checks: List[HealthCheck]

  def runChecks()(implicit ec: ExecutionContext, actorSystem: ActorSystem): Future[List[HealthCheckResult]] =
    Future.sequence(checks.map(_.checkWithRecovery))

  implicit class RichHealthCheck(check: HealthCheck) {
    def checkWithRecovery()(implicit ec: ExecutionContext, actorSystem: ActorSystem): Future[HealthCheckResult] =
      check
        .runCheck()
        .timeoutAfter(5 seconds)
        .recover {
          case e => HealthCheckResult(
            name = check.name,
            status = HealthCheckStatus.Critical,
            timestamp = DateTime.now.clicks,
            message = Some(s"Error while executing health check: ${e.getMessage}")
          )
        }
  }

  implicit class FutureExtensions[T](f: Future[T]) {
    import akka.pattern._

    def timeoutAfter(d: FiniteDuration)(implicit ec: ExecutionContext, sys: ActorSystem): Future[T] = {
      val eventualTimeout = after(d, sys.scheduler)(Future.failed(new TimeoutException(s"Timed out after ${d.toMillis} ms")))
      Future firstCompletedOf (f :: eventualTimeout :: Nil)
    }

  }
} 

//设置package包名称以及导入依赖的类
package com.bwsw.tstreamstransactionserver.netty.client


import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException

import com.bwsw.tstreamstransactionserver.exception.Throwable.RequestTimeoutException

import scala.concurrent.{ExecutionContext, Future => ScalaFuture, Promise => ScalaPromise}
import scala.concurrent.duration.Duration
import io.netty.util.{HashedWheelTimer, Timeout}
import org.slf4j.LoggerFactory


object TimeoutScheduler{
  private val logger = LoggerFactory.getLogger(this.getClass)
  private val timer = new HashedWheelTimer(10, TimeUnit.MILLISECONDS)
  def scheduleTimeout(promise:ScalaPromise[_], after:Duration, reqId: Long): Timeout = {
    timer.newTimeout((timeout: Timeout) => {
      val requestTimeoutException = new RequestTimeoutException(reqId, after.toMillis)
      val isExpired = promise.tryFailure(requestTimeoutException)
      if (isExpired && logger.isDebugEnabled) logger.debug(requestTimeoutException.getMessage)
    }, after.toNanos, TimeUnit.NANOSECONDS)
  }

  def withTimeout[T](fut:ScalaFuture[T])(implicit ec:ExecutionContext, after:Duration, reqId: Long): ScalaFuture[T] = {
    val prom = ScalaPromise[T]()
    val timeout = TimeoutScheduler.scheduleTimeout(prom, after, reqId)
    val combinedFut = ScalaFuture.firstCompletedOf(collection.immutable.Seq(fut, prom.future))
    fut onComplete (_ => timeout.cancel())
    combinedFut
  }
} 

//设置package包名称以及导入依赖的类
import java.nio.file.Path
import java.util.concurrent.{TimeUnit, TimeoutException}

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._
import scala.concurrent.{Await, Future}
import scala.io.Source


  def analyse(root: Path, pmdRoot: Path): Future[String] = {
    Future {
      val builder = new ProcessBuilder()
      builder.directory(root.toFile)
      // Set pmd website
      val command = Seq(pmdRoot.resolve("bin").resolve("run.sh").toString, "pmd",
        "-d", ".",
        "-language", "java",
        "-version", "1.8",
        "-f", "html",
        "-rulesets", rulesets.map(rule => s"${pmdRoot.toString}/rulesets/java/$rule.xml").mkString(","))
      builder.command(command: _*)

      val process = builder.start()
      val output = process.getInputStream
      val outputString = Future {
        val out = Source.fromInputStream(output)
          // Remove unnecessary project tmp path
          .getLines().map(line => line.replace(root.toString, ""))
          // Make one long string
          .mkString

        // Strip away surrounding <html>, <head> and <h*> tags
        val stripped = out.substring(119, out.length - 15)

        // Seriously, it's easier to parse the links than read the documentation on how to set the pmd website -_-
        pmdWebsiteRegex.replaceAllIn(stripped, pmdWebsite)
      }

      if (process.waitFor(2, TimeUnit.MINUTES)) {
        Await.result(outputString, 10 seconds)
      } else {
        process.destroyForcibly()
        throw new TimeoutException("Code analysis did not complete within 2 minutes. Exit code " + process.exitValue())
      }
    }
  }

} 

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

import java.util.concurrent.TimeoutException
import javax.inject.{ Inject, Named, Singleton }

import akka.actor.{ Actor, ActorLogging, ActorRef }
import play.api.libs.concurrent.InjectedActorSupport

import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration._

@Singleton
class TimeoutSettings() {
  val UnexpectedErrorRetryTimeout = 10.seconds
  val TimeoutErrorRetryTimeout = 0.seconds
}

object EtcdMainActor {
  final val name = "etcd-main"
}

class EtcdMainActor @Inject() (
    configListener: ConfigListener,
    @Named(EtcdWatcherActor.name) watcherActor: ActorRef,
    timeoutSettings: TimeoutSettings
) extends Actor with InjectedActorSupport with ActorLogging {

  override def preStart(): Unit = {
    watcherActor ! RetrieveKeys
  }

  override def receive = {
    case UpdateKeys(keys) =>
      log.debug(s"Received updated keys: $keys")
      configListener.keysUpdated(keys)
      watcherActor ! WatchKeys

    case HandleFailure(exception) =>
      val retryTimeout = exception match {
        case exc: TimeoutException =>
          log.info(s"Disconnected by timeout [${exc.getMessage}]")
          timeoutSettings.TimeoutErrorRetryTimeout
        case err =>
          log.error(err, s"Unexpected error while requesting key [${err.getMessage}]")
          timeoutSettings.UnexpectedErrorRetryTimeout
      }
      context.system.scheduler.scheduleOnce(retryTimeout, watcherActor, RetrieveKeys)

    case _ => log.warning("Unknown message received")
  }
} 

//设置package包名称以及导入依赖的类
import java.io.IOException
import java.net.ConnectException
import java.util.concurrent.TimeoutException

import akka.actor.ActorSystem
import akka.stream.{ActorMaterializer, ActorMaterializerSettings, Supervision}
import com.codahale.metrics.jvm.{ThreadStatesGaugeSet, MemoryUsageGaugeSet, GarbageCollectorMetricSet}

import scala.util.control.NonFatal

object RemoraApp extends App with nl.grons.metrics.scala.DefaultInstrumented {

  private val actorSystemName: String = "remora"
  implicit val actorSystem = ActorSystem(actorSystemName)

  metricRegistry.registerAll(new GarbageCollectorMetricSet)
  metricRegistry.registerAll(new MemoryUsageGaugeSet)
  metricRegistry.registerAll(new ThreadStatesGaugeSet)

  lazy val decider: Supervision.Decider = {
    case _: IOException | _: ConnectException | _: TimeoutException => Supervision.Restart
    case NonFatal(err: Throwable) =>
      actorSystem.log.error(err, "Unhandled Exception in Stream: {}", err.getMessage)
      Supervision.Stop
  }

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

  implicit val executionContext = actorSystem.dispatchers.lookup("kafka-consumer-dispatcher")
  val kafkaSettings = KafkaSettings(actorSystem.settings.config)
  val consumer = new RemoraKafkaConsumerGroupService(kafkaSettings)
  val kafkaClientActor = actorSystem.actorOf(KafkaClientActor.props(consumer), name = "kafka-client-actor")

  Api(kafkaClientActor).start()

} 

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

import java.util.concurrent.TimeoutException

import akka.actor.Status.Failure
import akka.actor.{Actor, ActorRef, Props}
import akka.util.Timeout

import com.akkademy.messages.{GetRequest, SetRequest}

class TellDemoArticleParser(cacheActorPath: String,
                            httpClientActorPath: String,
                            articleParserActorPath: String,
                            implicit val timeout: Timeout) extends Actor {
  val cacheActor = context.actorSelection(cacheActorPath)
  val httpClientActor = context.actorSelection(httpClientActorPath)
  val articleParserActor = context.actorSelection(articleParserActorPath)

  implicit val ec = context.dispatcher

  
  private def buildExtraActor(senderRef: ActorRef, uri: String): ActorRef = {
    context.actorOf(Props(new Actor{
      override def receive: Receive = {
        case body: String => // This executes when a cached result is found.
          senderRef ! body
          context.stop(self)
        // the articleParserActor will reply with an ArticleBody which is the parsing results.
        case HttpResponse(body) => articleParserActor ! ParseHtmlArticle(uri, body)
        case ArticleBody(uri, body) => // This executes when a raw parse of the html is done.
          cacheActor ! SetRequest(uri, body)
          senderRef ! body
          context.stop(self)
        case "timeout" =>
          senderRef ! Failure(new TimeoutException("timeout!"))
          context.stop(self)
        case t => println("ignoring msg: " + t.getClass)
      }
    }))
  }
} 

//设置package包名称以及导入依赖的类
package io.plasmap.query.engine

import java.util.concurrent.TimeoutException

import _root_.io.plasmap.geo.data.OsmStorageService
import _root_.io.plasmap.geo.mappings.{IndexingService, MappingService}
import akka.NotUsed
import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.HttpMethods._
import akka.http.scaladsl.model._
import akka.http.scaladsl.model.ws.{Message, UpgradeToWebSocket}
import akka.stream._
import akka.stream.scaladsl._
import com.typesafe.scalalogging.Logger
import org.slf4j.LoggerFactory

import scala.concurrent.Await
import scala.concurrent.duration._


object WSServer extends App {

  // required actorsystem and flow materializer
  implicit val system = ActorSystem("websockets")
  implicit val fm = ActorMaterializer()

  val log = Logger(LoggerFactory.getLogger(WSServer.getClass.getName))

  implicit val ec = scala.concurrent.ExecutionContext.Implicits.global

  val storage = OsmStorageService()
  val index = IndexingService()
  val mapping = MappingService()

  log.info(s"Initialised $storage, $index, $mapping")

  val binding = Http().bindAndHandleSync(requestHandler(), interface = "0.0.0.0", port = 9000)

  def requestHandler(flow : Flow[Message, Message, NotUsed] = Flows(fm,ec)): HttpRequest ? HttpResponse = {
    case req @ HttpRequest(GET, Uri.Path("/api/websocket"), _, _, _) ?
      req.header[UpgradeToWebSocket] match {
        case Some(upgrade) ? upgrade.handleMessages(flow)
        case None          ? HttpResponse(400, entity = "Not a valid websocket request!")
      }
    case _: HttpRequest ? HttpResponse(404, entity = "Unknown resource!")
  }

  // binding is a future, we assume it's ready within a second or timeout
  try {
    Await.result(binding, 1 second)
  } catch {
    case exc: TimeoutException =>
      println("Server took too long to startup, shutting down")
      system.terminate()
  }

} 

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

import java.util.concurrent.TimeoutException

import akka.actor.Status.Failure
import akka.actor.{Actor, ActorRef, Props}
import akka.util.Timeout
import com.akkademy.messages.{GetRequest, SetRequest}

class TellDemoArticleParser(cacheActorPath: String,
                            httpClientActorPath: String,
                            articleParserActorPath: String,
                            implicit val timeout: Timeout
                           ) extends Actor {
  val cacheActor = context.actorSelection(cacheActorPath)
  val httpClientActor = context.actorSelection(httpClientActorPath)
  val articleParserActor = context.actorSelection(articleParserActorPath)

  implicit val ec = context.dispatcher

  override def receive: Receive = {
    case [email protected](uri) =>
      val extraActor = buildExtraActor(sender(), uri)

      cacheActor.tell(GetRequest(uri), extraActor)
      httpClientActor.tell("test", extraActor)

      context.system.scheduler.scheduleOnce(timeout.duration, extraActor, "timeout")
  }

  private def buildExtraActor(senderRef: ActorRef, uri: String): ActorRef = {
    return context.actorOf(Props(new Actor {
      override def receive = {
        case "timeout" =>
          senderRef ! Failure(new TimeoutException("timeout!"))
          context.stop(self)

        case HttpResponse(body) =>
          articleParserActor ! ParseHtmlArticle(uri, body)

        case body: String =>
          senderRef ! body
          context.stop(self)

        case ArticleBody(uri, body) =>
          cacheActor ! SetRequest(uri, body)
          senderRef ! body
          context.stop(self)

        case t =>
          println("ignoring msg: " + t.getClass)
      }
    }))
  }

}