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

import app.config.application.ApplicationProperty
import app.v1.Api
import com.twitter.finagle.Http
import com.twitter.logging.Logger
import com.twitter.util.{ Await, Duration, StorageUnit }

trait ServerModule {

  self: com.twitter.app.App with Api with ApplicationProperty =>

  private val log = Logger.get(getClass)

  premain {
    log.info("[Finch] server is starting ...")

    val server = Http.server
      .withLabel(applicationProperty.systemId)
      .withRequestTimeout(Duration.fromSeconds(applicationProperty.requestTimeoutInSeconds))
      .withMaxRequestSize(StorageUnit.fromMegabytes(applicationProperty.maxRequestSizeInMB))
      .configured(Http.Netty4Impl)
      .serve(s":${applicationProperty.applicationPort}", apiService)

    onExit {
      server.close()
    }

    Await.result(server)
  }

} 

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

import com.google.common.collect.ConcurrentHashMultiset
import com.twitter.util.{Duration, Future}


case class Throttler(maxConcurrentRequestsPerResource: Int = 1) {

  private val pending = ConcurrentHashMultiset.create[String]()

  def access[T](path: String,
                minTimeDelta: Option[Duration])
               (f: => Future[T]): Future[T] = {

    val prevAccessCount = pending.add(path, 1)
    if (prevAccessCount >= maxConcurrentRequestsPerResource) {
      pending.remove(path, 1)
      Future.exception(Throttled(path))
    } else {
      f.respond(_ => minTimeDelta match {
        case Some(minTime) => DefaultTimer.onceAfter(minTime)(pending.remove(path, 1))
        case None => pending.remove(path, 1)
      })
    }
  }
}

case class Throttled(id: String) extends RuntimeException(s"Throttled request '$id' (perhaps already pending?)") 

//设置package包名称以及导入依赖的类
package se.gigurra.dcs.remote.util

import java.nio.file.{Files, Paths, StandardOpenOption}
import com.twitter.finagle.util.DefaultTimer
import com.twitter.io.Charsets
import com.twitter.util.Duration

case class SnapshotLogger(outputFilePath: String,
                          dtFlush: Duration,
                          enabled: Boolean,
                          fGetSnapshot: () => String) {

  private val path = Paths.get(outputFilePath)

  if (enabled) {
    DefaultTimer.twitter.schedule(dtFlush) {
      Files.write(path, fGetSnapshot().getBytes(Charsets.Utf8), StandardOpenOption.WRITE, StandardOpenOption.CREATE)
    }
  }

} 

//设置package包名称以及导入依赖的类
package se.gigurra.dcs.remote.util

import java.nio.file.{Files, Paths, StandardOpenOption}
import java.util.concurrent.atomic.AtomicLong
import com.twitter.finagle.util.DefaultTimer
import com.twitter.io.Charsets
import com.twitter.util.Duration
import scala.util.Try
import scala.collection.JavaConversions._

case class DataUseLogger(outputFilePath: String, dtFlush: Duration, enabled: Boolean) {

  private val atomicDelta = new AtomicLong(0L)
  private val path = Paths.get(outputFilePath)

  if (enabled) {
    DefaultTimer.twitter.schedule(dtFlush) {
      val delta = atomicDelta.getAndSet(0)
      val prevBytes = Try(Files.readAllLines(path, Charsets.Utf8).head.toLong).getOrElse(0L)
      val newVal = ((prevBytes + delta).toString + "\n").getBytes(Charsets.Utf8)
      Files.write(path, newVal, StandardOpenOption.WRITE, StandardOpenOption.CREATE)
    }
  }

  def log(nBytes: Long): Unit = {
    atomicDelta.addAndGet(nBytes)
  }

} 

//设置package包名称以及导入依赖的类
package com.github.ikhoon.app.v1.fake

import javax.inject.{ Inject, Named }

import com.fasterxml.jackson.databind.JsonNode
import com.twitter.finatra.httpclient.{ HttpClient, RequestBuilder }
import com.twitter.util.{ Await, Duration, Future }
import com.typesafe.config.Config

class FakeService @Inject() (@Named("fake") httpClient: HttpClient, config: Config) {
  def withSleepAsync(sec: Int): Future[JsonNode] = {
    val url = s"/api/?sleep=$sec"
    httpClient.executeJson[JsonNode](RequestBuilder.get(url))
  }

  def withSleepSync(sec: Int): JsonNode = {
    val url = s"/api/?sleep=$sec"
    val jsonNode = httpClient.executeJson[JsonNode](RequestBuilder.get(url))
    Await.result(jsonNode, Duration.fromSeconds(100))
  }
} 

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

import com.twitter.util.{Duration, Return, Throw, Stopwatch, Future}
import com.twitter.finagle.{SimpleFilter, Service}
import com.twitter.logging.Logger

trait LogFormatter[-Req, Rep] {
  def format(request: Req, reply: Rep, replyTime: Duration): String

  def formatException(request: Req, throwable: Throwable, replyTime: Duration): String
}


trait LoggingFilter[Req, Rep] extends SimpleFilter[Req, Rep] {
  val log: Logger
  val formatter: LogFormatter[Req, Rep]

  def apply(request: Req, service: Service[Req, Rep]): Future[Rep] = {
    val elapsed = Stopwatch.start()
    val future = service(request)
    future respond {
      case Return(reply) =>
        log(elapsed(), request, reply)
      case Throw(throwable) =>
        logException(elapsed(), request, throwable)
    }
    future
  }

  protected def log(replyTime: Duration, request: Req, reply: Rep) {
    val line = formatter.format(request, reply, replyTime)
    log.info(line)
  }

  protected def logException(replyTime: Duration, request: Req, throwable: Throwable) {
    val line = formatter.formatException(request, throwable, replyTime)
    log.info(throwable, line)
  }

} 

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

import com.twitter.finagle.Service
import com.twitter.finagle.service.ExpiringService
import com.twitter.finagle.stats.StatsReceiver
import com.twitter.finagle.transport.Transport
import com.twitter.util.{Duration, Timer, Time, Closable}

object ExpiringServerDispatcher {
  def apply[Req, Rep, In, Out](
    maxIdleTime: Option[Duration],
    maxLifeTime: Option[Duration],
    timer: Timer,
    statsReceiver: StatsReceiver,
    newDispatcher: (Transport[In, Out], Service[Req, Rep]) => Closable
  ): (Transport[In, Out], Service[Req, Rep]) => Closable =
    (transport: Transport[In, Out], service: Service[Req, Rep]) =>
        new ExpiringService(service, maxIdleTime, maxLifeTime, timer, statsReceiver) {
          private[this] val dispatcher = newDispatcher(transport, this)
          protected def onExpire() { dispatcher.close(Time.now) }
        }
} 

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

import com.twitter.finagle._
import com.twitter.util.{Future, Duration, Timer}

object TimeoutFactory {
  val role = Stack.Role("ServiceTimeout")

  
class TimeoutFactory[Req, Rep](
    self: ServiceFactory[Req, Rep],
    timeout: Duration,
    exception: ServiceTimeoutException,
    timer: Timer)
  extends ServiceFactoryProxy[Req, Rep](self) {
  private[this] val failure = Future.exception(Failure.adapt(exception, Failure.Restartable))

  override def apply(conn: ClientConnection) = {
    val res = super.apply(conn)
    res.within(timer, timeout) rescue {
      case exc: java.util.concurrent.TimeoutException =>
        res.raise(exc)
        res onSuccess { _.close() }
        failure
    }
  }
} 

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

import com.twitter.finagle.Addr.Metadata
import com.twitter.finagle.client.AddrMetadataExtraction.AddrMetadata
import com.twitter.finagle.{Addr, ServiceFactory, Stack, Stackable}
import com.twitter.util.Duration
import java.util.concurrent.atomic.AtomicReference


  private[finagle] case class Compensation(howlong: Duration) {
    def mk(): (Compensation, Stack.Param[Compensation]) =
      (this, Compensation.param)
  }
  private[finagle] object Compensation {
    implicit val param =
      Stack.Param(Compensation(Duration.Zero))
  }


  def module[Req, Rep]: Stackable[ServiceFactory[Req, Rep]] =
    new Stack.Module[ServiceFactory[Req, Rep]] {
      val role = Role
      val description = "Sets a latency compensation to be added based on the destination address"
      val parameters = Seq(
        implicitly[Stack.Param[AddrMetadata]],
        implicitly[Stack.Param[Compensator]]
      )
      def make(prms: Stack.Params, next: Stack[ServiceFactory[Req, Rep]]) = {

        // If the caller has configured a Compensator, use that.
        // If there is no configured compensator, look for a default override.
        val Compensator(configured) = prms[Compensator]
        val compensator = DefaultOverride() match {
          case Some(v) if !prms.contains[Compensator] => v.compensator
          case _ => configured
        }

        val AddrMetadata(metadata) = prms[AddrMetadata]
        val compensation = compensator(metadata)
        val compensated = next.make(prms + Compensation(compensation))
        Stack.Leaf(this, compensated)
      }
    }

} 

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

import com.twitter.util.Duration


  object longHex {
    def unapply(s: String): Option[Long] = {
      try Some(java.lang.Long.parseLong(s.stripPrefix("0x"), 16)) catch {
        case _: NumberFormatException => None
      }
    }
  }

  object bool {
    def unapply(s: String): Option[Boolean] =
      s.toLowerCase match {
        case "1" | "true" => Some(true)
        case "0" | "false" => Some(false)
        case _ => None
       }
  }

  object duration {
    def unapply(s: String): Option[Duration] =
      try Some(Duration.parse(s)) catch {
        case _: NumberFormatException => None
      }
  }

} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.netty3.channel

import com.twitter.finagle.WriteTimedOutException
import com.twitter.util.{Time, Duration, Timer}
import org.jboss.netty.channel._


private[finagle] class WriteCompletionTimeoutHandler(timer: Timer, timeout: Duration)
  extends SimpleChannelDownstreamHandler
{
  override def writeRequested(ctx: ChannelHandlerContext, e: MessageEvent) {
    val task = timer.schedule(Time.now + timeout) {
      val channel = ctx.getChannel
      Channels.fireExceptionCaught(
        channel, new WriteTimedOutException(if (channel != null) channel.getRemoteAddress else null))
    }
    e.getFuture.addListener(new ChannelFutureListener {
      override def operationComplete(f: ChannelFuture): Unit =
        if (!f.isCancelled) { // on success or failure
          task.cancel()
        }
    })

    super.writeRequested(ctx, e)
  }
} 

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

import com.twitter.finagle._
import com.twitter.finagle.context.Deadline
import com.twitter.finagle.stats.StatsReceiver
import com.twitter.util.{Duration, Future, Time}

object DeadlineStatsFilter {

  val role = new Stack.Role("DeadlineStatsFilter")

  
private[finagle] class DeadlineStatsFilter[Req, Rep](statsReceiver: StatsReceiver)
    extends SimpleFilter[Req, Rep] {

  private[this] val exceededStat = statsReceiver.counter("exceeded")
  private[this] val transitTimeStat = statsReceiver.stat("transit_latency_ms")
  private[this] val expiredTimeStat = statsReceiver.stat("expired_ms")

  def apply(request: Req, service: Service[Req, Rep]): Future[Rep] = {
    Deadline.current match {
      case Some(deadline) =>
        val now = Time.now
        if (deadline.deadline < now) {
          val exceeded = now - deadline.deadline
          expiredTimeStat.add(exceeded.inMilliseconds)
        }
        transitTimeStat.add((now - deadline.timestamp).max(Duration.Zero).inMilliseconds)

        if (deadline.expired)
          exceededStat.incr()

      case None =>
    }
    service(request)
  }
} 

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

import com.twitter.util.{Time, Duration, Return}
import org.junit.runner.RunWith
import org.scalacheck.Gen
import org.scalatest.FunSuite
import org.scalatest.junit.{AssertionsForJUnit, JUnitRunner}
import org.scalatest.prop.GeneratorDrivenPropertyChecks

@RunWith(classOf[JUnitRunner])
class DeadlineTest 
  extends FunSuite 
  with AssertionsForJUnit
  with GeneratorDrivenPropertyChecks {

  val time = for (t <- Gen.choose(0L, Long.MaxValue)) yield Time.fromNanoseconds(t)
  val dur = for (d <- Gen.choose(0L, Long.MaxValue)) yield Duration.fromNanoseconds(d)
  val deadline = for (t <- time; d <- dur) yield Deadline(t, t + d)
  val deadlineWithoutTop  = deadline.filter(_.deadline != Time.Top)

  test("Deadline marshalling") {
    // won't pass Time.Top as deadline for marshalling
    forAll(deadlineWithoutTop) { d =>
      assert(Deadline.tryUnmarshal(Deadline.marshal(d)) == Return(d))
    }
  }

  test("Deadline.combined") {
    forAll(deadline, deadline) { (d1, d2) =>
      assert(Deadline.combined(d1, d2).timestamp == (d1.timestamp max d2.timestamp))
      assert(Deadline.combined(d1, d2).deadline == (d1.deadline min d2.deadline))
      assert(Deadline.combined(d1, d2) == Deadline.combined(d2, d1))
    }
  }
} 

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

import com.twitter.finagle.thrift.{ThriftClientRequest, ClientId}
import com.twitter.finagle.{Filter, Service}
import com.twitter.finagle.stats.{DefaultStatsReceiver, StatsReceiver}
import com.twitter.util.Duration

object ThriftForwardingWarmUpFilter {
  val thriftForwardingWarmupFilter = new Filter[Array[Byte], Array[Byte], ThriftClientRequest, Array[Byte]] {
    override def apply(request: Array[Byte], service: Service[ThriftClientRequest, Array[Byte]]) =
      service(new ThriftClientRequest(request, false))
  }
}

import ThriftForwardingWarmUpFilter.thriftForwardingWarmupFilter

class ThriftForwardingWarmUpFilter(
  warmupPeriod: Duration,
  forwardTo: Service[ThriftClientRequest, Array[Byte]],
  statsReceiver: StatsReceiver = DefaultStatsReceiver,
  isBypassClient: ClientId => Boolean = _ => true
) extends ForwardingWarmUpFilter[Array[Byte], Array[Byte]](
  warmupPeriod,
  thriftForwardingWarmupFilter andThen forwardTo,
  statsReceiver
) {

  override def bypassForward: Boolean = ClientId.current.forall(isBypassClient)
} 

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

import com.twitter.conversions.time._
import com.twitter.util.Duration
import org.jboss.netty.handler.codec.http.{Cookie => NettyCookie, DefaultCookie}
import scala.collection.JavaConverters._


class Cookie(private[http] val underlying: NettyCookie) {
  def this(name: String, value: String) = {
    this(new DefaultCookie(name, value))
  }

  def comment: String    = underlying.getComment
  def commentUrl: String = underlying.getCommentUrl
  def domain: String     = underlying.getDomain
  def maxAge: Duration   = underlying.getMaxAge.seconds
  def name: String       = underlying.getName
  def path: String       = underlying.getPath
  def ports: Set[Int]    = underlying.getPorts.asScala.toSet map { i: Integer => i.intValue }
  def value: String      = underlying.getValue
  def version: Int       = underlying.getVersion
  def httpOnly: Boolean  = underlying.isHttpOnly
  def isDiscard: Boolean = underlying.isDiscard
  def isSecure: Boolean  = underlying.isSecure

  def comment_=(comment: String)       { underlying.setComment(comment) }
  def commentUrl_=(commentUrl: String) { underlying.setCommentUrl(commentUrl) }
  def domain_=(domain: String)         { underlying.setDomain(domain) }
  def maxAge_=(maxAge: Duration)       { underlying.setMaxAge(maxAge.inSeconds) }
  def path_=(path: String)             { underlying.setPath(path) }
  def ports_=(ports: Seq[Int])         { underlying.setPorts(ports: _*) }
  def value_=(value: String)           { underlying.setValue(value) }
  def version_=(version: Int)          { underlying.setVersion(version) }
  def httpOnly_=(httpOnly: Boolean)    { underlying.setHttpOnly(httpOnly) }
  def isDiscard_=(discard: Boolean)    { underlying.setDiscard(discard) }
  def isSecure_=(secure: Boolean)      { underlying.setSecure(secure) }

  override def equals(obj: Any): Boolean = obj match {
    case c: Cookie => underlying.equals(c.underlying)
    case _ => false
  }

  override def hashCode(): Int = underlying.hashCode()
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.mux.lease.exp

import com.twitter.finagle.Stack
import com.twitter.util.Duration


  def observeArrival()
}

private[twitter] object Lessor {
  case class Param(lessor: Lessor)
  implicit object Param extends Stack.Param[Param] {
    val default = Param(ClockedDrainer.flagged)
  }

  val nil: Lessor = new Lessor {
    def register(lessee: Lessee) = ()
    def unregister(lessee: Lessee) = ()
    def observe(d: Duration) = ()
    def observeArrival() = ()
  }
} 

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

import com.twitter.finagle.{ClientConnection, Service, ServiceFactory}
import com.twitter.finagle.stats.StatsReceiver
import com.twitter.finagle.util.DefaultTimer
import com.twitter.util.{Duration, Future, Time}
import java.util.concurrent.atomic.AtomicInteger

private trait Server extends ServiceFactory[Unit, Unit] {
  
  def apply(
    id: String,
    nextLatency: () => Duration,
    sr: StatsReceiver
  ) = new Server {
    private val _load = new AtomicInteger(0)
    private val _maxLoad = new AtomicInteger(0)
    private val _numRequests = new AtomicInteger(0)

    private val service = new Service[Unit, Unit] {
      val numRequests = sr.counter("count")
      val gauges = Seq(
        sr.addGauge("load") { _load.get() },
        sr.addGauge("maxload") { _maxLoad.get() }
      )

      def apply(req: Unit) = {
        synchronized {
          val l = _load.incrementAndGet()
          if (l > _maxLoad.get()) _maxLoad.set(l)
        }
        numRequests.incr()
        _numRequests.incrementAndGet()
        Future.sleep(nextLatency())(DefaultTimer.twitter).ensure {
          _load.decrementAndGet()
        }
      }
    }

    def maxLoad = _maxLoad.get().toLong
    def count = _numRequests.get().toLong
    def apply(conn: ClientConnection) = Future.value(service)
    def close(deadline: Time) = Future.Done
    override def toString = id
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.kestrel.protocol

import com.twitter.io.Buf
import com.twitter.util.{Time, Duration}

sealed abstract class Command(val name: String)

sealed abstract class GetCommand(name: String) extends Command(name) {
  val queueName: Buf
  val timeout: Option[Duration]
}

case class Get(val queueName: Buf, val timeout: Option[Duration] = None)             extends GetCommand("Get")
case class Open(val queueName: Buf, val timeout: Option[Duration] = None)            extends GetCommand("Open")
case class Close(val queueName: Buf, val timeout: Option[Duration] = None)           extends GetCommand("Close")
case class CloseAndOpen(val queueName: Buf, val timeout: Option[Duration] = None)    extends GetCommand("CloseAndOpen")
case class Abort(val queueName: Buf, val timeout: Option[Duration] = None)           extends GetCommand("Abort")
case class Peek(val queueName: Buf, val timeout: Option[Duration] = None)            extends GetCommand("Peek")
case class Set(queueName: Buf, expiry: Time, value: Buf)                             extends Command("Set")

case class Delete(queueName: Buf)                                                    extends Command("Delete")
case class Flush(queueName: Buf)                                                     extends Command("Flush")
case class FlushAll()                                                                extends Command("FlushAll")

case class Version()                                                                 extends Command("Version")
case class ShutDown()                                                                extends Command("ShutDown")
case class Reload()                                                                  extends Command("Reload")
case class DumpConfig()                                                              extends Command("DumpConfig")
case class Stats()                                                                   extends Command("Stats")
case class DumpStats()                                                               extends Command("DumpStats") 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.http.filter

import com.twitter.finagle.Service
import com.twitter.finagle.http.{Request, Response}
import com.twitter.finagle.stats.InMemoryStatsReceiver
import com.twitter.util.{Await, Duration, Future, Time}
import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.junit.JUnitRunner
import org.mockito.Mockito.{spy, verify}

@RunWith(classOf[JUnitRunner])
class StatsFilterTest extends FunSuite {

  val service = new Service[Request, Response] {
    def apply(request: Request): Future[Response] = {
      val response = request.response
      response.statusCode = 404
      response.write("hello")
      Future.value(response)
    }
  }

  test("increment stats") {
    val receiver = spy(new InMemoryStatsReceiver)

    val filter = new StatsFilter(receiver) andThen service

    Time.withCurrentTimeFrozen { _ =>
      Await.result(filter(Request()), Duration.fromSeconds(5))
    }

    assert(receiver.counters(Seq("status", "404")) == 1)
    assert(receiver.counters(Seq("status", "4XX")) == 1)
    assert(receiver.stats(Seq("time", "404")) == Seq(0.0))
    assert(receiver.stats(Seq("time", "4XX")) == Seq(0.0))
    assert(receiver.stats(Seq("response_size")) == Seq(5.0))
  }

  test("status and time counters and stats are memoised") {
    val receiver = spy(new InMemoryStatsReceiver)

    val filter = new StatsFilter(receiver) andThen service

    Time.withCurrentTimeFrozen { _ =>
      Await.result(filter(Request()), Duration.fromSeconds(5))
      Await.result(filter(Request()), Duration.fromSeconds(5))
    }

    // Verify that the counters and stats were only created once
    verify(receiver).counter("status", "404")
    verify(receiver).counter("status", "4XX")
    verify(receiver).stat("time", "404")
    verify(receiver).stat("time", "4XX")
    verify(receiver).stat("response_size")
  }
} 

//设置package包名称以及导入依赖的类
package com.twitter.finagle.serverset2.client

import com.twitter.finagle.stats.StatsReceiver
import com.twitter.util.{Closable, Duration, Future, Time, Timer}

class StateTracker(
  statsReceiver: StatsReceiver,
  samplePeriod: Duration,
  timer: Timer
) extends Closable {

  private[this] var currState: Option[SessionState] = None
  private[this] var lastSample: Time = Time.now

  private[this] val timerTask = timer.schedule(Time.now + samplePeriod, samplePeriod) {
    sample()
  }

  def close(deadline: Time): Future[Unit] = {
    timerTask.close(deadline)
  }

  def transition(newState: SessionState): Unit = synchronized {
    sample()
    currState = Some(newState)
  }

  private[this] def sample(): Unit = synchronized {
    val now = Time.now
    val delta = now - lastSample
    lastSample = now
    currState foreach { state =>
      statsReceiver.counter(s"${state.name}_duration_ms").incr(delta.inMilliseconds.toInt)
    }
  }
}