package xmg.quest.netty.core;
/**
* @author 作者 : xuminggang
* @version 创建时间：2020年6月8日 上午10:05:15
* 
*/
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.DecoderException;
import io.netty.util.Signal;
import io.netty.util.internal.StringUtil;import java.util.List;/*** {@link ByteToMessageDecoder}的一个特殊变体，在阻塞I/O范示中可以实现一种非阻塞的解码器。* <p>* {@link ReplayingDecoder} 和 {@link ReplayingDecoder} 的最大区别在{@link ReplayingDecoder}允许* 	你实现 {@code decode()} 和 {@code decodeLast()} 方法将像所有需要的字节都已经被接收到了一样，而不必去检查所需* 	字节是不是可用或存在。* 	举例，下面是 {@link ByteToMessageDecoder} 的实现。* <pre>* public class IntegerHeaderFrameDecoder extends {@link ByteToMessageDecoder} {**   {@code @Override}*   protected void decode({@link ChannelHandlerContext} ctx,*                           {@link ByteBuf} buf, List&lt;Object&gt; out) throws Exception {**     if (buf.readableBytes() &lt; 4) {*        return;*     }**     buf.markReaderIndex();*     int length = buf.readInt();**     if (buf.readableBytes() &lt; length) {*        buf.resetReaderIndex();*        return;*     }**     out.add(buf.readBytes(length));*   }* }* </pre>* 	下面使用 {@link ReplayingDecoder} 简化了实现:* <pre>* public class IntegerHeaderFrameDecoder*      extends {@link ReplayingDecoder}&lt;{@link Void}&gt; {**   protected void decode({@link ChannelHandlerContext} ctx,*                           {@link ByteBuf} buf) throws Exception {**     out.add(buf.readBytes(buf.readInt()));*   }* }* </pre>** <h3>它是怎么工作的?</h3>* <p>* {@link ReplayingDecoder} 传递了一个特殊的 {@link ByteBuf} 的实现，当缓冲中没有足够的数据* 	时候就来抛出一个 确定类型的 {@link Error}。 在上面的 {@code IntegerHeaderFrameDecoder}例子中* 	当调用 {@code buf.readInt()}你只是假设buffer中有4个或4个以上字节，如果buffer中真的有4个* 	字节，它将返回你期待的integer头。否则，就会抛出一个{@link Error}并且 控制会返回给 {@link ReplayingDecoder}。* 	如果 {@link ReplayingDecoder} 捕获到这个 {@link Error}，就会将 buffer的 {@code readerIndex} 重置为* 	‘初始的’位置（如，buffer的起始位置）并且当buffer收到数据时，再次调用 {@code decode(..)}方法。* <p>*	请注意 {@link ReplayingDecoder}总是抛出相同的缓冲 {@link Error}实例，从而来避免创建一个新 {@link Error}的负担*	并每次抛出时都添加堆栈信息。** <h3>限制</h3>* <p>* 	以简洁为代价，{@link ReplayingDecoder} 施加了几个限制：* <ul>* <li>某些buffer操作是被禁止的。</li>* <li>如果网络很慢而且消息格式很复杂，性能可能会很差。在这种情况下，你的解码器不得不周而复始的解码消息相同的部分。</li>* <li>你必须记住 {@code decode(..)} 方法会调用很多次来解码一个单独的消息。例如，下面的代码就可能不能正常使用：* <pre> public class MyDecoder extends {@link ReplayingDecoder}&lt;{@link Void}&gt; {**   private final Queue&lt;Integer&gt; values = new LinkedList&lt;Integer&gt;();**   {@code @Override}*   public void decode(.., {@link ByteBuf} buf, List&lt;Object&gt; out) throws Exception {**     // 一个消息包含了两个integers.*     values.offer(buf.readInt());*     values.offer(buf.readInt());**     // 这个断言可能会间歇性的失败， 因为values.offer() 可能会被调用两次以上*     assert values.size() == 2;*     out.add(values.poll() + values.poll());*   }* }</pre>*      	正确的实现是下面这个样子，你还可以使用检查点的方式来去实现，在下一段中会详细解释。* <pre> public class MyDecoder extends {@link ReplayingDecoder}&lt;{@link Void}&gt; {**   private final Queue&lt;Integer&gt; values = new LinkedList&lt;Integer&gt;();**   {@code @Override}*   public void decode(.., {@link ByteBuf} buf, List&lt;Object&gt; out) throws Exception {**     // 重置变量的状态，因为最后一次阶段解码可能会改变变量状态*     values.clear();**     // 一个消息包含了两个integers.*     values.offer(buf.readInt());*     values.offer(buf.readInt());**     // 现在我们知道这个断言永不会失败。*     assert values.size() == 2;*     out.add(values.poll() + values.poll());*   }* }</pre>*     </li>* </ul>** <h3>性能改进</h3>* <p>* 	幸好，一个复杂的解码器实现的性能可以借助 {@code checkpoint()}的方法来改进。 {@code checkpoint()} * 	这个方法会更新buffer的初始位置，这样 当，{@link ReplayingDecoder} 就可以重新定位buffer的* 	{@code readerIndex} 到你上一次调用{@code checkpoint()}方法时的位置。** <h4>使用一个{@link Enum} 调用 {@code checkpoint(T)} </h4>* <p>* 	尽管你自己可以仅仅使用 {@code checkpoint()} 方法来管理解码器的状态，但是最简单的方式来管理解码器的状态是* 	创建一个 {@link Enum} 类型来代表当前解码器的状态，并可以在状态改变时调用 {@code checkpoint(T)} 方法。* 	你可以根据自己需要设置任意多的状态，这个取决于你想要解码的消息的复杂性：** <pre>* public enum MyDecoderState {*   READ_LENGTH,*   READ_CONTENT;* }** public class IntegerHeaderFrameDecoder*      extends {@link ReplayingDecoder}&lt;<strong>MyDecoderState</strong>&gt; {**   private int length;**   public IntegerHeaderFrameDecoder() {*     // 设置初始状态.*     <strong>super(MyDecoderState.READ_LENGTH);</strong>*   }**   {@code @Override}*   protected void decode({@link ChannelHandlerContext} ctx,*                           {@link ByteBuf} buf, List&lt;Object&gt; out) throws Exception {*     switch (state()) {*     case READ_LENGTH:*       length = buf.readInt();*       <strong>checkpoint(MyDecoderState.READ_CONTENT);</strong>*     case READ_CONTENT:*       ByteBuf frame = buf.readBytes(length);*       <strong>checkpoint(MyDecoderState.READ_LENGTH);</strong>*       out.add(frame);*       break;*     default:*       throw new Error("Shouldn't reach here.");*     }*   }* }* </pre>** <h4>调用不带参数的 {@code checkpoint()} </h4>* <p>* 	另外一种管理解码器状态的方式是你自己管理。* <pre>* public class IntegerHeaderFrameDecoder*      extends {@link ReplayingDecoder}&lt;<strong>{@link Void}</strong>&gt; {**   <strong>private boolean readLength;</strong>*   private int length;**   {@code @Override}*   protected void decode({@link ChannelHandlerContext} ctx,*                           {@link ByteBuf} buf, List&lt;Object&gt; out) throws Exception {*     if (!readLength) {*       length = buf.readInt();*       <strong>readLength = true;</strong>*       <strong>checkpoint();</strong>*     }**     if (readLength) {*       ByteBuf frame = buf.readBytes(length);*       <strong>readLength = false;</strong>*       <strong>checkpoint();</strong>*       out.add(frame);*     }*   }* }* </pre>** <h3>用管道里的另一个解码器替换一个解码器</h3>* <p>* 	如果你打算编写一个协议的多路复用器，你可能想要用另外一个替换一个 {@link ReplayingDecoder}，* 	{@link ByteToMessageDecoder} 或者 {@link MessageToMessageDecoder}(实际的协议解码器)来替换一个* 	已存在的 {@link ReplayingDecoder} （协议探测器）。*	不可能直接调用 {@link ChannelPipeline#replace(ChannelHandler, String, ChannelHandler)}，但是*	需要一些额外的步骤：* It is not possible to achieve this simply by calling* {@link ChannelPipeline#replace(ChannelHandler, String, ChannelHandler)}, but* some additional steps are required:* <pre>* public class FirstDecoder extends {@link ReplayingDecoder}&lt;{@link Void}&gt; {**     {@code @Override}*     protected void decode({@link ChannelHandlerContext} ctx,*                             {@link ByteBuf} buf, List&lt;Object&gt; out) {*         ...*         // Decode the first message*         Object firstMessage = ...;**         // Add the second decoder*         ctx.pipeline().addLast("second", new SecondDecoder());**         if (buf.isReadable()) {*             // Hand off the remaining data to the second decoder*             out.add(firstMessage);*             out.add(buf.readBytes(<b>super.actualReadableBytes()</b>));*         } else {*             // Nothing to hand off*             out.add(firstMessage);*         }*         // Remove the first decoder (me)*         ctx.pipeline().remove(this);*     }* </pre>* @param <S>*			状态类型：通常为 {@link Enum}; 如果不需要状态管理时使用 {@link Void}*/
public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder {static final Signal REPLAY = Signal.valueOf(ReplayingDecoder.class, "REPLAY");private final ReplayingDecoderByteBuf replayable = new ReplayingDecoderByteBuf();private S state;private int checkpoint = -1;/*** Creates a new instance with no initial state (i.e: {@code null}).*/protected ReplayingDecoder() {this(null);}/*** Creates a new instance with the specified initial state.*/protected ReplayingDecoder(S initialState) {state = initialState;}/*** Stores the internal cumulative buffer's reader position.*/protected void checkpoint() {checkpoint = internalBuffer().readerIndex();}/*** Stores the internal cumulative buffer's reader position and updates* the current decoder state.*/protected void checkpoint(S state) {checkpoint();state(state);}/*** Returns the current state of this decoder.* @return the current state of this decoder*/protected S state() {return state;}/*** Sets the current state of this decoder.* @return the old state of this decoder*/protected S state(S newState) {S oldState = state;state = newState;return oldState;}@Overridefinal void channelInputClosed(ChannelHandlerContext ctx, List<Object> out) throws Exception {try {replayable.terminate();if (cumulation != null) {callDecode(ctx, internalBuffer(), out);} else {replayable.setCumulation(Unpooled.EMPTY_BUFFER);}decodeLast(ctx, replayable, out);} catch (Signal replay) {// Ignorereplay.expect(REPLAY);}}@Overrideprotected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {replayable.setCumulation(in);try {while (in.isReadable()) {int oldReaderIndex = checkpoint = in.readerIndex();int outSize = out.size();if (outSize > 0) {fireChannelRead(ctx, out, outSize);out.clear();// Check if this handler was removed before continuing with decoding.// If it was removed, it is not safe to continue to operate on the buffer.//// See:// - https://github.com/netty/netty/issues/4635if (ctx.isRemoved()) {break;}outSize = 0;}S oldState = state;int oldInputLength = in.readableBytes();try {decodeRemovalReentryProtection(ctx, replayable, out);// Check if this handler was removed before continuing the loop.// If it was removed, it is not safe to continue to operate on the buffer.//// See https://github.com/netty/netty/issues/1664if (ctx.isRemoved()) {break;}if (outSize == out.size()) {if (oldInputLength == in.readableBytes() && oldState == state) {throw new DecoderException(StringUtil.simpleClassName(getClass()) + ".decode() must consume the inbound " +"data or change its state if it did not decode anything.");} else {// Previous data has been discarded or caused state transition.// Probably it is reading on.continue;}}} catch (Signal replay) {replay.expect(REPLAY);// Check if this handler was removed before continuing the loop.// If it was removed, it is not safe to continue to operate on the buffer.//// See https://github.com/netty/netty/issues/1664if (ctx.isRemoved()) {break;}// Return to the checkpoint (or oldPosition) and retry.int checkpoint = this.checkpoint;if (checkpoint >= 0) {in.readerIndex(checkpoint);} else {// Called by cleanup() - no need to maintain the readerIndex// anymore because the buffer has been released already.}break;}if (oldReaderIndex == in.readerIndex() && oldState == state) {throw new DecoderException(StringUtil.simpleClassName(getClass()) + ".decode() method must consume the inbound data " +"or change its state if it decoded something.");}if (isSingleDecode()) {break;}}} catch (DecoderException e) {throw e;} catch (Exception cause) {throw new DecoderException(cause);}}
}