S - the state type which is usually an Enum; use Void if state management is
unusedpublic abstract class ReplayingDecoder<S> extends ByteToMessageDecoder
ByteToMessageDecoder which enables implementation
of a non-blocking decoder in the blocking I/O paradigm.
The biggest difference between ReplayingDecoder and
ByteToMessageDecoder 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
ByteToMessageDecoder implementation:
public class IntegerHeaderFrameDecoder extendsis simplified like the following withByteToMessageDecoder{@Overrideprotected void decode(ChannelHandlerContextctx,ByteBufbuf, List<Object> out) throws Exception { if (buf.readableBytes() < 4) { return; } buf.markReaderIndex(); int length = buf.readInt(); if (buf.readableBytes() < length) { buf.resetReaderIndex(); return; } out.add(buf.readBytes(length)); } }
ReplayingDecoder:
public class IntegerHeaderFrameDecoder
extends ReplayingDecoder<Void> {
protected void decode(ChannelHandlerContext ctx,
ByteBuf buf) throws Exception {
out.add(buf.readBytes(buf.readInt()));
}
}
ReplayingDecoder passes a specialized ByteBuf
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 ReplayingDecoder always throws the same cached
Error instance to avoid the overhead of creating a new Error
and filling its stack trace for every throw.
At the cost of the simplicity, ReplayingDecoder enforces you a few
limitations:
decode(..) method can be called many
times to decode a single message. For example, the following code will
not work:
public class MyDecoder extendsThe correct implementation looks like the following, and you can also utilize the 'checkpoint' feature which is explained in detail in the next section.ReplayingDecoder<Void> { private final Queue<Integer> values = new LinkedList<Integer>();@Overridepublic void decode(..,ByteBufbuf, List<Object> out) throws Exception { // A message contains 2 integers. values.offer(buf.readInt()); values.offer(buf.readInt()); // This assertion will fail intermittently since values.offer() // can be called more than two times! assert values.size() == 2; out.add(values.poll() + values.poll()); } }
public class MyDecoder extendsReplayingDecoder<Void> { private final Queue<Integer> values = new LinkedList<Integer>();@Overridepublic void decode(..,ByteBufbuf, List<Object> out) 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(buf.readInt()); values.offer(buf.readInt()); // Now we know this assertion will never fail. assert values.size() == 2; out.add(values.poll() + values.poll()); } }
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.
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 void decode(ChannelHandlerContext ctx,
ByteBuf buf, List<Object> out) throws Exception {
switch (state()) {
case READ_LENGTH:
length = buf.readInt();
checkpoint(MyDecoderState.READ_CONTENT);
case READ_CONTENT:
ByteBuf frame = buf.readBytes(length);
checkpoint(MyDecoderState.READ_LENGTH);
out.add(frame);
break;
default:
throw new Error("Shouldn't reach here.");
}
}
}
checkpoint() with no parameterAn alternative way to manage the decoder state is to manage it by yourself.
public class IntegerHeaderFrameDecoder
extends ReplayingDecoder<Void> {
private boolean readLength;
private int length;
@Override
protected void decode(ChannelHandlerContext ctx,
ByteBuf buf, List<Object> out) throws Exception {
if (!readLength) {
length = buf.readInt();
readLength = true;
checkpoint();
}
if (readLength) {
ByteBuf frame = buf.readBytes(length);
readLength = false;
checkpoint();
out.add(frame);
}
}
}
If you are going to write a protocol multiplexer, you will probably want to
replace a ReplayingDecoder (protocol detector) with another
ReplayingDecoder, ByteToMessageDecoder or MessageToMessageDecoder
(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 extendsReplayingDecoder<Void> {@Overrideprotected void decode(ChannelHandlerContextctx,ByteBufbuf, List<Object> 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(super.actualReadableBytes())); } else { // Nothing to hand off out.add(firstMessage); } // Remove the first decoder (me) ctx.pipeline().remove(this); }
ByteToMessageDecoder.CumulatorChannelHandler.SharableCOMPOSITE_CUMULATOR, MERGE_CUMULATOR| 限定符 | 构造器和说明 |
|---|---|
protected |
ReplayingDecoder()
Creates a new instance with no initial state (i.e:
null). |
protected |
ReplayingDecoder(S initialState)
Creates a new instance with the specified initial state.
|
| 限定符和类型 | 方法和说明 |
|---|---|
protected void |
callDecode(ChannelHandlerContext ctx,
ByteBuf in,
java.util.List<java.lang.Object> out)
Called once data should be decoded from the given
ByteBuf. |
protected void |
checkpoint()
Stores the internal cumulative buffer's reader position.
|
protected void |
checkpoint(S state)
Stores the internal cumulative buffer's reader position and updates
the current decoder state.
|
protected S |
state()
Returns the current state of this decoder.
|
protected S |
state(S newState)
Sets the current state of this decoder.
|
actualReadableBytes, channelInactive, channelRead, channelReadComplete, decode, decodeLast, discardSomeReadBytes, handlerRemoved, handlerRemoved0, internalBuffer, isSingleDecode, setCumulator, setDiscardAfterReads, setSingleDecode, userEventTriggeredchannelActive, channelRegistered, channelUnregistered, channelWritabilityChanged, exceptionCaughtensureNotSharable, handlerAdded, isSharableclone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, waithandlerAddedprotected ReplayingDecoder()
null).protected ReplayingDecoder(S initialState)
protected void checkpoint()
protected void checkpoint(S state)
protected S state()
protected S state(S newState)
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, java.util.List<java.lang.Object> out)
ByteToMessageDecoderByteBuf. This method will call
ByteToMessageDecoder.decode(ChannelHandlerContext, ByteBuf, List) as long as decoding should take place.callDecode 在类中 ByteToMessageDecoderctx - the ChannelHandlerContext which this ByteToMessageDecoder belongs toin - the ByteBuf from which to read dataout - the List to which decoded messages should be added