Scala Netty ByteToMessageDecoder 子类从输入中删除字节并传递到下一个管道级别答案

【问题标题】：Scala Netty ByteToMessageDecoder subclass remove bytes from input and pass to next pipeline levelScala Netty ByteToMessageDecoder 子类从输入中删除字节并传递到下一个管道级别
【发布时间】：2016-01-23 03:56:27
【问题描述】：

我正在尝试为 TCP 上接收的字节流实现一个 netty 解码器。

这是当前的实现：

import io.netty.handler.codec.ByteToMessageDecoder
import io.netty.channel.ChannelHandlerContext
import io.netty.buffer.ByteBuf
import io.netty.handler.codec.compression.{ZlibCodecFactory,ZlibWrapper}
import java.util.List

object MyCustomDecoder extends ByteToMessageDecoder {

  val GZIP_HEADER_1 = 0x1F
  val GZIP_HEADER_2 = 0x8B

  def decode(ctx: ChannelHandlerContext,in: ByteBuf,out: List[AnyRef]): Unit = {

    // Read values in Little Endian
    val blockTime = java.lang.Long.reverseBytes(in.readLong()) // little endian
    val blockSeq = java.lang.Integer.reverseBytes(in.readInt()) // little endian
    val blockSize = java.lang.Integer.reverseBytes(in.readInt()) // little endian

    // Checks if compressed block size matches block size published value
    if(in.readableBytes() - 16 == blockSize){

      val h1 = in.getUnsignedByte(in.readerIndex())
      val h2 = in.getUnsignedByte(in.readerIndex()+1)

      //
      if(isGzip(h1, h2))
        enableGzip(ctx)
      else 
        System.err.println("Not in GZip format")      
    }    

  }

  // Check if header matches GZIP for decompression
  private def isGzip(h1: Int, h2: Int):Boolean = h1 == GZIP_HEADER_1 && h2 == GZIP_HEADER_2

  private def enableGzip(ctx: ChannelHandlerContext): Unit = {
    val p = ctx.pipeline()
    //p.addLast("gzipdeflater", ZlibCodecFactory.newZlibEncoder(ZlibWrapper.GZIP))
    p.addLast("gzipinflater", ZlibCodecFactory.newZlibDecoder(ZlibWrapper.GZIP))
    //p.remove(this)
  }

}

解码器应该使用 netty 实现来接收消息，验证消息中的数据是否与消息头中发布的实际块大小匹配，然后发送到管道的下一级膨胀压缩内容使用 GZip。

我当前的实现能够读取数据，但我不确定如何从传递到下一个管道级别的输出中删除消息头字节（即：blockTime、blockSeq、@987654325 @ ) 以便获得解压缩的可行消息，然后传递给结束消息处理程序。

我的数据定义如下： <blockTime><blockSequence><blockSize><block>

我基于this example 为netty 管道创建了这个解码器子类。

对此的任何帮助将不胜感激。

谢谢。

【问题讨论】：

标签： scala netty decoder

【解决方案1】：

有两个步骤可以执行：

从ByteBuf对象中删除字节（in）

这已经在示例中通过执行以下操作来实现（read[Type]() 实质上是移动 ByteBuf 索引指针）：

val blockTime = java.lang.Long.reverseBytes(in.readLong()) // little endian
val blockSeq = java.lang.Integer.reverseBytes(in.readInt()) // little endian
val blockSize = java.lang.Integer.reverseBytes(in.readInt()) // little endian

此外，可以通过发出以下语句从 ByteBuf 对象中删除之前的（“使用的”）字节：in.discardReadBytes()

将消息传递到下一个级别

假设管道中的下一个解码器/处理程序将一个 ByteBuf 作为输入（例如另一个 ByteToMessageDecoder 子类），这简单地实现如下：

out.add(in)

鉴于已经在 (1) 中删除了标头字节，传递到下一个管道级别的是 ByteBuf 对象表示的剩余字节。

然后处理将根据初始化程序中定义的管道结构继续进行。

【讨论】：