//设置package包名称以及导入依赖的类
package im.actor.server.api.rpc

import akka.actor._
import cats.data.Xor
import im.actor.api.rpc.collections.ApiRawValue
import im.actor.api.rpc.FutureResultRpc
import im.actor.api.rpc.raw.RawApiService
import im.actor.api.rpc.{ AuthorizedClientData, ClientData, CommonRpcErrors, RpcError }

import scala.collection.concurrent.TrieMap
import scala.concurrent.Future

sealed trait RawApiExtension extends Extension

private[rpc] final class RawApiExtensionImpl(system: ExtendedActorSystem) extends RawApiExtension {
  import FutureResultRpc._
  import system.dispatcher

  private val services = TrieMap.empty[String, RawApiService]

  def register(name: String, clazz: Class[_ <: RawApiService]): Unit = {
    val service = system.dynamicAccess.createInstanceFor[RawApiService](clazz, List(classOf[ActorSystem] ? system)).get
    register(name, service)
  }

  def register(name: String, service: RawApiService): Unit = services.putIfAbsent(name, service)

  def register(serviceSeq: Seq[(String, RawApiService)]): Unit = services ++= serviceSeq

  def handle(service: String, method: String, params: Option[ApiRawValue], clientData: ClientData): Future[RpcError Xor ApiRawValue] =
    (for {
      serviceHandler ? fromOption(CommonRpcErrors.UnsupportedRequest)(services.get(service))
      response ? fromOption(CommonRpcErrors.UnsupportedRequest)(serviceHandler.handleRequests(clientData)(params).lift(method))
      result ? fromFutureXor(response)
    } yield result).value
}

object RawApiExtension extends ExtensionId[RawApiExtensionImpl] with ExtensionIdProvider {
  override def createExtension(system: ExtendedActorSystem) = new RawApiExtensionImpl(system)

  override def lookup(): ExtensionId[_ <: Extension] = RawApiExtension
} 

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

import java.util.concurrent.atomic.AtomicLong
import scala.collection.concurrent.TrieMap
import scala.concurrent.Future


case class Pet_Item(id: Long, name: String, price: Double)

trait Pet_Franchise {
  def list(): Seq[Pet_Item]
  def create(name: String, price: Double): Option[Pet_Item]
  def details(id: Long): Option[Pet_Item]
  def update(id: Long, name: String, price: Double): Option[Pet_Item]
  def delete(id: Long): Boolean
}

object Pet_Shop extends Pet_Franchise {

  private val items = TrieMap.empty[Long, Pet_Item]
  private val seq = new AtomicLong

  def list(): Seq[Pet_Item] = items.values.to[Seq]

  def update(id: Long, name: String, price: Double): Option[Pet_Item]   = {
    val item = Pet_Item(id, name, price)
    items.replace(id, item)
    Some(item)
  }

  def details(id: Long): Option[Pet_Item] = items.get(id)

  def delete(id: Long): Boolean = items.remove(id).isDefined

  def create(name: String, price: Double): Option[Pet_Item] = {
    val id = seq.incrementAndGet()
    val item = Pet_Item(id, name, price)
    items.put(id, item)
    Some(item)
  }
} 

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

import scala.collection.concurrent.TrieMap
import scala.collection.generic.{ CanBuildFrom, GenericSetTemplate, MutableSetFactory }
import scala.collection.mutable

final class ConcurrentSet[A](elems: A*)
    extends mutable.Set[A]
    with GenericSetTemplate[A, ConcurrentSet]
    with mutable.SetLike[A, ConcurrentSet[A]]
    with mutable.FlatHashTable[A]
    with Serializable {
  import ConcurrentSet._

  private[this] val underlying = TrieMap[A, AnyRef](elems.map(_ -> Dummy): _*)

  override def +=(elem: A): this.type = {
    underlying.putIfAbsent(elem, Dummy)
    this
  }

  override def -=(elem: A): this.type = {
    underlying.remove(elem)
    this
  }

  override def contains(elem: A): Boolean = underlying.contains(elem)
  override def iterator: Iterator[A] = underlying.keysIterator
  override def companion: MutableSetFactory[ConcurrentSet] = ConcurrentSet
}

object ConcurrentSet extends MutableSetFactory[ConcurrentSet] {
  private[ConcurrentSet] val Dummy = new AnyRef

  override def apply[A](elems: A*): ConcurrentSet[A] = new ConcurrentSet[A](elems: _*)

  override def empty[A]: ConcurrentSet[A] = new ConcurrentSet[A]

  override def newBuilder[A]: mutable.Builder[A, ConcurrentSet[A]] =
    new mutable.SetBuilder[A, ConcurrentSet[A]](new ConcurrentSet[A]())

  implicit def canBuildFrom[A]: CanBuildFrom[Coll, A, ConcurrentSet[A]] =
    setCanBuildFrom[A]

  override def setCanBuildFrom[A]: CanBuildFrom[ConcurrentSet[_], A, ConcurrentSet[A]] =
    new CanBuildFrom[ConcurrentSet[_], A, ConcurrentSet[A]] {
      override def apply(from: ConcurrentSet[_]): mutable.Builder[A, ConcurrentSet[A]] = newBuilder[A]
      override def apply(): mutable.Builder[A, ConcurrentSet[A]] = newBuilder[A]
    }

} 

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

import base.{ LazyLoggerSupport, MeterSupport }
import mapdomain.graph._
import mapdomain.repository.street.{ StreetRepositorySupport, StreetVertexRepository }
import mapdomain.utils.GraphUtils

import scala.collection.Map
import scala.collection.concurrent.TrieMap

trait StreetGraphContainer extends GeoGraphContainer[StreetEdge, StreetVertex[StreetEdge]] {
  def findNearestStreets(coordinate: Coordinate, radius: Double): List[StreetEdge]
  def vertices: List[StreetVertex[StreetEdge]]
}

case class LazyStreetGraphContainer() extends StreetGraphContainer with StreetRepositorySupport {

  protected val vertexById = new TrieMap[Long, StreetVertex[StreetEdge]]()
  protected val totalVertices: Long = streetVertexRepository.totalVertices

  override def vertices: List[StreetVertex[StreetEdge]] = {
    if (totalVertices != vertexById.size) {
      vertexById.keys
    }
    StreetVertexRepository.findAll
  }

  override def findNearestStreets(coordinate: Coordinate, radius: Double): List[StreetEdge] = streetEdgeRepository.findNearestStreets(coordinate, radius)

  override def findNearest(coordinate: Coordinate): Option[StreetVertex[StreetEdge]] = streetVertexRepository.findNearest(coordinate)

  
  def purgeStreets: UnsavedStreetGraphContainer = withTimeLogging({
    logger.info(s"Purge the unsaved street graph in order to get a connected graph")
    GraphUtils.getConnectedComponent[StreetEdgeUnsaved, UnsavedStreetVertex, UnsavedStreetGraphContainer](this, UnsavedStreetGraphContainer.apply)
  }, (time: Long) ? logger.info(s"Street graph was purged in $time ms."))

} 

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

import mapdomain.sidewalk.{ SidewalkVertex, StreetCrossingEdge }
import scala.collection.concurrent.TrieMap

case class StreetCrossingBuilderManager() {

  val _builders: TrieMap[(SidewalkVertexBuilder, SidewalkVertexBuilder), StreetCrossingBuilder] = TrieMap.empty

  def create(from: SidewalkVertexBuilder, to: SidewalkVertexBuilder): StreetCrossingBuilder = {
    _builders.getOrElseUpdate((from, to), StreetCrossingBuilder(from, to))
  }

  def builders: List[StreetCrossingBuilder] = _builders.values.toList
}

case class StreetCrossingBuilder(from: SidewalkVertexBuilder, to: SidewalkVertexBuilder) {

  def crossKey(fromId: Long, toId: Long): String = {
    val idPart = if (fromId > toId) s"$toId-$fromId" else s"$fromId-$toId"
    s"$idPart-cross"
  }

  def build(implicit idGenerator: SidewalkVertexIDGenerator): (StreetCrossingEdge, SidewalkVertex, SidewalkVertex) = {
    val vertexFrom: SidewalkVertex = from.build
    val vertexTo: SidewalkVertex = to.build
    (StreetCrossingEdge(vertexFrom.id, vertexTo.id, crossKey(vertexFrom.id, vertexTo.id)), vertexFrom, vertexTo)
  }
} 

//设置package包名称以及导入依赖的类
package au.id.tmm.senatedb.api.integrationtest

import java.time.Instant
import java.time.temporal.ChronoUnit

import play.api.cache.SyncCacheApi

import scala.collection.concurrent.TrieMap
import scala.concurrent.duration.Duration


class SimpleCacheApi extends SyncCacheApi {

  private val map = new TrieMap[String, (Instant, Any)]()

  override def set(key: String, value: Any, expiration: Duration): Unit = {
    val expiryTime = {
      if (expiration == Duration.Inf) {
        Instant.MAX
      } else {
        Instant.now().plus(expiration.toMillis, ChronoUnit.MILLIS)
      }
    }

    map.update(key, (expiryTime, value))
  }

  override def remove(key: String): Unit = {
    map.remove(key)
  }

  override def getOrElseUpdate[A](key: String, expiration: Duration)(orElse: => A)(implicit evidence$1: ClassManifest[A]): A = {
    val existingValue = get(key)

    if (existingValue.isEmpty) {
      set(key, orElse, expiration)
      orElse
    } else {
      existingValue.get
    }
  }

  override def get[T](key: String)(implicit evidence$2: ClassManifest[T]): Option[T] = {
    map.get(key)
      .flatMap { case (expiryTime, value) =>

        if (expiryTime.isAfter(Instant.now())) {
          remove(key)
          None
        } else {
          Some(value)
        }
      }
      .map(_.asInstanceOf[T])
  }
} 

//设置package包名称以及导入依赖的类
package im.actor.server.sequence

import akka.actor.ActorSystem
import com.google.protobuf.wrappers.{ Int32Value, StringValue }
import com.relayrides.pushy.apns.PushNotificationResponse
import com.relayrides.pushy.apns.util.{ SimpleApnsPushNotification, TokenUtil }
import im.actor.server.model.push.ApplePushCredentials
import io.netty.util.concurrent.{ Future ? NFuture }
import scodec.bits.BitVector

import scala.collection.concurrent.TrieMap

trait APNSSend {

  private val listeners = TrieMap.empty[String, PushFutureListener]

  protected def sendNotification(payload: String, creds: ApplePushCredentials, userId: Int)(implicit client: ApplePushExtension#Client, system: ActorSystem): NFuture[PushNotificationResponse[SimpleApnsPushNotification]] = {
    // when topic is null, it will be taken from APNs certificate
    // http://relayrides.github.io/pushy/apidocs/0.6/com/relayrides/pushy/apns/ApnsPushNotification.html#getTopic--
    val token = BitVector(creds.token.toByteArray).toHex
    val topic: String = (creds.apnsKey, creds.bundleId) match {
      case (_, Some(bundleId)) ? bundleId.value
      case (Some(key), _)      ? ApplePushExtension(system).apnsBundleId.get(key.value).orNull
      case _ ?
        system.log.warning("Wrong creds format on sending notification. Creds: {}", creds)
        null
    }
    system.log.debug(s"Sending APNS, token: {}, key: {}, isVoip: {}, topic: {}, payload: $payload", token, creds.apnsKey, creds.isVoip, topic)
    val notification = new SimpleApnsPushNotification(TokenUtil.sanitizeTokenString(token), topic, payload)
    val listener = listeners.getOrElseUpdate(token, new PushFutureListener(userId, creds, extractCredsId(creds))(system))
    client.sendNotification(notification).addListener(listener)
  }

  protected def extractCredsId(creds: ApplePushCredentials): String = (creds.apnsKey, creds.bundleId) match {
    case (Some(Int32Value(key)), _)       ? key.toString
    case (_, Some(StringValue(bundleId))) ? bundleId
    case _                                ? throw new RuntimeException("Wrong credentials format")
  }

} 

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

import shared.models.{Product, ProductQuantity}

import scala.collection.concurrent.TrieMap


sealed trait FakeDatabaseData {
  protected val products = TrieMap(
    Product(1, "iPhone") -> ProductQuantity(5),
    Product(2, "The Witcher") -> ProductQuantity(3),
    Product(3, "Computer") -> ProductQuantity(1),
    Product(4, "Keyboard") -> ProductQuantity(2)
  )
}

sealed trait DatabaseService {
  def checkProductAvailability(product: Product): Boolean
  def increaseQuantityOfProduct(product: Product, by: Int = 1): Unit
}

object DatabaseServiceImpl extends DatabaseService with FakeDatabaseData {
  def checkProductAvailability(product: Product): Boolean = {
    products.get(product) match {
      case Some(ProductQuantity(quantity)) if quantity > 0 =>
        products.update(product, ProductQuantity(quantity - 1))
        true
      case _ => false
    }
  }
  def increaseQuantityOfProduct(product: Product, by: Int = 1): Unit = {
    products.get(product) match {
      case Some(ProductQuantity(quantity)) =>
        products.update(product, ProductQuantity(quantity + by))
      case _ => throw new RuntimeException("Product not found!")
    }
  }
} 

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

import com.typesafe.config.ConfigFactory
import model.Tweet
import play.api.Logger

import scala.collection.concurrent.TrieMap


object TweetIndexedStorage {
  private val config = ConfigFactory.load()
  private val StopWords = config.getList("stop_words")
  private val indexedTweets = TrieMap.empty[String, Set[Tweet]]

  def addTweets(tweets: Set[Tweet]) =
    tweets.foreach(tweet => getTokens(tweet).foreach(token => addTweet(token, tweet)))

  def getSimilarInIndex(tweet:Tweet) = getTokens(tweet).flatten(token => indexedTweets.get(token))
    .toList
    .flatten
    .groupBy(identity).mapValues(_.size)
    .groupBy { case (key, value) => value }
    .mapValues(_.keys.toSet)

  private def getTokens(tweet:Tweet) =
    tweet.text.replaceAll("[.,[email protected]\"]", "")
    .split(" ")
    .filterNot(_.startsWith("https"))
    .filterNot(_.isEmpty)
    .filterNot(StopWords.contains(_))
    .map(_.trim.toLowerCase)
    .toSet

  def getSize:(Int, Int) = (TweetIndexedStorage.indexedTweets.keys.size,
    TweetIndexedStorage.indexedTweets.values.flatten.size)

  def addTweet(token: String, tweet: Tweet) = {
    Logger.info(s"Added tweet $tweet for token $token to Index ( keys - ${TweetIndexedStorage.indexedTweets.keys.size}," +
      s" values - ${TweetIndexedStorage.indexedTweets.values.size} )")

    indexedTweets.get(token) match {
      case Some(foundTweets) =>
        indexedTweets += (token -> (foundTweets + tweet))
      case None =>
        indexedTweets += (token -> Set(tweet))
    }
  }

  def clear() = indexedTweets.clear()
} 

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

import scala.collection.concurrent.TrieMap

trait CanComplexElement[T] {
  def upsert(existing: T, upsert: T): T
  def remove[A](existing: T, remove: A): T
  def isEmpty(existing: T): Boolean
}

// todo: this is too complicated, refactoring or documentation is needed
class ComplexTrieMap[K, V] {
  protected val map = new TrieMap[K, V]

  def get(key: K): Option[V] = map.get(key)

  def getOrElse(key: K, default: => V): V = map.getOrElse(key, default)

  def upsert(key: K, value: V)(implicit evidence: CanComplexElement[V]): Unit = {
    this.synchronized {
      map.putIfAbsent(key, value).map { existing =>
        val n = evidence.upsert(existing, value)
        val x = map.put(key, n)
        x
      }
    }
  }

  def remove[A](key: K, value: A)(implicit evidence: CanComplexElement[V]): Unit = {
    this.synchronized {
      map.get(key).map { existing =>
        val nv = evidence.remove(existing, value)
        if (evidence.isEmpty(nv))
          map.remove(key)
        else
          map.put(key, nv)
      }
    }
  }

  def foreach(code: ((K, V)) ? Unit): Unit = map.foreach(code)

  def map[O](code: ((K, V)) ? O): Iterable[O] = map.map(code)

  def clear(): Unit = map.clear()
} 

//设置package包名称以及导入依赖的类
package com.microsoft.spark.perf.core

import scala.collection.concurrent.TrieMap

import sun.misc.Unsafe

import org.apache.spark.TaskContext

private[core] object UnsafeUtils {
  private val unsafeMap = new TrieMap[Int, Unsafe]

  private[core] def getUnsafeInstance: Unsafe = {
    val partitionId = TaskContext.getPartitionId()
    unsafeMap.getOrElseUpdate(partitionId, {
      val unsafeField = classOf[sun.misc.Unsafe].getDeclaredField("theUnsafe")
      unsafeField.setAccessible(true)
      unsafeField.get(null).asInstanceOf[sun.misc.Unsafe]}
    )
  }
} 

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

import client.{ Client, DisposableResponse }
import com.twitter.finagle.{ Address, Name, Http => Netty3Http, ServiceFactory }
import com.twitter.finagle.http.{ Request => FinagleRequest, Response => FinagleResponse }
import com.twitter.finagle.netty4.Netty4Http
import scala.collection.concurrent.TrieMap
import scalaz.concurrent.Task
import scalaz.syntax.monad._
import FinagleConverters._

object FinagleClient {

  def getAddress(req: Request): Name.Bound = {
    val port = req.uri.port orElse {
      req.uri.scheme map { _.value.toLowerCase } flatMap {
        case "http"  => Some(80)
        case "https" => Some(443)
        case _       => None
      }
    } getOrElse 80
    Name.bound(Address(req.uri.host.get.value, port))
  }

  def apply(useNetty4: Boolean): Client = {
    val clients = new TrieMap[Name.Bound, ServiceFactory[FinagleRequest, FinagleResponse]]

    def getClient(name: Name.Bound): ServiceFactory[FinagleRequest, FinagleResponse] =
      clients.getOrElseUpdate(
        name,
        if (useNetty4)
          Netty4Http.newClient(name, "")
        else
          Netty3Http.newClient(name, "")
      )

    def service(req: Request): Task[DisposableResponse] = Task.suspend {
      val fResponse = for {
        service <- getClient(getAddress(req))()
        fResponse <- service(request.from(req))
        rep = response.to(fResponse)
      } yield DisposableResponse(rep, Task.suspend(service.close().asTask))
      fResponse.asTask
    }

    val shutdown: Task[Unit] = Task.suspend {
      Task.gatherUnordered(clients.values.toSeq map {
        _.close().asTask
      }).void
    }

    Client(Service.lift(service), Task.suspend(shutdown))
  }

} 

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

import io.prometheus.client._

import scala.collection.concurrent.TrieMap

// TODO: Make namespace optional
class Telemetry(registry: CollectorRegistry, namespace: String) {

  private val counters = TrieMap.empty[String, Counter]
  private val histograms = TrieMap.empty[String, Histogram]
  private val gauges = TrieMap.empty[String, Gauge]

  private def cacheKeyFor(name: String): String = s"${namespace}_$name"

  // TODO: Support injecting default labels

  def counter(name: String, help: String = "No help provided", labelNames: Seq[String] = Seq.empty): Counter = {
    counters.getOrElseUpdate(cacheKeyFor(name), {
      Counter.build()
        .namespace(namespace)
        .name(name)
        .help(help)
        .labelNames(labelNames: _*)
        .register(registry)
    })
  }

  def histogram(name: String, help: String = "No help provided", labelNames: Seq[String] = Seq.empty,
                buckets: Seq[Double] = Seq(0.1, 0.5, 1.0, 5.0)): Histogram = {
    histograms.getOrElseUpdate(cacheKeyFor(name), {
      Histogram.build()
        .namespace(namespace)
        .name(name)
        .help(help)
        .buckets(buckets: _*)
        .labelNames(labelNames: _*)
        .register(registry)
    })
  }

  def gauge(name: String, help: String = "No help provided", labelNames: Seq[String] = Seq.empty): Gauge = {
    gauges.getOrElseUpdate(cacheKeyFor(name), {
      Gauge.build()
        .namespace(namespace)
        .name(name)
        .help(help)
        .labelNames(labelNames: _*)
        .register(registry)
    })
  }
} 

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

import dynamite.Response.TableDescription

import scala.collection.concurrent.TrieMap
import scala.util.{ Success, Try }
import com.amazonaws.services.dynamodbv2.model.ResourceNotFoundException

class TableCache(describeTable: String => Try[TableDescription]) {
  private[this] val tableNameCache = TrieMap[String, TableDescription]()

  def clear() = tableNameCache.clear()

  def get(tableName: String): Try[Option[TableDescription]] = {
    tableNameCache.get(tableName).map { value =>
      Success(Some(value))
    }.getOrElse {
      describeTable(tableName).map { result =>
        tableNameCache.put(tableName, result)
        Option(result)
      }.recoverWith {
        case _: ResourceNotFoundException => Success(None)
      }
    }
  }
} 

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

import carads.backend.{Record, Storage}
import scala.util.{Success, Try}
import scala.collection.concurrent.{TrieMap}

class MemStorage extends Storage {
  val kv = TrieMap[Int, Record]()
  def createTable: Try[String] = { Success("ACTIVE") }

  def getAll(limit: Int): Try[List[Record]] = {
    Try { kv.values.toList }
  }
  def get(id: Int): Try[Record] = {
    Try { kv(id) }
  }
  def put(record: Record): Try[Unit] = {
    Try { kv.put(record.id, record) }
  }
  def modify(record: Record, attrs: Set[String]): Try[Unit] = Try {
    synchronized {
      var r = get(record.id).get
      for (attr <- attrs) attr match {
        case "title" => r = r.copy(title = record.title)
        case "fuel" => r = r.copy(fuel = record.fuel)
        case "price" => r = r.copy(price = record.price)
        case "new" => r = r.copy(`new` = record.`new`)
        case "mileage" => r = r.copy(mileage = record.mileage)
        case "registration" => r = r.copy(registration = record.registration)
      }
      put(r).get
    }
  }
  def delete(id: Int): Try[Unit] = {
    Try { kv.remove(id) }
  }
} 

//设置package包名称以及导入依赖的类
package eu.inn.metrics

import java.util.concurrent.atomic.AtomicReference

import com.codahale.metrics._

import scala.collection.concurrent.TrieMap

private [metrics] class MetricsTrackerImpl(registry: MetricRegistry) extends MetricsTracker {
  protected val gauges = TrieMap[String, ReplaceableGauge[_]]()

  override def counter(name: String): Counter = {
    registry.counter(name)
  }

  override def meter(name: String): Meter = {
    registry.meter(name)
  }

  override def histogram(name: String): Histogram = {
    registry.histogram(name)
  }

  override def timer(name: String): Timer = {
    registry.timer(name)
  }

  def gauge[T](name: String, gauge: Gauge[T]): Unit = {
    gauges.putIfAbsent(name, new ReplaceableGauge[T](gauge)) match {
      case Some(existing) ? existing.replaceUnderlying(gauge)
      case None ? registry.register(name, gauge)
    }
  }

  def remove(name: String): Unit = {
    gauges.remove(name)
    registry.remove(name)
  }

  def removeAll(): Unit = {
    gauges.clear()
    registry.removeMatching(MetricFilter.ALL)
  }
}

private [metrics] class ReplaceableGauge[T](initUnderlying: Gauge[T]) extends Gauge[T] {
  private val underlyingGauge = new AtomicReference[Gauge[T]](initUnderlying)

  override def getValue = {
    underlyingGauge.get().getValue
  }

  def replaceUnderlying[X](newUnderlyingGauge: Gauge[X]): Unit = {
    underlyingGauge.set(newUnderlyingGauge.asInstanceOf[Gauge[T]])
  }
} 

//设置package包名称以及导入依赖的类
package com.programmaticallyspeaking.ncd.chrome.net

import akka.actor.Actor.Receive
import akka.actor.{Actor, ActorRef, ActorRefFactory, Props, Terminated}
import com.programmaticallyspeaking.ncd.chrome.domains.{DefaultDomainFactory, DomainFactory}
import com.programmaticallyspeaking.ncd.ioc.Container

import scala.collection.concurrent.TrieMap

class CapturingDomainFactory(implicit container: Container, factory: ActorRefFactory) extends DomainFactory {
  private val defaultFactory = new DefaultDomainFactory(container)

  private var actorMustNotExist = false
  private val actors = TrieMap[String, ActorRef]()
  private val watcher = factory.actorOf(Props(new ActorWatcher))

  def actorByName(name: String): Option[ActorRef] = actors.get(name)

  def requireNoOldActor(): Unit = {
    actorMustNotExist = true
  }

  override def create(domain: String): ActorRef = {
    actors.get(domain) match {
      case Some(ar) if actorMustNotExist => throw new IllegalStateException("Found an old domain actor: " + ar)
      case _ => // noop
    }
    val actor = defaultFactory.create(domain)
    watcher ! actor
    actors(domain) = actor
    actor
  }

  class ActorWatcher extends Actor {
    override def receive: Receive = {
      case actorRef: ActorRef =>
        context.watch(actorRef)
      case Terminated(actorRef) =>
        val key = actors.find(_._2 == actorRef).map(_._1)
        key.foreach(actors.-=)
    }
  }
} 

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

import java.time.Instant
import java.time.temporal.ChronoUnit

import org.scalatest.{FlatSpec, Matchers}
import org.slf4j.LoggerFactory

import scala.collection.concurrent.TrieMap

class InMemoryStoreSpec extends FlatSpec with Matchers {
  val logger = LoggerFactory.getLogger(this.getClass)
  "Cleaner" should "remove expired entries" in {
    val cache = TrieMap.empty[(String, Long), Any]
    val cleaner = InMemoryStore.createCleaner(cache, 1, logger)

    cache += ("a", Instant.now.minus(2, ChronoUnit.DAYS).toEpochMilli) -> 1
    cache += ("b", Instant.now.minus(1, ChronoUnit.DAYS).toEpochMilli) -> 2
    cache += ("c", Instant.now.toEpochMilli) -> 3
    cleaner.run()
    cache.size shouldBe 1
    cache.head._1._1 shouldBe "c"
  }

} 

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

import eveapi.utils.Decoders._
import eveapi.errors.{EveApiError, EveApiStatusFailed}

import java.time.Clock
import scalaz.stream.async.mutable.Topic
import scala.concurrent.duration.Duration
import scala.collection.concurrent.TrieMap
import scalaz.stream.{Exchange, Process, Sink, async}
import scalaz.concurrent.Task
import scalaz._
import java.util.concurrent.ScheduledExecutorService

import org.http4s.{ Response, Uri }
import org.http4s.Uri.Authority
import org.http4s.util.CaseInsensitiveString

import models._
import oauth._
import shared._
import eveapi._
import eveapi.oauth._

case class EveServer(server: Uri.RegName)

case class TopicKey(fleetId: Long, userId: Long)

case class TopicHolder(pollInterval: Duration, oauth: OAuth2, server: EveServer)(implicit s: ScheduledExecutorService) {
  def fleetUri(id: Long, server: EveServer) = Uri(scheme = Some(CaseInsensitiveString("https")), authority = Some(Authority(host=server.server)), path = s"/fleets/$id/")

  private val topics = TrieMap[TopicKey, Topic[EveApiError \/ FleetState]]()

  def apply(user: User, fleetId: Long): Topic[EveApiError \/ FleetState] = {
    topics
      .retain({ case (id, topic) =>
        ! topic.subscribe.isHalt
      })
      .getOrElseUpdate(TopicKey(fleetId, user.id), {
        async.topic(
          ApiStream.fleetPollSource(fleetUri(fleetId, server), pollInterval, Execute.OAuthInterpreter)
            .translate[Task](ApiStream.fromApiStream(oauth, user.token))
            , true)
      }
    )
  }
} 

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

import scala.collection.concurrent.TrieMap
import scalaz.stream.{Process, Sink, channel}
import scalaz.concurrent.Task
import scalaz._

import doobie.imports._

import models._
import shared._

case class DBHolder(xa: Transactor[Task]) {
  private val dbs = TrieMap[Process[Task, FleetState], Process[Task, Unit]]()

  def apply(owner: User, p: Process[Task, FleetState]): Process[Task, Unit] = {
    dbs
      .retain({ case (id, process) =>
        ! process.isHalt
      })
      .getOrElseUpdate(p, {
        p.observeThrough(channel.lift(f => FleetHistory.insert(owner, f).transact(xa))).map(_._2)
      })
  }
}