//设置package包名称以及导入依赖的类
package elo.scala.web

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


class DiscardServerHandler extends ChannelInboundHandlerAdapter { // (1)

  override def channelRead(ctx: ChannelHandlerContext, msg: AnyRef): Unit = { // (2)
    // Discard the received data silently.
//    ((ByteBuf) msg).release(); // (3)
    msg.asInstanceOf[ByteBuf].release()
  }

  override def exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) { // (4)
    // Close the connection when an exception is raised.
    cause.printStackTrace()
    ctx.close()
  }
} 

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

import com.twitter.finagle.netty4.http.exp.initClient
import com.twitter.finagle.netty4.transport.ChannelTransport
import com.twitter.finagle.Stack
import io.netty.channel.{ChannelInboundHandlerAdapter, ChannelHandlerContext}
import io.netty.handler.codec.http.HttpClientUpgradeHandler.UpgradeEvent
import scala.collection.JavaConverters.iterableAsScalaIterableConverter


private[http2] class UpgradeRequestHandler(
    params: Stack.Params)
  extends ChannelInboundHandlerAdapter {

  override def userEventTriggered(ctx: ChannelHandlerContext, event: Any): Unit = {
    event match {
      case [email protected]_REJECTED =>
        ctx.fireChannelRead(rejected)
        // disable autoread if we fail the upgrade
        ctx.channel.config.setAutoRead(false)
        ctx.pipeline.remove(this)
      case [email protected]_SUCCESSFUL =>
        val p = ctx.pipeline
        p.asScala
          .toList
          .dropWhile(_.getKey != UpgradeRequestHandler.HandlerName)
          .tail
          .takeWhile(_.getKey != ChannelTransport.HandlerName)
          .foreach { entry =>
            p.remove(entry.getValue)
          }
        p.addBefore(
          ChannelTransport.HandlerName,
          "aggregate",
          new AdapterProxyChannelHandler({ pipeline =>
            pipeline.addLast("schemifier", new SchemifyingHandler("http"))
            initClient(params)(pipeline)
          })
        )
        ctx.fireChannelRead(successful)
        ctx.pipeline.remove(this)
      case _ => // nop
    }
    super.userEventTriggered(ctx, event)
  }
}

private[http2] object UpgradeRequestHandler {
  val HandlerName = "pipelineUpgrader"
} 

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

import com.twitter.finagle.Stack
import com.twitter.finagle.netty4.http.exp._
import io.netty.channel.{Channel, ChannelHandlerContext, ChannelInitializer}
import io.netty.handler.codec.http2.{Http2Codec, Http2ServerDowngrader}
import io.netty.handler.ssl.{ApplicationProtocolNames, ApplicationProtocolNegotiationHandler}

private[http2] class NpnOrAlpnHandler(init: ChannelInitializer[Channel], params: Stack.Params)
  extends ApplicationProtocolNegotiationHandler(ApplicationProtocolNames.HTTP_1_1) {

  @throws(classOf[Exception])
  protected def configurePipeline(ctx: ChannelHandlerContext, protocol: String) {

    protocol match {
      case ApplicationProtocolNames.HTTP_2 =>
        // Http2 has been negotiated, replace the HttpCodec with an Http2Codec
        val initializer = new ChannelInitializer[Channel] {
          def initChannel(ch: Channel): Unit = {
            ch.pipeline.addLast(new Http2ServerDowngrader(false  , initializer))

      case ApplicationProtocolNames.HTTP_1_1 =>
      // The Http codec is already in the pipeline, so we are good!
      case _ =>
        throw new IllegalStateException("unknown protocol: " + protocol)
    }
  }
} 

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

import com.twitter.finagle.http.Fields
import com.twitter.util.StorageUnit
import io.netty.buffer.ByteBufHolder
import io.netty.channel.{ChannelFutureListener, ChannelHandlerContext, ChannelInboundHandlerAdapter}
import io.netty.handler.codec.http._
import java.util.logging.{Level, Logger}

private object PayloadSizeHandler {
  def mk413(v: HttpVersion): FullHttpResponse = {
    val resp = new DefaultFullHttpResponse(v, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE)
    resp.headers().set(Fields.Connection, "close")
    resp.headers().set(Fields.ContentLength, "0")
    resp
  }
}


private[http] class PayloadSizeHandler(limit: StorageUnit, log: Option[Logger])
  extends ChannelInboundHandlerAdapter {

  def this(limit: StorageUnit) = this(limit, None)

  private[this] val limitBytes = limit.inBytes

  // we don't worry about thread-safety because netty guarantees that reads are
  // serialized and on the same-thread.
  private[this] var discarding = false

  import PayloadSizeHandler.mk413

  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = msg match {
    case http: HttpMessage if HttpUtil.getContentLength(http, -1) > limitBytes =>
      discarding = true
      if (http.isInstanceOf[ByteBufHolder]) {
        http.asInstanceOf[ByteBufHolder].release()
      }
      ctx.writeAndFlush(mk413(http.protocolVersion))
        .addListener(ChannelFutureListener.CLOSE)

      log match {
        case Some(l) if l.isLoggable(Level.FINE) =>
          l.log(Level.FINE, s"rejected an oversize payload (${HttpUtil.getContentLength(http)} bytes) from ${ctx.channel.remoteAddress}")
        case _ =>
      }

    // the session is doomed so we reject chunks
    case chunk: HttpContent if discarding =>
      chunk.release()

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

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

import com.twitter.conversions.storage._
import com.twitter.finagle.ChannelBufferUsageException
import com.twitter.finagle.http.codec.ChannelBufferUsageTracker
import io.netty.buffer.Unpooled
import io.netty.channel.{ChannelPromise, ChannelHandlerContext}
import org.junit.runner.RunWith
import org.scalatest.FunSuite
import org.scalatest.junit.JUnitRunner
import org.scalatest.mock.MockitoSugar

@RunWith(classOf[JUnitRunner])
class ByteBufManagerTest extends FunSuite with MockitoSugar {

  def usageTrackerFactory() = {
    val usageTracker = new ChannelBufferUsageTracker(1000.bytes)
    assert(usageTracker.usageLimit == (1000.bytes))
    usageTracker
  }

  val ctx = mock[ChannelHandlerContext]
  val p = mock[ChannelPromise]

  test("tracks buffer usage between writes") {
    val tracker = usageTrackerFactory()
    val handler = new ByteBufManager(tracker)

    handler.channelRead(ctx, Unpooled.buffer(100))
    assert(tracker.currentUsage == 100.bytes)
    handler.channelRead(ctx, Unpooled.buffer(50))
    assert(tracker.currentUsage == 150.bytes)

    handler.write(ctx, Unpooled.EMPTY_BUFFER, p)
    assert(tracker.currentUsage == 0.bytes)

    handler.channelRead(ctx, Unpooled.buffer(123))
    assert(tracker.currentUsage == 123.bytes)
    handler.close(ctx, p)
    assert(tracker.currentUsage == 0.bytes)
  }

  test("throws if aggregate limit is exceeded") {
    val tracker = usageTrackerFactory()
    val handler = new ByteBufManager(tracker)

    handler.channelRead(ctx, Unpooled.buffer(1000))

    intercept[ChannelBufferUsageException] {
      handler.channelRead(ctx, Unpooled.buffer(1))
    }
  }
} 

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

import com.twitter.finagle.Failure
import com.twitter.finagle.framer.Framer
import com.twitter.io.Buf
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}


private[finagle] class FrameHandler(framer: Framer) extends ChannelInboundHandlerAdapter {
  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = msg match {
    case buf: Buf =>
      var idx = 0
      val frames = framer(buf)
      while (idx < frames.length) {
        ctx.fireChannelRead(frames(idx))
        idx += 1
      }

    case _ =>
      ctx.fireExceptionCaught(Failure(
        s"FrameHandler saw non-Buf message: ${msg.toString}"))
  }
} 

//设置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.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.twitter.finagle.netty4.codec

import com.twitter.finagle.Failure
import com.twitter.finagle.netty4.{BufAsByteBuf, ByteBufAsBuf}
import com.twitter.io.Buf
import io.netty.buffer.ByteBuf
import io.netty.channel.{ChannelPromise, ChannelHandlerContext, ChannelDuplexHandler}


private[finagle] class BufCodec extends ChannelDuplexHandler {
  override def write(ctx: ChannelHandlerContext, msg: Any, p: ChannelPromise): Unit =
    msg match {
      case buf: Buf => ctx.write(BufAsByteBuf.Owned(buf), p)
      case typ => p.setFailure(Failure(
        s"unexpected type ${typ.getClass.getSimpleName} when encoding to ByteBuf"))
    }

  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit =
    msg match {
      case bb: ByteBuf => ctx.fireChannelRead(ByteBufAsBuf.Owned(bb))
      case typ => ctx.fireExceptionCaught(Failure(
          s"unexpected type ${typ.getClass.getSimpleName} when encoding to Buf"))
    }
} 

//设置package包名称以及导入依赖的类
package tk.dasb.handler

import io.netty.channel.{ChannelFuture, ChannelFutureListener, ChannelHandlerContext, ChannelInboundHandlerAdapter}
import tk.dasb.codec.{InitReplyEncoder, InitReqDecoder, ReplyEncoder, ReqDecoder}
import tk.dasb.protocol.{InitReply, InitReq}
import tk.dasb.util.{FutureConv, Log}


class InitReqHandler extends ChannelInboundHandlerAdapter with Log with FutureConv {
  override def channelRead(ctx: ChannelHandlerContext, msg: scala.Any): Unit = {
    val req = msg.asInstanceOf[InitReq]
    log.debug("receive init req:{}",req)
    if (req.ver == 0x05 && req.methods == 0x00) {
      ctx.writeAndFlush(InitReply(0x05, 0x00)).success {
        writeFuture=>
          ctx.pipeline.remove(classOf[InitReqDecoder])
          ctx.pipeline.remove(classOf[InitReplyEncoder])
          ctx.pipeline.remove(InitReqHandler.this)
          ctx.pipeline.addLast(new ReqDecoder)
          ctx.pipeline.addLast(new ReplyEncoder)
          ctx.pipeline.addLast(new ReqHandler)
      }.failed{
        writeFuture=>
          ctx.channel.close()
          log.info("unsupport {}", req)
      }
    }
  }
} 

//设置package包名称以及导入依赖的类
package tk.dasb.handler

import java.net.InetSocketAddress

import io.netty.buffer.{ByteBuf, Unpooled}
import io.netty.channel.socket.DatagramPacket
import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}
import io.netty.util.ByteProcessor
import tk.dasb.protocol.UdpReq
import tk.dasb.util.{FutureConv, Log}


class UdpPipHandler(var clientAddress:InetSocketAddress) extends ChannelInboundHandlerAdapter with Log with FutureConv {


  def isClientAddressInValid() :Boolean =  clientAddress == null || clientAddress.getPort == 0


  def procClientToProxy(ctx: ChannelHandlerContext, msg:UdpReq) = {
    val newMessage = new DatagramPacket(Unpooled.wrappedBuffer(msg.data),msg.dstAddress)
    ctx.writeAndFlush(newMessage)
    if(isClientAddressInValid){
      clientAddress = msg.raw.sender
      log.debug("reset client address:{}",clientAddress)
    }
  }


  def procRemoteToProxy(ctx: ChannelHandlerContext, msg:DatagramPacket) = {

    if(!isClientAddressInValid ){
      val msgHead = ctx.alloc.buffer(16,32)
      msgHead.writeShort(0).writeByte(0).writeByte(1)
      msgHead.writeBytes(msg.sender.getAddress.getAddress).writeShort(msg.sender.getPort)
      val newMessage = new DatagramPacket(Unpooled.wrappedBuffer(msgHead,msg.content),clientAddress)
      log.debug("send back {}",newMessage.content.readableBytes())
      ctx.writeAndFlush(newMessage)

    }else{
      log.warn("unknow client address,drop messsage")
      msg.release()
    }
  }

  override def channelRead(ctx: ChannelHandlerContext, msg: scala.Any): Unit = msg match {
      case udp:UdpReq => procClientToProxy(ctx,udp)
      case udp:DatagramPacket => procRemoteToProxy(ctx,udp)
  }

} 

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

import io.netty.buffer.ByteBufHolder
import io.netty.channel.{ ChannelHandlerContext, ChannelInboundHandlerAdapter }

// Need to retain the ByteBuf before it's forwarded to a Twitter Future
private[finagle] object ContentRetainer extends ChannelInboundHandlerAdapter {
  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit =
    msg match {
      case bb: ByteBufHolder =>
        super.channelRead(ctx, bb.retain())
      case other =>
        super.channelRead(ctx, other)
    }
} 

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

import io.netty.buffer.ByteBuf
import io.netty.channel.ChannelHandlerContext
import io.netty.handler.codec.{ByteToMessageDecoder, MessageToByteEncoder}


object LoginCodec {
  case class LoginRequest(username: String, password: String)
  case class LoginResponse(responseCode: Int)

  type Ctx = ChannelHandlerContext
  type Buffer = ByteBuf
  type Output = java.util.List[AnyRef]

  final class LoginRequestEncoder extends MessageToByteEncoder[LoginRequest] {
    override def encode(ctx: Ctx, msg: LoginRequest, out: Buffer): Unit = {
      out.writeCString(msg.username)
      out.writeCString(msg.password)
    }
  }

  final class LoginResponseDecoder extends ByteToMessageDecoder {
    override def decode(ctx: Ctx, in: Buffer, out: Output): Unit = {
      (in.readableBytes() compare 1) signum match {
        case 0 | 1 => out.add(LoginResponse(in.readUnsignedByte()))
        case -1 => // nothing
      }
    }
  }
} 

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

import java.util.concurrent.ScheduledFuture

import com.wavesplatform.state2.ByteStr
import io.netty.channel.{ChannelDuplexHandler, ChannelHandlerContext, ChannelPromise}
import scorex.utils.ScorexLogging

import scala.concurrent.duration.FiniteDuration

class ExtensionSignaturesLoader(syncTimeout: FiniteDuration, peerDatabase: PeerDatabase)
  extends ChannelDuplexHandler with ScorexLogging {

  private var currentTimeout = Option.empty[ScheduledFuture[Unit]]
  private var lastKnownSignatures = Seq.empty[ByteStr]

  override def channelRead(ctx: ChannelHandlerContext, msg: AnyRef) = msg match {
    case s: Signatures =>
      val (known, unknown) = s.signatures.span(id => lastKnownSignatures.contains(id))
      currentTimeout.foreach(_.cancel(true))
      currentTimeout = None
      known.lastOption.foreach { lastKnown =>
        log.debug(s"${id(ctx)} Got extension with ${known.length}/${s.signatures.length} known signatures")
        ctx.fireChannelRead(ExtensionIds(lastKnown, unknown))
      }
    case _ => super.channelRead(ctx, msg)
  }

  override def channelInactive(ctx: ChannelHandlerContext) = {
    currentTimeout.foreach(_.cancel(false))
    currentTimeout = None
  }

  override def write(ctx: ChannelHandlerContext, msg: AnyRef, promise: ChannelPromise) = msg match {
    case LoadBlockchainExtension(sigs) if currentTimeout.isEmpty =>
      lastKnownSignatures = sigs

      log.debug(s"${id(ctx)} Loading extension, last ${sigs.length} are ${formatSignatures(sigs)}")

      currentTimeout = Some(ctx.executor().schedule(syncTimeout) {
        if (currentTimeout.nonEmpty && ctx.channel().isActive) {
          peerDatabase.blacklistAndClose(ctx.channel(),"Timeout expired while loading extension")
        }
      })

      ctx.writeAndFlush(GetSignatures(sigs), promise)

    case LoadBlockchainExtension(_) =>
      log.debug(s"${id(ctx)} Received request to load signatures while waiting for extension, ignoring for now")
      promise.setSuccess()

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

//设置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)
  }
}