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

import com.twitter.finagle._
import com.twitter.finagle.ServiceFactory
import com.twitter.util.{Closable, Future, Time}

object Client {
  def apply (factory: ServiceFactory[Request, Result]): Client = 
    new StdClient(factory)
}

trait Client extends Closable {
  def query(req: Request): Future[Result]
}

final class StdClient(val factory: ServiceFactory[Request, Result]) extends Client {
  private[this] val service = factory.toService

  def query(req: Request): Future[Result] = {
    val query = new Query(req.query)
    service(req)
  }

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

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

import com.twitter.finagle.{ClientConnection, ServiceFactory, Service}
import com.twitter.util.{Time, Future}


class MockService extends Service[Request, Result] {
  var requests = List[Request]()
  val resultSet = new ResultSet(fields=Seq(), rows=Seq())

  def apply(request: Request): Future[Result] = {
    requests = requests ++ List(request)
    Future.value(resultSet)
  }
}

class MockServiceFactory(service: Service[Request, Result]) extends ServiceFactory[Request, Result] {
  def apply(conn: ClientConnection): Future[Service[Request, Result]] = Future.value(service)
  def close(deadline: Time): Future[Unit] = Future.Unit
} 

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

import com.twitter.finagle.Mysql
import com.twitter.finagle.mysql.transport.TransportImpl
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
import org.scalatest.{MustMatchers, FunSuite}
import org.scalatest.mock.MockitoSugar
import org.mockito.Mockito._
import org.mockito.Matchers._
import com.twitter.util.Time


@RunWith(classOf[JUnitRunner])
class ClientTest extends FunSuite with MockitoSugar with MustMatchers {
  private val sqlQuery = "SELECT * FROM FOO"

  test("basic test creates a new service for each query") {
    val service = new MockService()
    val factory = spy(new MockServiceFactory(service))
    val client = spy(Client(factory))

    client.query(sqlQuery)
    client.query(sqlQuery)

    service.requests must equal (List(
     sqlQuery, sqlQuery
    ).map(QueryRequest(_)))

    verify(client, times(2)).query(sqlQuery)
    verify(factory, times(2)).apply()
    verify(factory, times(0)).close(any[Time])
  }

  test("Client uses Netty3 by default, but can be toggled to netty4") {
    val params = Mysql.client.params

    assert(params[TransportImpl].transporter(params).toString.equals("Netty3Transporter"))
  }
} 

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

import com.twitter.finagle.ServiceFactory
import com.twitter.finagle.util.OnReady
import com.twitter.util.{Time, Activity, Future, Promise}


  private[this] val observation = activity.states.respond {
    case Activity.Pending =>

    case Activity.Ok(newList) =>
      update(newList)
      ready.setDone()

    case Activity.Failed(_) =>
      // On resolution failure, consider the
      // load balancer ready (to serve errors).
      ready.setDone()
  }

  override def close(deadline: Time): Future[Unit] = {
    observation.close(deadline).transform { _ =>
      super.close(deadline)
    }.ensure {
      ready.setDone()
    }
  }
} 

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

import com.twitter.finagle.{ClientConnection, Service, ServiceFactory, ServiceFactoryProxy, ServiceProxy}
import com.twitter.finagle.service.FailingFactory
import com.twitter.finagle.stats.StatsReceiver
import com.twitter.finagle.util.Rng
import com.twitter.util.{Throw, Time, Future, Return}
import java.util.concurrent.atomic.AtomicInteger


private[loadbalancer] trait LeastLoaded[Req, Rep] { self: Balancer[Req, Rep] =>
  protected def rng: Rng

  protected case class Node(
      factory: ServiceFactory[Req, Rep],
      counter: AtomicInteger,
      token: Int)
    extends ServiceFactoryProxy[Req, Rep](factory)
    with NodeT[Req, Rep] {

    type This = Node

    def load: Double = counter.get
    def pending: Int = counter.get

    override def apply(conn: ClientConnection): Future[Service[Req, Rep]] = {
      counter.incrementAndGet()
      super.apply(conn).transform {
        case Return(svc) =>
          Future.value(new ServiceProxy(svc) {
            override def close(deadline: Time) =
              super.close(deadline).ensure {
                counter.decrementAndGet()
              }
          })

        case [email protected](_) =>
          counter.decrementAndGet()
          Future.const(t)
      }
    }
  }

  protected def newNode(factory: ServiceFactory[Req, Rep], statsReceiver: StatsReceiver) =
    Node(factory, new AtomicInteger(0), rng.nextInt())

  private[this] val failingLoad = new AtomicInteger(0)
  protected def failingNode(cause: Throwable) = Node(new FailingFactory(cause), failingLoad, 0)
} 

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

import com.twitter.util.{Time, Future}
import com.twitter.jvm.Jvm
import com.twitter.finagle.{Service, SimpleFilter}
import com.twitter.finagle.tracing.{Trace, Record, Annotation}
import com.twitter.conversions.time._


class MkJvmFilter(jvm: Jvm) {
  private[this] val buffer = jvm.monitorGcs(1.minute)

  def apply[Req, Rep](): SimpleFilter[Req, Rep] = new SimpleFilter[Req, Rep] {
    def apply(req: Req, service: Service[Req, Rep]): Future[Rep] = {
      val begin = Time.now
      if (Trace.isActivelyTracing) {
        service(req) ensure {
          buffer(begin) foreach { gc =>
            Trace.record {
              Record(Trace.id, gc.timestamp, Annotation.Message(gc.toString), Some(gc.duration))
            }
          }
        }
      }
      else
        service(req)
    }
  }
} 

//设置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.util

import com.twitter.util.{Closable, Future, Promise, Return, Time}


    def onClose(h: => Unit) = {
      if (closing.isDefined)
        h
      else
        closeHandlers ::= { () => h }
    }

     // Invokes close handlers in reverse order from which they were added.
    closing ensure { closeHandlers foreach { handler =>
      handler()
    }}
  }

  def makeLifoCloser(): CloseNotifier with Closable = new CloseNotifier with Closable {
    private[this] val closing = new Promise[Unit]
    private[this] val notifier = makeLifo(closing)

    def close(deadline: Time) = {
      closing.updateIfEmpty(Return(()))
      Future.Done
    }
    def onClose(h: => Unit) = notifier.onClose(h)
  }
} 

//设置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.service

import com.twitter.finagle.{Service, ServiceProxy}
import com.twitter.finagle.util.AsyncLatch
import com.twitter.util.{Future, Promise, Time, Try}


private[finagle] class RefcountedService[Req, Rep](underlying: Service[Req, Rep])
  extends ServiceProxy[Req, Rep](underlying) {

  private[this] val replyLatch = new AsyncLatch()

  private[this] val decrLatchFn: Try[Rep] => Unit =
    _ => replyLatch.decr()

  override def apply(request: Req): Future[Rep] = {
    replyLatch.incr()
    underlying(request).respond(decrLatchFn)
  }

  override final def close(deadline: Time): Future[Unit] = {
    val p = new Promise[Unit]
    replyLatch.await {
      p.become(underlying.close(deadline))
    }
    p
  }
} 

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

import com.twitter.finagle.util.AsyncLatch
import com.twitter.finagle.{Service, ServiceFactory, ClientConnection}
import com.twitter.util.{Future, Promise, Time}


class SingletonFactory[Req, Rep](service: Service[Req, Rep])
  extends ServiceFactory[Req, Rep]
{
  private[this] var latch = new AsyncLatch

  def apply(conn: ClientConnection) = Future {
    latch.incr()
    new Service[Req, Rep] {
      def apply(request: Req) = service(request)
      override def close(deadline: Time) = { latch.decr(); Future.Done }
    }
  }

  def close(deadline: Time) = {
    val p = new Promise[Unit]
    latch.await {
      service.close()
      p.setDone()
    }
    p
  }

  override def status = service.status

  override val toString = "singleton_factory_%s".format(service.toString)
} 

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

import com.twitter.finagle.{Status, Service, ServiceClosedException, ServiceProxy, WriteException}
import com.twitter.util.{Future, Time}
import java.util.concurrent.atomic.AtomicBoolean


private[finagle] class CloseOnReleaseService[Req, Rep](underlying: Service[Req, Rep])
  extends ServiceProxy[Req, Rep](underlying)
{
  private[this] val wasReleased = new AtomicBoolean(false)

  override def apply(request: Req) = {
    if (!wasReleased.get) {
      super.apply(request)
    } else {
      Future.exception(
        WriteException(new ServiceClosedException))
    }
  }

  override def close(deadline: Time) = {
    if (wasReleased.compareAndSet(false, true))
      super.close(deadline)
    else
      Future.Done
  }

  override def status = 
    if (wasReleased.get) Status.Closed
    else super.status
} 

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

import com.twitter.finagle.Service
import com.twitter.finagle.stats.InMemoryStatsReceiver
import com.twitter.util.{Stopwatch, Time, Future}
import com.twitter.conversions.time._
import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.junit.JUnitRunner

@RunWith(classOf[JUnitRunner])
class ServerStatsFilterTest extends FunSuite {
  test("Records handletime for a service") {
    Time.withCurrentTimeFrozen { ctl =>
      val inMemory = new InMemoryStatsReceiver
      val svc = Service.mk[Unit, Unit] { unit =>
        ctl.advance(5.microseconds)
        Future.never
      }
      val filter = new ServerStatsFilter[Unit, Unit](inMemory, Stopwatch.timeNanos)
      filter.andThen(svc)(())
      val expected = 5
      val actual = inMemory.stats(Seq("handletime_us"))(0)
      assert(actual == expected)
    }
  }
} 

//设置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.http.exp

import com.twitter.finagle.Status
import com.twitter.finagle.transport.Transport
import com.twitter.util.{Future, Time}
import java.net.SocketAddress
import java.security.cert.Certificate


  def read(): Future[Multi[Out]]
}

private[finagle] abstract class StreamTransportProxy[In, Out](self: Transport[_, _])
  extends StreamTransport[In, Out] {

  def status: Status = self.status
  val onClose: Future[Throwable] = self.onClose
  def localAddress: SocketAddress = self.localAddress
  def remoteAddress: SocketAddress = self.remoteAddress
  def close(deadline: Time): Future[Unit] = self.close(deadline)
  def peerCertificate: Option[Certificate] = self.peerCertificate
} 

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

import com.twitter.util.{Time, MockTimer}
import com.twitter.conversions.time.intToTimeableNumber
import com.twitter.conversions.storage.intToStorageUnitableWholeNumber
import org.junit.runner.RunWith
import org.mockito.Mockito.when
import org.scalatest.FunSuite
import org.scalatest.junit.JUnitRunner
import org.scalatest.mock.MockitoSugar

@RunWith(classOf[JUnitRunner])
class RequestSnooperTest extends FunSuite with MockitoSugar {
  test("RequestSnooper should compute handleBytes reasonably") {
    val ctr = mock[ByteCounter]
    val quantile = 0.50

    when(ctr.rate()).thenReturn(1)

    Time.withCurrentTimeFrozen { ctl =>
      when(ctr.lastGc).thenReturn(Time.now - 5.seconds)

      val tmr = new MockTimer()
      val snooper = new RequestSnooper(ctr, quantile, timer = tmr)
      for (_ <- 0 until 50)
        snooper.observe(1.second)
      for (_ <- 0 until 50)
        snooper.observe(2.seconds)
      for (_ <- 0 until 50)
        snooper.observe(3.seconds)
      ctl.advance(12.seconds)
      tmr.tick()
      assert(snooper.handleBytes() == 2000.bytes)
    }
  }

  test("RequestSnooper should discard results that overlap with a gc") {
    val ctr = mock[ByteCounter]
    val quantile = 0.50

    when(ctr.rate()).thenReturn(1)

    Time.withCurrentTimeFrozen { ctl =>
      when(ctr.lastGc).thenReturn(Time.now - 5.seconds)

      val tmr = new MockTimer()
      val snooper = new RequestSnooper(ctr, quantile, timer = tmr)
      for (_ <- 0 until 50)
        snooper.observe(1.second)
      for (_ <- 0 until 50)
        snooper.observe(2.seconds)
      for (_ <- 0 until 50)
        snooper.observe(3.seconds)
      for (_ <- 0 until 1000)
        snooper.observe(8.seconds)
      ctl.advance(12.seconds)
      tmr.tick()
      assert(snooper.handleBytes() == 2000.bytes)
    }
  }
} 

//设置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.tracing

import com.twitter.finagle.benchmark.StdBenchAnnotations
import com.twitter.finagle.context.{Deadline, Contexts}
import com.twitter.finagle.thrift.ClientId
import com.twitter.util.Time
import org.openjdk.jmh.annotations._
import org.openjdk.jmh.infra.Blackhole

@OperationsPerInvocation(50)
@State(Scope.Benchmark)
class TraceBenchmark extends StdBenchAnnotations {

  private[this] val traceId = TraceId(
    Some(SpanId(5L)),
    Some(SpanId(6L)),
    SpanId(7L),
    None,
    Flags.apply()
  )

  // Note that this needs to == to @OperationsPerInvocation above
  private[this] val Iterations = 50

  private[this] val clientId = ClientId("bench")

  private[this] val deadline = Deadline(Time.Top, Time.Top)

  @Benchmark
  def contexts0(hole: Blackhole): Unit = {
    // Because Contexts are only scoped within a `let`, we want
    // to amortize the cost of setting up the context and try
    // to focus this measurement on the call to `Trace.id`.
    var i = 0
    while (i < Iterations) {
      hole.consume(Trace.id)
      i += 1
    }
  }

  @Benchmark
  def contexts1(hole: Blackhole): Unit =
    Trace.letId(traceId, terminal = false) { contexts0(hole) }

  @Benchmark
  def contexts2(hole: Blackhole): Unit =
    clientId.asCurrent { contexts1(hole) }

  @Benchmark
  def contexts3(hole: Blackhole): Unit =
    Contexts.broadcast.let(Deadline, deadline) { contexts2(hole) }

  @Benchmark
  def contexts4(hole: Blackhole): Unit =
    Contexts.broadcast.letClear(Deadline) { contexts3(hole) }

  @Benchmark
  def contexts5(hole: Blackhole): Unit =
    Trace.letTracer(NullTracer) { contexts4(hole) }

} 

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

import com.twitter.finagle.redis.RedisRequestTest
import com.twitter.finagle.redis.tags.CodecTest
import com.twitter.io.Buf
import com.twitter.util.Time
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner

@RunWith(classOf[JUnitRunner])
final class KeyCodecSuite extends RedisRequestTest {

  test("DEL", CodecTest) { checkMultiKey("DEL", Del.apply) }
  test("DUMP", CodecTest) { checkSingleKey("DUMP", Dump.apply) }
  test("EXISTS", CodecTest) { checkSingleKey("EXISTS", Exists.apply) }
  test("KEYS", CodecTest) { checkSingleKey("KEYS", Keys.apply) }
  test("MOVE", CodecTest) { checkSingleKeySingleVal("MOVE", Move.apply) }
  test("PERSIST", CodecTest) { checkSingleKey("PERSIST", Persist.apply) }
  test("RENAME", CodecTest) { checkSingleKeySingleVal("RENAME", Rename.apply) }
  test("RENAMENX", CodecTest) { checkSingleKeySingleVal("RENAMENX", RenameNx.apply) }
  test("RANDOMKEY", CodecTest) { assert(encode(Randomkey) == Seq("RANDOMKEY")) }
  test("TTL", CodecTest) { checkSingleKey("TTL", Ttl.apply) }
  test("PTTL", CodecTest) { checkSingleKey("PTTL", PTtl.apply) }
  test("TYPE", CodecTest) { checkSingleKey("TYPE", Type.apply) }
  test("EXPIRE", CodecTest) { checkSingleKeyArbitraryVal("EXPIRE", Expire.apply) }
  test("EXPIREAT", CodecTest) {
    checkSingleKeyArbitraryVal("EXPIREAT", (k: Buf, v: Int) => ExpireAt(k, Time.fromSeconds(v)))
  }
  test("PEXPIRE", CodecTest) { checkSingleKeyArbitraryVal("PEXPIRE", PExpire.apply) }
  test("PEXPIREAT", CodecTest) {
    checkSingleKeyArbitraryVal("PEXPIREAT",
      (k: Buf, v: Int) => PExpireAt(k, Time.fromMilliseconds(v.toLong))
    )
  }

  test("SCAN", CodecTest) {
    assert(encode(Scan(42, None, None)) == Seq("SCAN", "42"))
    assert(encode(Scan(42, Some(10L), None)) == Seq("SCAN", "42", "COUNT", "10"))
    assert(encode(Scan(42, None, Some(Buf.Utf8("foo")))) == Seq("SCAN", "42", "PATTERN", "foo"))
    assert(encode(Scan(42, Some(10L), Some(Buf.Utf8("foo")))) ==
      Seq("SCAN", "42", "COUNT", "10", "PATTERN", "foo"))
  }
}