org.jboss.netty.handler.codec.replay
类 ReplayingDecoder<T extends java.lang.Enum<T>>

java.lang.Object
  org.jboss.netty.channel.SimpleChannelUpstreamHandler
      org.jboss.netty.handler.codec.replay.ReplayingDecoder<T>

类型参数：

T - the state type; use VoidEnum if state management is unused

所有已实现的接口：

ChannelHandler, ChannelUpstreamHandler

直接已知子类：

CompatibleObjectDecoder, HttpMessageDecoder, WebSocketFrameDecoder

public abstract class ReplayingDecoder<T extends java.lang.Enum<T>>
extends SimpleChannelUpstreamHandler

A specialized variation of FrameDecoder which enables implementation of a non-blocking decoder in the blocking I/O paradigm.

The biggest difference between ReplayingDecoder and FrameDecoder is that ReplayingDecoder allows you to implement the decode() and decodeLast() methods just like all required bytes were received already, rather than checking the availability of the required bytes. For example, the following FrameDecoder implementation:

 public class IntegerHeaderFrameDecoder extends FrameDecoder {

   @Override
   protected Object decode(ChannelHandlerContext ctx,
                           Channel channel,
                           ChannelBuffer buf) throws Exception {

     if (buf.readableBytes() < 4) {
        return null;
     }

     buf.markReaderIndex();
     int length = buf.readInt();

     if (buf.readableBytes() < length) {
        buf.resetReaderIndex();
        return null;
     }

     return buf.readBytes(length);
   }
 }
 

is simplified like the following with ReplayingDecoder:

 public class IntegerHeaderFrameDecoder
      extends ReplayingDecoder<VoidEnum> {

   protected Object decode(ChannelHandlerContext ctx,
                           Channel channel,
                           ChannelBuffer buf,
                           VoidEnum state) throws Exception {

     return buf.readBytes(buf.readInt());
   }
 }
 

How does this work?

ReplayingDecoder passes a specialized ChannelBuffer implementation which throws an Error of certain type when there's not enough data in the buffer. In the IntegerHeaderFrameDecoder above, you just assumed that there will be 4 or more bytes in the buffer when you call buf.readInt(). If there's really 4 bytes in the buffer, it will return the integer header as you expected. Otherwise, the Error will be raised and the control will be returned to ReplayingDecoder. If ReplayingDecoder catches the Error, then it will rewind the readerIndex of the buffer back to the 'initial' position (i.e. the beginning of the buffer) and call the decode(..) method again when more data is received into the buffer.

Please note that the overhead of throwing an Error is minimal unlike throwing a new Exception in an ordinary way. ReplayingDecoder reuses the same Error instance so that it does not need to fill its stack trace, which takes most of Exception initialization time.

Limitations

At the cost of the simplicity, ReplayingDecoder enforces you a few limitations:

• Some buffer operations are prohibited.
• Performance can be worse if the network is slow and the message format is complicated unlike the example above. In this case, your decoder might have to decode the same part of the message over and over again.
• You must keep in mind that decode(..) method can be called many times to decode a single message. For example, the following code will not work:

   public class MyDecoder extends ReplayingDecoder<VoidEnum> {
  
     private final Queue<Integer> values = new LinkedList<Integer>();
  
     @Override
     public Object decode(.., ChannelBuffer buffer, ..) throws Exception {
  
       // A message contains 2 integers.
       values.offer(buffer.readInt());
       values.offer(buffer.readInt());
  
       // This assertion will fail intermittently since values.offer()
       // can be called more than two times!
       assert values.size() == 2;
       return values.poll() + values.poll();
     }
   }

  The correct implementation looks like the following, and you can also utilize the 'checkpoint' feature which is explained in detail in the next section.

   public class MyDecoder extends ReplayingDecoder<VoidEnum> {
  
     private final Queue<Integer> values = new LinkedList<Integer>();
  
     @Override
     public Object decode(.., ChannelBuffer buffer, ..) throws Exception {
  
       // Revert the state of the variable that might have been changed
       // since the last partial decode.
       values.clear();
  
       // A message contains 2 integers.
       values.offer(buffer.readInt());
       values.offer(buffer.readInt());
  
       // Now we know this assertion will never fail.
       assert values.size() == 2;
       return values.poll() + values.poll();
     }
   }

Improving the performance

Fortunately, the performance of a complex decoder implementation can be improved significantly with the checkpoint() method. The checkpoint() method updates the 'initial' position of the buffer so that ReplayingDecoder rewinds the readerIndex of the buffer to the last position where you called the checkpoint() method.

Calling checkpoint(T) with an Enum

Although you can just use checkpoint() method and manage the state of the decoder by yourself, the easiest way to manage the state of the decoder is to create an Enum type which represents the current state of the decoder and to call checkpoint(T) method whenever the state changes. You can have as many states as you want depending on the complexity of the message you want to decode:

 public enum MyDecoderState {
   READ_LENGTH,
   READ_CONTENT;
 }

 public class IntegerHeaderFrameDecoder
      extends ReplayingDecoder<MyDecoderState> {

   private int length;

   public IntegerHeaderFrameDecoder() {
     // Set the initial state.
     super(MyDecoderState.READ_LENGTH);
   }

   @Override
   protected Object decode(ChannelHandlerContext ctx,
                           Channel channel,
                           ChannelBuffer buf,
                           MyDecoderState state) throws Exception {
     switch (state) {
     case READ_LENGTH:
       length = buf.readInt();
       checkpoint(MyDecoderState.READ_CONTENT);
     case READ_CONTENT:
       ChannelBuffer frame = buf.readBytes(length);
       checkpoint(MyDecoderState.READ_LENGTH);
       return frame;
     default:
       throw new Error("Shouldn't reach here.");
     }
   }
 }
 

Calling checkpoint() with no parameter

An alternative way to manage the decoder state is to manage it by yourself.

 public class IntegerHeaderFrameDecoder
      extends ReplayingDecoder<VoidEnum> {

   private boolean readLength;
   private int length;

   @Override
   protected Object decode(ChannelHandlerContext ctx,
                           Channel channel,
                           ChannelBuffer buf,
                           VoidEnum state) throws Exception {
     if (!readLength) {
       length = buf.readInt();
       readLength = true;
       checkpoint();
     }

     if (readLength) {
       ChannelBuffer frame = buf.readBytes(length);
       readLength = false;
       checkpoint();
       return frame;
     }
   }
 }
 

Replacing a decoder with another decoder in a pipeline

If you are going to write a protocol multiplexer, you will probably want to replace a ReplayingDecoder (protocol detector) with another ReplayingDecoder or FrameDecoder (actual protocol decoder). It is not possible to achieve this simply by calling ChannelPipeline.replace(ChannelHandler, String, ChannelHandler), but some additional steps are required:

 public class FirstDecoder extends ReplayingDecoder<VoidEnum> {

     public FirstDecoder() {
         super(true); // Enable unfold
     }

     @Override
     protected Object decode(ChannelHandlerContext ctx,
                             Channel ch,
                             ChannelBuffer buf,
                             VoidEnum state) {
         ...
         // Decode the first message
         Object firstMessage = ...;

         // Add the second decoder
         ctx.getPipeline().addLast("second", new SecondDecoder());

         // Remove the first decoder (me)
         ctx.getPipeline().remove(this);

         if (buf.readable()) {
             // Hand off the remaining data to the second decoder
             return new Object[] { firstMessage, buf.readBytes(super.actualReadableBytes()) };
         } else {
             // Nothing to hand off
             return firstMessage;
         }
     }
 

嵌套类摘要

从接口 org.jboss.netty.channel.ChannelHandler 继承的嵌套类/接口

ChannelHandler.Sharable

方法摘要

void	channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) 当一个Channel被关闭且它所有关联的资源被释放时调用.
void	channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) 当一个Channel被远程端断开连接时调用.
void	exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) 当一个I/O线程或ChannelHandler抛出异常时被调用.
void	messageReceived(ChannelHandlerContext ctx, MessageEvent e) 当一个从远端发来的消息对象(如: ChannelBuffer)被接收时调用.

从类 org.jboss.netty.channel.SimpleChannelUpstreamHandler 继承的方法

channelBound, channelConnected, channelInterestChanged, channelOpen, channelUnbound, childChannelClosed, childChannelOpen, handleUpstream, writeComplete

从类 java.lang.Object 继承的方法

equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

方法详细信息

messageReceived

public void messageReceived(ChannelHandlerContext ctx,
                            MessageEvent e)
                     throws java.lang.Exception

从类 SimpleChannelUpstreamHandler 复制的描述

当一个从远端发来的消息对象(如: ChannelBuffer)被接收时调用.

覆盖：

类 SimpleChannelUpstreamHandler 中的 messageReceived

抛出：

java.lang.Exception

channelDisconnected

public void channelDisconnected(ChannelHandlerContext ctx,
                                ChannelStateEvent e)
                         throws java.lang.Exception

从类 SimpleChannelUpstreamHandler 复制的描述

当一个Channel被远程端断开连接时调用.

覆盖：

类 SimpleChannelUpstreamHandler 中的 channelDisconnected

抛出：

java.lang.Exception

channelClosed

public void channelClosed(ChannelHandlerContext ctx,
                          ChannelStateEvent e)
                   throws java.lang.Exception

从类 SimpleChannelUpstreamHandler 复制的描述

当一个Channel被关闭且它所有关联的资源被释放时调用.

覆盖：

类 SimpleChannelUpstreamHandler 中的 channelClosed

抛出：

java.lang.Exception

exceptionCaught

public void exceptionCaught(ChannelHandlerContext ctx,
                            ExceptionEvent e)
                     throws java.lang.Exception

从类 SimpleChannelUpstreamHandler 复制的描述

当一个I/O线程或ChannelHandler抛出异常时被调用.

覆盖：

类 SimpleChannelUpstreamHandler 中的 exceptionCaught

抛出：

java.lang.Exception