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

import com.twitter.finagle.context.Contexts
import com.twitter.finagle.util.ByteArrays
import com.twitter.io.Buf
import com.twitter.util.{Return, Throw, Try}

case class UserContext(userId: Long)


  def current: Option[UserContext] = Contexts.broadcast.get(UserContext)

  override def marshal(userContext: UserContext): Buf = {
    val bytes = new Array[Byte](bodyLengthBytes)
    ByteArrays.put64be(bytes, 0, userContext.userId)
    Buf.ByteArray.Owned(bytes)
  }

  override def tryUnmarshal(body: Buf): Try[UserContext] = {
    if (body.length != bodyLengthBytes) {
      return Throw(new IllegalArgumentException(s"Invalid body. Length ${body.length} but required 16"))
    }

    val bytes = Buf.ByteArray.Owned.extract(body)
    val userId = ByteArrays.get64be(bytes, 0)

    Return(UserContext(userId))
  }
} 

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

import com.twitter.finagle.context.Contexts
import com.twitter.util.{Return, Throw}
import com.twitter.io.Buf

case class ClientId(name: String) {
  
object ClientId {
  // As a matter of legacy, we need to support the notion of
  // an empty client id. Old version of contexts could serialize
  // the absence of a client id with an empty buffer.
  private[finagle] val clientIdCtx = new Contexts.broadcast.Key[Option[ClientId]]("com.twitter.finagle.thrift.ClientIdContext") {
    def marshal(clientId: Option[ClientId]): Buf = clientId match {
      case None => Buf.Empty
      case Some(ClientId(name)) => Buf.Utf8(name)
    }

    def tryUnmarshal(buf: Buf) = buf match {
      case buf if buf.isEmpty => Return.None
      case Buf.Utf8(name) => Return(Some(ClientId(name)))
      case invalid => Throw(new IllegalArgumentException("client id not a utf8 string"))
    }
  }

  private[this] val NoClientFn: () => Option[ClientId] = () => None

  def current: Option[ClientId] =
    Contexts.broadcast.getOrElse(clientIdCtx, NoClientFn)

  private[finagle] def let[R](clientId: ClientId)(f: => R): R =
    Contexts.broadcast.let(clientIdCtx, Some(clientId))(f)

  private[finagle] def let[R](clientId: Option[ClientId])(f: => R): R = {
    clientId match {
      case Some(id) => Contexts.broadcast.let(clientIdCtx, Some(id))(f)
      case None => Contexts.broadcast.letClear(clientIdCtx)(f)
    }
  }
} 

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

import com.twitter.concurrent.AsyncMeter
import com.twitter.finagle.{Failure, Service, SimpleFilter}
import com.twitter.util.{Future, Throw}

import java.util.concurrent.RejectedExecutionException


class RequestMeterFilter[Req, Rep](meter: AsyncMeter) extends SimpleFilter[Req, Rep] {
  def apply(request: Req, service: Service[Req, Rep]) = {
    meter.await(1).transform {
      case Throw(noPermit) => noPermit match {
        case e: RejectedExecutionException =>
          Future.exception(Failure.rejected(noPermit))
        case e => Future.exception(e)
      }
      case _ => service(request)
    }
  }
} 

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

import com.twitter.finagle._
import com.twitter.finagle.util.DefaultMonitor
import com.twitter.util.{Future, Monitor, Throw, Try}
import scala.util.control.NonFatal

private[finagle] object MonitorFilter {
  val role: Stack.Role = Stack.Role("Monitoring")
  val description: String = "Act as last-resort exception handler"

  
class MonitorFilter[Req, Rep](monitor: Monitor) extends SimpleFilter[Req, Rep] {
  private[this] val RespondFn: Try[Rep] => Unit = {
    case Throw(exc) => monitor.handle(exc)
    case _ =>
  }

  def apply(request: Req, service: Service[Req, Rep]): Future[Rep] = {
    val saved = Monitor.get
    Monitor.set(monitor)
    try {
      service(request).respond(RespondFn)
    } catch {
      case NonFatal(e) =>
        monitor.handle(e)
        Future.exception(e)
    } finally {
      Monitor.set(saved)
    }
  }
} 

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

import com.twitter.finagle.util.{BufReader, BufWriter}
import com.twitter.io.Buf
import com.twitter.util.{Return, Throw, Try}


private[finagle] case class Retries(val retries: Int)

private[finagle] object Retries
  extends Contexts.broadcast.Key[Retries]("com.twitter.finagle.Retries")
{

  def current: Option[Retries] =
    Contexts.broadcast.get(Retries)

  override def marshal(retries: Retries): Buf = {
    val bw: BufWriter = BufWriter.fixed(4)
    bw.writeIntBE(retries.retries)
    bw.owned()
  }

  override def tryUnmarshal(buf: Buf): Try[Retries] = {
    if (buf.length != 4) {
      Throw(new IllegalArgumentException(
        s"Could not extract Retries from Buf. Length ${buf.length} but required 4"))
    } else {
      val retries: Int = BufReader(buf).readIntBE()
      Return(Retries(retries))
    }
  }
} 

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

import com.twitter.finagle._
import com.twitter.util.Throwables
import com.twitter.finagle.stats.{StatsReceiver, RollupStatsReceiver}
import com.twitter.util.{Future, Stopwatch, Return, Throw}

private[finagle] object StatsFactoryWrapper {
  val role = Stack.Role("ServiceCreationStats")

  
class StatsFactoryWrapper[Req, Rep](
    self: ServiceFactory[Req, Rep],
    statsReceiver: StatsReceiver)
  extends ServiceFactoryProxy[Req, Rep](self)
{
  private[this] val failureStats = statsReceiver.scope("failures")
  private[this] val latencyStat = statsReceiver.stat("service_acquisition_latency_ms")

  override def apply(conn: ClientConnection): Future[Service[Req, Rep]] = {
    val elapsed = Stopwatch.start()
    super.apply(conn) respond {
      case Throw(t) =>
        failureStats.counter(Throwables.mkString(t): _*).incr()
        latencyStat.add(elapsed().inMilliseconds)
      case Return(_) =>
        latencyStat.add(elapsed().inMilliseconds)
    }
  }
} 

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

import org.scalatest.FunSuite
import org.scalatest.junit.JUnitRunner
import org.junit.runner.RunWith
import com.twitter.finagle.Service
import com.twitter.util.TimeConversions._
import com.twitter.util.{Throw, Await, Try}

@RunWith(classOf[JUnitRunner])
class ServiceTest extends FunSuite {
  test("Service should rescue") {
    val e = new RuntimeException("yargs")
    val exceptionThrowingService = new Service[Int, Int] {
      def apply(request: Int) = {
        throw e
      }
    }

    assert(Try(Await.result(Service.rescue(exceptionThrowingService)(1), 1.second)) == Throw(e))
  }
} 

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

import com.twitter.finagle.Failure
import com.twitter.finagle.service.ResponseClass._
import com.twitter.util.{Return, Throw}
import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.junit.JUnitRunner

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

  test("Default classification") {
    assert("DefaultResponseClassifier" == ResponseClassifier.Default.toString)
    assert(Success ==
      ResponseClassifier.Default(ReqRep(null, Return("hi"))))

    assert(RetryableFailure ==
      ResponseClassifier.Default(ReqRep(null, Throw(Failure.rejected))))

    assert(NonRetryableFailure ==
      ResponseClassifier.Default(ReqRep(null, Throw(Failure("nope")))))
  }

  test("composition") {
    val aThrow = Throw(Failure("nope"))
    val aReturn = Return("yep")

    val evens: ResponseClassifier = {
      case ReqRep(i: Int, Throw(_)) if i % 2 == 0 => RetryableFailure
    }
    val odds: ResponseClassifier = {
      case ReqRep(i: Int, Throw(_)) if i % 2 == 1 => NonRetryableFailure
    }
    val classifier = evens.orElse(odds)

    assert(RetryableFailure == classifier(ReqRep(2, aThrow)))
    assert(NonRetryableFailure == classifier(ReqRep(1, aThrow)))

    assert(!classifier.isDefinedAt(ReqRep(0, aReturn)))
    assert(Success == classifier.applyOrElse(ReqRep(0, aReturn), ResponseClassifier.Default))
  }

} 

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

import com.twitter.finagle.dispatch.GenSerialClientDispatcher.wrapWriteException
import com.twitter.finagle.dispatch.SerialClientDispatcher
import com.twitter.finagle.stats.StatsReceiver
import com.twitter.finagle.transport.Transport
import com.twitter.util.{Future, Promise, Return, Throw}

private[finagle] class ThriftSerialClientDispatcher(
    transport: Transport[ThriftClientRequest, Array[Byte]],
    statsReceiver: StatsReceiver)
  extends SerialClientDispatcher[ThriftClientRequest, Array[Byte]](transport, statsReceiver) {

  
  override protected def dispatch(
    req: ThriftClientRequest,
    p: Promise[Array[Byte]]
  ): Future[Unit] = {
    if (!req.oneway) super.dispatch(req, p)
    else {
      write(req)
        .respond {
          case Return(_) => p.updateIfEmpty(ThriftSerialClientDispatcher.EmptyByteArray)
          case Throw(err) => wrapWriteException(err).respond(p.updateIfEmpty(_))
        }
    }

  }
}

private object ThriftSerialClientDispatcher {
  val EmptyByteArray = Return(new Array[Byte](0))
} 

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

import java.util.UUID

import algebra.Eq
import cats.Show
import com.twitter.io.Buf
import com.twitter.util.{Return, Throw, Try}
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner


@RunWith(classOf[JUnitRunner])
trait MissingInstances {
  implicit def eqTry[A](implicit A: Eq[A]): Eq[Try[A]] = Eq.instance {
    case (Return(a), Return(b)) => A.eqv(a, b)
    case (Throw(x), Throw(y)) => x == y
    case _ => false
  }

  implicit def eqUUID: Eq[UUID] = Eq.fromUniversalEquals

  implicit def showUUID: Show[UUID] = Show.fromToString

  implicit def eqBuf: Eq[Buf] = Eq.fromUniversalEquals
} 

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

import com.redbubble.util.http.Errors.jsonDecodeFailedError
import com.redbubble.util.json.CodecOps
import com.twitter.io.Buf
import com.twitter.util.{Return, Throw, Try}
import io.circe.Decoder
import io.finch.Decode

trait RequestOps {
  type JsonCleaner = (Buf) => Buf

  
  final def decodeRootJson[A](d: Decoder[A], c: JsonCleaner = identity): Decode.Json[A] =
    Decode.json((payload, _) => decodePayload(c(payload), rootObjectDecoder(d)))

  private def decodePayload[A](payload: Buf, decoder: Decoder[A]): Try[A] = {
    val decodedPayload = CodecOps.decode(payload)(decoder)
    decodedPayload.fold(
      error => Throw(jsonDecodeFailedError(s"Unable to decode JSON payload: ${error.getMessage}", error)),
      value => Return(value)
    )
  }

  private def dataFieldObjectDecoder[A](implicit d: Decoder[A]): Decoder[A] =
    Decoder.instance(c => c.downField("data").as[A](d))

  private def rootObjectDecoder[A](implicit d: Decoder[A]): Decoder[A] = Decoder.instance(c => c.as[A](d))
}

object RequestOps extends RequestOps 

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

import com.twitter.conversions.time._
import com.twitter.finagle.{Service, SimpleFilter, NoStacktrace}
import com.twitter.finagle.stats.StatsReceiver
import com.twitter.util.{Future, Return, Throw, Duration, Timer}


class BackupRequestFilter[Req, Rep](delay: Duration, timer: Timer, statsReceiver: StatsReceiver)
  extends SimpleFilter[Req, Rep]
{
  assert(delay >= 0.seconds)

  private[this] val backups = statsReceiver.scope("backup")
  private[this] val timeouts = backups.counter("timeouts")
  private[this] val won = backups.counter("won")
  private[this] val lost = backups.counter("lost")

  def apply(req: Req, service: Service[Req, Rep]): Future[Rep] = {
    val backup = timer.doLater(delay) {
      timeouts.incr()
      service(req)
    } flatten

    Future.select(Seq(service(req), backup)) flatMap {
      case (Return(res), Seq(other)) =>
        if (other eq backup) won.incr() else lost.incr()
        other.raise(BackupRequestLost)
        Future.value(res)
      case (Throw(_), Seq(other)) => other
    }
  }
}

object BackupRequestLost extends Exception with NoStacktrace 

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

import com.twitter.finagle.{ClientConnection, Service, ServiceFactory,
  ServiceFactoryProxy, ServiceProxy}
import com.twitter.util.{Future, Return, Throw, Time}
import java.util.concurrent.atomic.AtomicReference
import scala.annotation.tailrec


class ReusingPool[Req, Rep](underlying: ServiceFactory[Req, Rep])
    extends ServiceFactoryProxy[Req, Rep](underlying)
{
  private[this] val current: AtomicReference[Future[ServiceProxy[Req, Rep]]] =
    new AtomicReference(Future.exception(new Exception))

  private[this] def newService(conn: ClientConnection) =
    underlying(conn) map { service =>
      new ServiceProxy(service) {
        override def close(deadline: Time) = Future.Done
      }
    }

  @tailrec
  override final def apply(conn: ClientConnection): Future[Service[Req, Rep]] = {
    val f = current.get
    f.poll match {
      case Some(Return(service)) if service.isAvailable => f
      // Still waiting for connect.
      case None => f
      // Last connect failed.
      case Some(Throw(_)) =>
        if (!current.compareAndSet(f, Future.never)) apply(conn) else {
          current.set(newService(conn))
          current.get
        }
      // Service is dead. Give it back.
      case Some(Return(dead)) =>
        if (!current.compareAndSet(f, Future.never)) apply(conn) else {
          dead.self.close()
          current.set(newService(conn))
          current.get
        }
    }
  }
} 

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

import java.util.concurrent.atomic.AtomicInteger

import com.twitter.util.{Future, Stopwatch, Throw}
import com.twitter.finagle.stats.StatsReceiver
import com.twitter.finagle.{SourcedException, Service, SimpleFilter}

class StatsFilter[Req, Rep](statsReceiver: StatsReceiver)
  extends SimpleFilter[Req, Rep]
{
  private[this] val outstandingRequestCount = new AtomicInteger(0)
  private[this] val dispatchCount = statsReceiver.counter("requests")
  private[this] val successCount = statsReceiver.counter("success")
  private[this] val latencyStat = statsReceiver.stat("request_latency_ms")
  private[this] val outstandingRequestCountgauge =
    statsReceiver.addGauge("pending") { outstandingRequestCount.get }

  def apply(request: Req, service: Service[Req, Rep]): Future[Rep] = {
    val elapsed = Stopwatch.start()
    dispatchCount.incr()

    outstandingRequestCount.incrementAndGet()
    val result = service(request)

    result respond { response =>
      outstandingRequestCount.decrementAndGet()
      latencyStat.add(elapsed().inMilliseconds)
      response match {
        case Throw(e) =>
          def flatten(ex: Throwable): Seq[String] =
            if (ex eq null) Seq[String]() else ex.getClass.getName +: flatten(ex.getCause)
          statsReceiver.scope("failures").counter(flatten(e): _*).incr()
          e match {
            case sourced: SourcedException if sourced.serviceName != "unspecified" =>
              statsReceiver
                .scope("sourcedfailures")
                .counter(sourced.serviceName +: flatten(sourced): _*)
                .incr()
            case _ =>
          }
        case _ => successCount.incr()
      }
    }

    result
  }
} 

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

import org.specs.SpecificationWithJUnit

import com.twitter.finagle.{Service, Filter}
import com.twitter.util.{Throw, Future}
import com.twitter.conversions.time._

class FilterSpec extends SpecificationWithJUnit {
  "filters" should {
    val stringToInt =
      new Filter[Int, Int, String, String] {
        def apply(request: Int, service: Service[String, String]) =
          service(request.toString) map (_.toInt)
      }

    val intToString =
      new Filter[String, String, Int, Int] {
        def apply(request: String, service: Service[Int, Int]) =
          service(request.toInt) map (_.toString)
      }

    "compose" in {
      "when it's all chill" in {
        val filter = stringToInt andThen intToString

        val service = new Service[Int, Int] {
          def apply(request: Int) = Future(2 * request.intValue)
        }

        val result = (filter andThen service)(123)

        result.isReturn must beTrue
        result() must be_==(123 * 2)
      }

      "when synchronous exceptions are thrown" in {
        val e = new RuntimeException("yargs")
        val exceptionThrowingService = new Service[Int, Int] {
          def apply(request: Int) = {
            throw e
            Future.value(request + 1)
          }
        }

        "with simple composition" in {
          intToString.andThen(exceptionThrowingService)("1").get(1.second) must
            be_==(Throw(e))
        }

        "with transitive composition" in {
          stringToInt.andThen(intToString.andThen(exceptionThrowingService))(1).get(1.second) must
            be_==(Throw(e))
          (stringToInt.andThen(intToString)).andThen(exceptionThrowingService)(1).get(1.second) must
            be_==(Throw(e))
        }
      }
    }
  }
} 

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

import org.specs.SpecificationWithJUnit

import com.twitter.finagle.Service
import com.twitter.util.{Future, Throw}
import com.twitter.conversions.time._

class ServiceSpec extends SpecificationWithJUnit {
  "Service" should {
    "rescue" in {
      val e = new RuntimeException("yargs")
      val exceptionThrowingService = new Service[Int, Int] {
        def apply(request: Int) = {
          throw e
          Future.value(request + 1)
        }
      }

      Service.rescue(exceptionThrowingService)(1).get(1.second) must be_==(Throw(e))
    }
  }
} 

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

import com.twitter.finagle.{Service, SimpleFilter}
import com.twitter.finagle.stats.StatsReceiver
import com.twitter.finagle.http.{Request, Response, Status}
import com.twitter.util.{Duration, Future, Return, Stopwatch, Throw}



class StatsFilter[REQUEST <: Request](stats: StatsReceiver)
  extends SimpleFilter[REQUEST, Response] {

  def apply(request: REQUEST, service: Service[REQUEST, Response]): Future[Response] = {
    val elapsed = Stopwatch.start()
    val future = service(request)
    future respond {
      case Return(response) =>
        count(elapsed(), response)
      case Throw(_) =>
        // Treat exceptions as empty 500 errors
        val response = Response(request.version, Status.InternalServerError)
        count(elapsed(), response)
    }
    future
  }

  protected def count(duration: Duration, response: Response) {
    val statusCode = response.statusCode.toString
    val statusClass = (response.statusCode / 100).toString + "XX"
    stats.scope("status").counter(statusCode).incr()
    stats.scope("status").counter(statusClass).incr()

    stats.scope("time").stat(statusCode).add(duration.inMilliseconds)
    stats.scope("time").stat(statusClass).add(duration.inMilliseconds)

    stats.stat("response_size").add(response.length)
  }
} 

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

import com.twitter.finagle.ThriftMux
import com.twitter.util.{Future, Try, Throw}
import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.junit.JUnitRunner

@RunWith(classOf[JUnitRunner])
class ThriftIfaceTest extends FunSuite {
  test("invalid thrift ifaces") {
    trait FakeThriftIface {
      def query(x: String): Future[String]
    }

    val res = Try {
      ThriftMux.serveIface(":*", new FakeThriftIface { def query(x: String) = Future.value(x) })
    }
    assert(res.isThrow)
  }
}