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

import java.net.InetAddress
import java.util.concurrent.RejectedExecutionException

import org.elasticsearch.client.Client
import org.elasticsearch.client.transport.TransportClient
import org.elasticsearch.common.settings.Settings
import org.elasticsearch.common.transport.InetSocketTransportAddress
import webby.api.Application
import webby.mvc.AppStub

trait ElasticPluginConf extends webby.api.Plugin {

  private var _client: Client = _

  val clusterName: String
  val httpHost: InetAddress
  val httpPort: Int
  def transportPort: Int = 9300

  def client: Client = {
    if (_client == null) synchronized {
      if (_client == null) _client = startClient()
    }
    _client
  }

  override def onStart() {
    if (AppStub.isRealProd(app)) _client = startClient()
  }

  override def onStop() {
    if (_client != null) {
      try {
        _client.close()
      } catch {
        case e: RejectedExecutionException => webby.api.Logger(getClass).info("ElasticSearch bug on shutdown: " + e.toString)
      }
    }
  }

  protected def app: Application

  protected def startClient(): Client = {
    TransportClient.builder()
      .settings(Settings.builder().put("cluster.name", clusterName))
      .build()
      .addTransportAddress(new InetSocketTransportAddress(httpHost, transportPort))
  }
} 

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

import java.util.concurrent.RejectedExecutionException

import akka.event.Logging.Error
import com.codahale.metrics.MetricRegistry
import com.evolutiongaming.util.dispatchers.{Instrumented, InstrumentedConfig}

trait InstrumentedDispatcherMixin extends Dispatcher {

  private lazy val instrumented = {
    val config = InstrumentedConfig(configurator.config)
    new Instrumented(config, metricRegistry)
  }

  override def execute(runnable: Runnable): Unit = {
    instrumented(runnable, super.execute)
  }

  
  protected[akka] override def registerForExecution(mbox: Mailbox, hasMessageHint: Boolean, hasSystemMessageHint: Boolean): Boolean = {
    if (mbox.canBeScheduledForExecution(hasMessageHint, hasSystemMessageHint)) { //This needs to be here to ensure thread safety and no races
      if (mbox.setAsScheduled()) {
        try {
          instrumented(mbox, executorService.execute)
          true
        } catch {
          case _: RejectedExecutionException ?
            try {
              instrumented(mbox, executorService.execute)
              true
            } catch { //Retry once
              case e: RejectedExecutionException ?
                mbox.setAsIdle()
                eventStream.publish(Error(e, getClass.getName, getClass, "registerForExecution was rejected twice!"))
                throw e
            }
        }
      } else false
    } else false
  }

  def metricRegistry: MetricRegistry
}