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

import io.netty.buffer.Unpooled
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}
import io.netty.handler.codec.http._



@Sharable
private[http] object RespondToExpectContinue extends ChannelInboundHandlerAdapter {

  def newContinue(): FullHttpResponse =
    new DefaultFullHttpResponse(
      HttpVersion.HTTP_1_1,
      HttpResponseStatus.CONTINUE,
      Unpooled.EMPTY_BUFFER)

  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = {
    msg match {
      case http: HttpMessage if HttpUtil.is100ContinueExpected(http) =>
        ctx.writeAndFlush(newContinue())
        http.headers.remove(HttpHeaderNames.EXPECT)
      case _ =>
    }
    super.channelRead(ctx, msg)
  }
} 

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

import com.twitter.finagle.thrift.ThriftClientRequest
import com.twitter.finagle.thrift.transport.ExceptionFactory
import io.netty.buffer.Unpooled
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{
  ChannelHandler, ChannelHandlerContext, ChannelOutboundHandlerAdapter,
  ChannelPromise, CombinedChannelDuplexHandler}


  @Sharable
  private object ThriftClientArrayToByteBufEncoder extends ChannelOutboundHandlerAdapter {

    override def write(ctx: ChannelHandlerContext, msg: Any, promise: ChannelPromise): Unit = msg match {
      case request: ThriftClientRequest =>
        val buf = Unpooled.wrappedBuffer(request.message)
        ctx.writeAndFlush(buf, promise)

      case other =>
        val ex = ExceptionFactory.wrongClientWriteType(other)
        promise.setFailure(ex)
        throw ex
    }
  }
} 

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

import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}


@Sharable
private[netty4] object ThriftByteBufToArrayDecoder extends ChannelInboundHandlerAdapter {
  override def channelRead(
    ctx: ChannelHandlerContext,
    msg: scala.Any
  ): Unit = msg match {
    case buffer: ByteBuf =>
      // toArray takes ownership of the buffer
      val array = toArray(buffer)
      ctx.fireChannelRead(array)

    case _ => throw new IllegalArgumentException("no byte buffer")
  }

  // takes ownership of the passed `ByteBuf`
  private def toArray(buffer: ByteBuf): Array[Byte] = {
    val array = new Array[Byte](buffer.readableBytes())
    buffer.readBytes(array)
    buffer.release()    // If you love it, set it free.
    array
  }
} 

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

import com.twitter.finagle.thrift.transport.ExceptionFactory
import io.netty.buffer.Unpooled
import io.netty.channel.{
  ChannelHandler, ChannelHandlerContext, ChannelOutboundHandlerAdapter,
  ChannelPromise, CombinedChannelDuplexHandler}
import io.netty.channel.ChannelHandler.Sharable


private[netty4] object ServerByteBufCodec {

  def apply(): ChannelHandler = {
    val encoder = ThriftServerArrayToByteBufEncoder
    val decoder = ThriftByteBufToArrayDecoder
    new CombinedChannelDuplexHandler(decoder, encoder)
  }

  @Sharable
  private object ThriftServerArrayToByteBufEncoder extends ChannelOutboundHandlerAdapter {
    override def write(ctx: ChannelHandlerContext, msg: Any, promise: ChannelPromise): Unit =
      msg match {
        case array: Array[Byte] =>
          val buf = Unpooled.wrappedBuffer(array)
          ctx.writeAndFlush(buf, promise)

        case other =>
          val ex = ExceptionFactory.wrongServerWriteType(other)
          promise.setFailure(ex)
          throw ex
      }
  }
} 

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

import com.twitter.finagle.transport.Transport
import io.netty.channel.{ChannelInitializer, Channel}
import io.netty.channel.ChannelHandler.Sharable


@Sharable
private[netty4] class ServerBridge[In, Out](
    transportFac: Channel => Transport[In, Out],
    serveTransport: Transport[In, Out] => Unit)
  extends ChannelInitializer[Channel] {

  def initChannel(ch: Channel): Unit = {
    val transport: Transport[In, Out] = transportFac(ch)
    serveTransport(transport)
  }
} 

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

import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}


@Sharable
object DirectToHeapInboundHandler extends ChannelInboundHandlerAdapter  {
  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = msg match {
    case bb: ByteBuf if bb.isDirect =>
      val heapBuf = ctx.alloc().heapBuffer(bb.readableBytes, bb.capacity)
      heapBuf.writeBytes(bb)

      bb.release()
      ctx.fireChannelRead(heapBuf)
    case _ => ctx.fireChannelRead(msg)
  }
} 

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

import com.twitter.finagle.stats.StatsReceiver
import io.netty.channel._
import io.netty.channel.ChannelHandler.Sharable
import io.netty.util.AttributeKey
import java.util.concurrent.atomic.AtomicInteger

private[finagle] object ChannelRequestStatsHandler {
  private[channel] val ConnectionRequestsKey: AttributeKey[AtomicInteger] =
    AttributeKey.valueOf("ChannelRequestStatsHandler.connection_requests")
}


@Sharable
private[finagle] class ChannelRequestStatsHandler(statsReceiver: StatsReceiver)
  extends ChannelInboundHandlerAdapter
{
  import ChannelRequestStatsHandler.ConnectionRequestsKey

  private[this] val requestCount = statsReceiver.stat("connection_requests")

  override def handlerAdded(ctx: ChannelHandlerContext): Unit = {
    ctx.channel.attr(ConnectionRequestsKey).set(new AtomicInteger(0))
    super.handlerAdded(ctx)
  }

  override def channelInactive(ctx: ChannelHandlerContext): Unit = {
    requestCount.add(ctx.channel.attr(ConnectionRequestsKey).get.get)
    super.channelInactive(ctx)
  }

  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = {
    val readCount = ctx.channel.attr(ConnectionRequestsKey).get
    readCount.incrementAndGet()
    super.channelRead(ctx, msg)
  }
} 

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

import com.twitter.finagle.stats.StatsReceiver
import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}


@Sharable
private[netty4] class RecvBufferSizeStatsHandler(stats: StatsReceiver)
  extends ChannelInboundHandlerAdapter {

  private[this] val receiveBufferBytes = stats.stat("receive_buffer_bytes")

  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = {
    msg match {
      case bb: ByteBuf => receiveBufferBytes.add(bb.readableBytes().toFloat)
      case _ => // NOOP
    }

    ctx.fireChannelRead(msg)
  }
} 

//设置package包名称以及导入依赖的类
package com.wavesplatform.network

import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}
import scorex.transaction.{History}
import scorex.utils.ScorexLogging

@Sharable
class HistoryReplier(history: History, maxChainLength: Int) extends ChannelInboundHandlerAdapter with ScorexLogging {
  override def channelRead(ctx: ChannelHandlerContext, msg: AnyRef): Unit = msg match {
    case GetSignatures(otherSigs) =>
      otherSigs.view
        .map(parent => parent -> history.blockIdsAfter(parent, maxChainLength))
        .find(_._2.nonEmpty) match {
        case Some((parent, extension)) =>
          log.debug(s"${id(ctx)} Got GetSignatures with ${otherSigs.length}, found common parent $parent and sending ${extension.length} more signatures")
          ctx.writeAndFlush(Signatures(parent +: extension))
        case None if otherSigs.length == 1 && otherSigs.head == history.lastBlock.get.uniqueId =>
          // this is the special case when both nodes only have genesis block
          log.debug(s"${id(ctx)} Both local and remote nodes only have genesis block")
          ctx.writeAndFlush(Signatures(otherSigs))
        case _ =>
          log.debug(s"${id(ctx)} Got GetSignatures with ${otherSigs.length} signatures, but could not find an extension")
      }

    case GetBlock(sig) =>
      for (h <- history.heightOf(sig); bytes <- history.blockBytes(h)) {
        ctx.writeAndFlush(RawBytes(BlockMessageSpec.messageCode, bytes))
      }

    case _: Handshake =>
      ctx.writeAndFlush(LocalScoreChanged(history.score()))

    case _ => super.channelRead(ctx, msg)
  }
} 

//设置package包名称以及导入依赖的类
package com.wavesplatform.network

import java.net.{InetAddress, InetSocketAddress}
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger

import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelFuture, ChannelHandlerContext}
import io.netty.handler.ipfilter.AbstractRemoteAddressFilter
import scorex.utils.ScorexLogging

@Sharable
class InboundConnectionFilter(peerDatabase: PeerDatabase, maxInboundConnections: Int, maxConnectionsPerHost: Int)
  extends AbstractRemoteAddressFilter[InetSocketAddress] with ScorexLogging {
  private val inboundConnectionCount = new AtomicInteger(0)
  private val perHostConnectionCount = new ConcurrentHashMap[InetAddress, Int]

  private def dec(remoteAddress: InetAddress) = {
    inboundConnectionCount.decrementAndGet()
    perHostConnectionCount.compute(remoteAddress, (_, cnt) => cnt - 1)
    null.asInstanceOf[ChannelFuture]
  }

  override def accept(ctx: ChannelHandlerContext, remoteAddress: InetSocketAddress) = {
    val newTotal = inboundConnectionCount.incrementAndGet()
    val newCountPerHost = perHostConnectionCount.compute(remoteAddress.getAddress, (_, cnt) => Option(cnt).fold(1)(_ + 1))
    val isBlacklisted = peerDatabase.blacklistedHosts.contains(remoteAddress.getAddress)

    log.trace(s"Check inbound connection from $remoteAddress: new inbound total = $newTotal, " +
      s"connections with this host = $newCountPerHost, address ${if (isBlacklisted) "IS" else "is not"} blacklisted")

    newTotal <= maxInboundConnections &&
      newCountPerHost <= maxConnectionsPerHost &&
      !isBlacklisted
  }

  override def channelAccepted(ctx: ChannelHandlerContext, remoteAddress: InetSocketAddress) =
    ctx.channel().closeFuture().addListener((_: ChannelFuture) => dec(remoteAddress.getAddress))

  override def channelRejected(ctx: ChannelHandlerContext, remoteAddress: InetSocketAddress) =
    dec(remoteAddress.getAddress)
} 

//设置package包名称以及导入依赖的类
package com.wavesplatform.network

import java.util

import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.ChannelHandlerContext
import io.netty.handler.codec.MessageToMessageCodec
import scorex.network.message._
import scorex.utils.ScorexLogging

@Sharable
class MessageCodec extends MessageToMessageCodec[RawBytes, Message] with ScorexLogging {

  private val specs: Map[Byte, MessageSpec[_ <: AnyRef]] = BasicMessagesRepo.specs.map(s => s.messageCode -> s).toMap

  override def encode(ctx: ChannelHandlerContext, msg: Message, out: util.List[AnyRef]) = msg match {
    case LocalScoreChanged(score) => out.add(RawBytes(ScoreMessageSpec.messageCode, ScoreMessageSpec.serializeData(score)))
    case GetPeers => out.add(RawBytes(GetPeersSpec.messageCode, Array[Byte]()))
    case k: KnownPeers => out.add(RawBytes(PeersSpec.messageCode, PeersSpec.serializeData(k)))
    case gs: GetSignatures => out.add(RawBytes(GetSignaturesSpec.messageCode, GetSignaturesSpec.serializeData(gs)))
    case s: Signatures => out.add(RawBytes(SignaturesSpec.messageCode, SignaturesSpec.serializeData(s)))
    case g: GetBlock => out.add(RawBytes(GetBlockSpec.messageCode, GetBlockSpec.serializeData(g)))
    case BlockForged(b) => out.add(RawBytes(BlockMessageSpec.messageCode, b.bytes))
    case r: RawBytes => out.add(r)
  }

  override def decode(ctx: ChannelHandlerContext, msg: RawBytes, out: util.List[AnyRef]) =
    out.add(specs(msg.code).deserializeData(msg.data).get)
} 

//设置package包名称以及导入依赖的类
package com.wavesplatform.network

import java.util.concurrent.Executors

import akka.dispatch.ExecutionContexts
import com.wavesplatform.UtxPool
import com.wavesplatform.state2.diffs.TransactionDiffer.TransactionValidationError
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.group.ChannelGroup
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}
import scorex.transaction.Transaction
import scorex.utils.ScorexLogging

import scala.concurrent.Future

@Sharable
class UtxPoolSynchronizer(utx: UtxPool, allChannels: ChannelGroup)
  extends ChannelInboundHandlerAdapter with ScorexLogging {

  private implicit val executor = ExecutionContexts.fromExecutor(Executors.newSingleThreadExecutor())

  override def channelRead(ctx: ChannelHandlerContext, msg: AnyRef): Unit = msg match {
    case t: Transaction => Future(utx.putIfNew(t) match {
      case Left(TransactionValidationError(e, _)) =>
        log.debug(s"${id(ctx)} Error processing transaction ${t.id}: $e")
      case Left(e) =>
        log.debug(s"${id(ctx)} Error processing transaction ${t.id}: $e")
      case Right(_) =>
        allChannels.broadcast(RawBytes(TransactionMessageSpec.messageCode, t.bytes), Some(ctx.channel()))
        log.trace(s"${id(ctx)} Added transaction ${t.id} to UTX pool")
    })
    case _ => super.channelRead(ctx, msg)
  }
} 

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

import io.netty.buffer.{ByteBuf, ByteBufHolder, EmptyByteBuf, Unpooled}
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}


@Sharable
object UnpoolHandler extends ChannelInboundHandlerAdapter {
  private[this] final def copyOnHeapAndRelease(bb: ByteBuf): ByteBuf = {
    try {
      if (bb.readableBytes > 0) Unpooled.buffer(bb.readableBytes, bb.capacity).writeBytes(bb)
      else Unpooled.EMPTY_BUFFER
    } finally {
      val _ = bb.release()
    }
  }

  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = msg match {
    case bb: ByteBuf =>
      val _ = ctx.fireChannelRead(copyOnHeapAndRelease(bb))

    // This case is special since it helps to avoid unnecessary `replace`
    // when the underlying content is already `EmptyByteBuffer`.
    case bbh: ByteBufHolder if bbh.content.isInstanceOf[EmptyByteBuf] =>
      val _ = ctx.fireChannelRead(bbh)

    case bbh: ByteBufHolder =>
      val onHeapContent = copyOnHeapAndRelease(bbh.content)
      val _ = ctx.fireChannelRead(bbh.replace(onHeapContent))

    case _ =>
      val _ = ctx.fireChannelRead(msg)
  }
} 

//设置package包名称以及导入依赖的类
package com.packt.masteringakka.bookstore.server

import io.netty.channel.ChannelHandler.Sharable
import unfiltered.netty.ServerErrorResponse
import unfiltered.request._
import java.util.Date
import org.json4s._
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}
import unfiltered.response._


@Sharable
object PretentCreditCardService extends unfiltered.netty.async.Plan with ServerErrorResponse{
  case class ChargeRequest(cardHolder:String, cardType:String, cardNumber:String, expiration:Date, amount:Double)
  case class ChargeResponse(confirmationCode:String)
  implicit val formats = Serialization.formats(NoTypeHints)
  
  def intent = {
    case req @ POST(Path(Seg("credit" :: "charge" :: Nil))) =>
      val body = Body.string(req)
      val request = read[ChargeRequest](body)
      val resp = write(ChargeResponse(java.util.UUID.randomUUID().toString))
      req.respond(JsonContent ~> ResponseString(resp))
  }
} 

//设置package包名称以及导入依赖的类
package com.packt.masteringakka.bookstore.common

import io.netty.channel.ChannelHandler.Sharable
import unfiltered.netty.ServerErrorResponse
import unfiltered.request._
import java.util.Date
import org.json4s._
import org.json4s.native.Serialization
import org.json4s.native.Serialization.{read, write}
import unfiltered.response._


@Sharable
object PretentCreditCardService extends unfiltered.netty.async.Plan with ServerErrorResponse{
  case class ChargeRequest(cardHolder:String, cardType:String, cardNumber:String, expiration:Date, amount:Double)
  case class ChargeResponse(confirmationCode:String)
  implicit val formats = Serialization.formats(NoTypeHints)
  
  def intent = {
    case req @ POST(Path(Seg("credit" :: "charge" :: Nil))) =>
      val body = Body.string(req)
      val request = read[ChargeRequest](body)
      val resp = write(ChargeResponse(java.util.UUID.randomUUID().toString))
      req.respond(JsonContent ~> ResponseString(resp))
  }
} 

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

import java.nio.charset.StandardCharsets

import com.trueaccord.scalapb.GeneratedMessage
import com.trueaccord.scalapb.json.JsonFormat
import io.grpc.ManagedChannel
import io.netty.buffer.Unpooled
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelFutureListener, ChannelHandlerContext, ChannelInboundHandlerAdapter}
import io.netty.handler.codec.http._

import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success}

@Sharable
abstract class GrpcGatewayHandler(channel: ManagedChannel)(implicit ec: ExecutionContext)
  extends ChannelInboundHandlerAdapter {

  def name: String

  def shutdown(): Unit =
    if (!channel.isShutdown) channel.shutdown()

  def unaryCall(method: HttpMethod, uri: String, body: String): Future[GeneratedMessage]

  override def channelRead(ctx: ChannelHandlerContext, msg: scala.Any): Unit = msg match {
    case req: FullHttpRequest =>
      val body = req.content().toString(StandardCharsets.UTF_8)
      unaryCall(req.method(), req.uri(), body)
        .map(JsonFormat.toJsonString)
        .map(_.getBytes(StandardCharsets.UTF_8))
        .onComplete {
          case Success(json) =>
            val res = new DefaultFullHttpResponse(
              req.protocolVersion(),
              HttpResponseStatus.OK,
              Unpooled.copiedBuffer(json)
            )
            res.headers().set(HttpHeaderNames.CONTENT_TYPE, "application/json")
            HttpUtil.setContentLength(res, json.length)
            HttpUtil.setKeepAlive(res, HttpUtil.isKeepAlive(req))
            ctx.writeAndFlush(res)

          case Failure(e) =>
            val status = e match {
              case _: UnsupportedOperationException => HttpResponseStatus.NOT_FOUND
              case _: NoSuchElementException => HttpResponseStatus.BAD_REQUEST
              case _ => HttpResponseStatus.INTERNAL_SERVER_ERROR
            }
            val res = new DefaultHttpResponse(req.protocolVersion(), status)
            ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE)
        }

    case _ => super.channelRead(ctx, msg)
  }
} 

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

import javax.activation.MimetypesFileTypeMap

import io.grpc.internal.IoUtils
import io.netty.buffer.Unpooled
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelFutureListener, ChannelHandlerContext, ChannelInboundHandlerAdapter}
import io.netty.handler.codec.http._

object SwaggerHandler {
  val SwaggerPath = "META-INF/resources/webjars/swagger-ui/3.0.10/"
  val DocsPrefix = "/docs/"
  val SpecsPrefix = "/specs/"
}

@Sharable
class SwaggerHandler extends ChannelInboundHandlerAdapter {
  import SwaggerHandler._
  override def channelRead(ctx: ChannelHandlerContext, msg: scala.Any): Unit = msg match {
    case req: FullHttpRequest =>
      val queryString = new QueryStringDecoder(req.uri())
      val path = queryString.path()

      if (path.startsWith(DocsPrefix)) {
        val filename = SwaggerPath + path.substring(DocsPrefix.length)
        writeResource(ctx, req, filename)
      } else if (path.startsWith(SpecsPrefix)) {
        val filename = path.substring(1) // remove heading slash
        writeResource(ctx, req, filename)
      } else super.channelRead(ctx, msg)
    case _ => super.channelRead(ctx, msg)
  }

  private def writeResource(ctx: ChannelHandlerContext, req: FullHttpRequest, filename: String): Unit = {
    Thread.currentThread().getContextClassLoader.getResourceAsStream(filename) match {
      case null =>
        val res = new DefaultFullHttpResponse(req.protocolVersion(), HttpResponseStatus.NOT_FOUND)
        ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE)
      case resource =>
        val bytes = IoUtils.toByteArray(resource)
        val res = new DefaultFullHttpResponse(
          req.protocolVersion(),
          HttpResponseStatus.OK,
          Unpooled.copiedBuffer(bytes)
        )
        res.headers().set(HttpHeaderNames.CONTENT_TYPE, new MimetypesFileTypeMap().getContentType(filename))
        HttpUtil.setContentLength(res, bytes.length)
        HttpUtil.setKeepAlive(res, HttpUtil.isKeepAlive(req))
        ctx.writeAndFlush(res)
    }
  }
} 

//设置package包名称以及导入依赖的类
package com.wavesplatform.network

import java.util.concurrent.atomic.AtomicBoolean

import com.wavesplatform.mining.Miner
import com.wavesplatform.settings.WavesSettings
import com.wavesplatform.state2.reader.StateReader
import com.wavesplatform.{Coordinator, UtxPool}
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.group.ChannelGroup
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}
import scorex.block.Block
import scorex.transaction._
import scorex.utils.{ScorexLogging, Time}

@Sharable
class CoordinatorHandler(checkpointService: CheckpointService, history: History, blockchainUpdater: BlockchainUpdater, time: Time,
                         stateReader: StateReader, utxStorage: UtxPool, blockchainReadiness: AtomicBoolean, miner: Miner,
                         settings: WavesSettings, peerDatabase: PeerDatabase, allChannels: ChannelGroup)
  extends ChannelInboundHandlerAdapter with ScorexLogging {

  import CoordinatorHandler._

  override def channelRead(ctx: ChannelHandlerContext, msg: AnyRef): Unit = msg match {
    case c: Checkpoint =>
      loggingResult(id(ctx), "Checkpoint",
        Coordinator.processCheckpoint(checkpointService, history, blockchainUpdater)(c))
        .fold(err => peerDatabase.blacklistAndClose(ctx.channel(), "Unable to process checkpoint due to " + err),
          score => allChannels.broadcast(LocalScoreChanged(score), Some(ctx.channel()))
        )
    case ExtensionBlocks(blocks) =>
      loggingResult(id(ctx), "ExtensionBlocks",
        Coordinator.processFork(checkpointService, history, blockchainUpdater, stateReader, utxStorage, time, settings, miner, blockchainReadiness)(blocks))
        .fold(
          err => peerDatabase.blacklistAndClose(ctx.channel(), "Unable to process ExtensionBlocks due to " + err),
          score => allChannels.broadcast(LocalScoreChanged(score))
        )
    case b: Block =>
      Signed.validateSignatures(b) match {
        case Right(_) =>
          loggingResult(id(ctx), "Block", Coordinator.processBlock(checkpointService, history, blockchainUpdater, time,
            stateReader, utxStorage, blockchainReadiness, miner, settings)(b, local = false))
            .foreach(score => allChannels.broadcast(LocalScoreChanged(score)))
        case Left(err) =>
          peerDatabase.blacklistAndClose(ctx.channel(), err.toString)
      }

  }
}

object CoordinatorHandler extends ScorexLogging {
  def loggingResult[R](idCtx: String, msg: String, f: => Either[ValidationError, R]): Either[ValidationError, R] = {
    log.debug(s"$idCtx Starting $msg processing")
    val result = f
    result match {
      case Left(error) => log.warn(s"$idCtx Error processing $msg: $error")
      case Right(newScore) => log.debug(s"$idCtx Finished $msg processing, new local score is $newScore")
    }
    result
  }
} 

//设置package包名称以及导入依赖的类
package com.wavesplatform.network

import java.util.concurrent.Executors

import akka.dispatch.ExecutionContexts
import com.wavesplatform.UtxPool
import com.wavesplatform.state2.diffs.TransactionDiffer.TransactionValidationError
import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.group.ChannelGroup
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}
import scorex.transaction.Transaction
import scorex.utils.ScorexLogging

import scala.concurrent.Future

@Sharable
class UtxPoolSynchronizer(utx: UtxPool)
  extends ChannelInboundHandlerAdapter with ScorexLogging {

  private implicit val executor = ExecutionContexts.fromExecutor(Executors.newSingleThreadExecutor())

  override def channelRead(ctx: ChannelHandlerContext, msg: AnyRef): Unit = msg match {
    case t: Transaction => Future(utx.putIfNew(t, Some(ctx.channel())) match {
      case Left(TransactionValidationError(e, _)) =>
        log.debug(s"${id(ctx)} Error processing transaction ${t.id}: $e")
      case Left(e) =>
        log.debug(s"${id(ctx)} Error processing transaction ${t.id}: $e")
      case Right(_) =>
        log.trace(s"${id(ctx)} Added transaction ${t.id} to UTX pool")
    })
    case _ => super.channelRead(ctx, msg)
  }
} 

//设置package包名称以及导入依赖的类
package com.wavesplatform.network

import io.netty.channel.ChannelHandler.Sharable
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}
import scorex.utils.ScorexLogging

import scala.concurrent.duration.FiniteDuration

@Sharable
class PeerSynchronizer(peerDatabase: PeerDatabase, peerRequestInterval: FiniteDuration) extends ChannelInboundHandlerAdapter
  with ScorexLogging {
  def requestPeers(ctx: ChannelHandlerContext): Unit = ctx.executor().schedule(peerRequestInterval) {
    if (ctx.channel().isActive) {
      log.trace(s"${id(ctx)} Requesting peers")
      ctx.writeAndFlush(GetPeers)
      requestPeers(ctx)
    }
  }

  override def channelActive(ctx: ChannelHandlerContext) = {
    requestPeers(ctx)
    super.channelActive(ctx)
  }

  override def channelRead(ctx: ChannelHandlerContext, msg: AnyRef) = msg match {
    case hs: Handshake =>
      hs.declaredAddress.foreach { declaredAddress =>
        peerDatabase.addCandidate(declaredAddress)
        ctx.channel().attr(AttributeKeys.DeclaredAddress).setIfAbsent(declaredAddress)
      }
      ctx.fireChannelRead(msg)
    case GetPeers =>
      ctx.channel().declaredAddress.foreach(peerDatabase.touch)
      ctx.writeAndFlush(KnownPeers(peerDatabase.knownPeers.keys.toSeq))
    case KnownPeers(peers) =>
      log.trace(s"${id(ctx)} Got known peers: ${peers.mkString("[", ", ", "]")}")
      ctx.channel().declaredAddress.foreach(peerDatabase.touch)
      peers.foreach(peerDatabase.addCandidate)
    case _ =>
      ctx.channel().declaredAddress.foreach(peerDatabase.touch)
      super.channelRead(ctx, msg)
  }
}