T
- the state type; use VoidEnum
if state management is unusedpublic abstract class ReplayingDecoder<T extends Enum<T>> extends FrameDecoder
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 extendsis simplified like the following withFrameDecoder
{@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); } }
ReplayingDecoder
:
public class IntegerHeaderFrameDecoder extendsReplayingDecoder
<VoidEnum
> { protected Object decode(ChannelHandlerContext
ctx,Channel
channel,ChannelBuffer
buf,VoidEnum
state) throws Exception { return buf.readBytes(buf.readInt()); } }
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 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
<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(); } }
public class MyDecoder extendsReplayingDecoder
<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(); } }
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 extendsReplayingDecoder
<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."); } } }
checkpoint()
with no parameterAn alternative way to manage the decoder state is to manage it by yourself.
public class IntegerHeaderFrameDecoder extendsReplayingDecoder
<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; } } }
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 extendsReplayingDecoder
<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; } }
ChannelHandler.Sharable
cumulation, DEFAULT_MAX_COMPOSITEBUFFER_COMPONENTS
Modifier | Constructor and Description |
---|---|
protected |
ReplayingDecoder()
Creates a new instance with no initial state (i.e:
null ). |
protected |
ReplayingDecoder(boolean unfold) |
protected |
ReplayingDecoder(T initialState)
Creates a new instance with the specified initial state.
|
protected |
ReplayingDecoder(T initialState,
boolean unfold) |
Modifier and Type | Method and Description |
---|---|
protected void |
checkpoint()
Stores the internal cumulative buffer's reader position.
|
protected void |
checkpoint(T state)
Stores the internal cumulative buffer's reader position and updates
the current decoder state.
|
protected void |
cleanup(ChannelHandlerContext ctx,
ChannelStateEvent e)
|
protected Object |
decode(ChannelHandlerContext ctx,
Channel channel,
ChannelBuffer buffer)
|
protected abstract Object |
decode(ChannelHandlerContext ctx,
Channel channel,
ChannelBuffer buffer,
T state)
Decodes the received packets so far into a frame.
|
protected Object |
decodeLast(ChannelHandlerContext ctx,
Channel channel,
ChannelBuffer buffer)
Decodes the received data so far into a frame when the channel is
disconnected.
|
protected Object |
decodeLast(ChannelHandlerContext ctx,
Channel channel,
ChannelBuffer buffer,
T state)
Decodes the received data so far into a frame when the channel is
disconnected.
|
protected T |
getState()
Returns the current state of this decoder.
|
protected ChannelBuffer |
internalBuffer()
Returns the internal cumulative buffer of this decoder.
|
void |
messageReceived(ChannelHandlerContext ctx,
MessageEvent e)
Invoked when a message object (e.g:
ChannelBuffer ) was received
from a remote peer. |
protected T |
setState(T newState)
Sets the current state of this decoder.
|
actualReadableBytes, afterAdd, afterRemove, appendToCumulation, beforeAdd, beforeRemove, channelClosed, channelDisconnected, exceptionCaught, getMaxCumulationBufferCapacity, getMaxCumulationBufferComponents, isUnfold, newCumulationBuffer, replace, setMaxCumulationBufferCapacity, setMaxCumulationBufferComponents, setUnfold, unfoldAndFireMessageReceived, updateCumulation
channelBound, channelConnected, channelInterestChanged, channelOpen, channelUnbound, childChannelClosed, childChannelOpen, handleUpstream, writeComplete
protected ReplayingDecoder()
null
).protected ReplayingDecoder(boolean unfold)
protected ReplayingDecoder(T initialState)
protected ReplayingDecoder(T initialState, boolean unfold)
protected ChannelBuffer internalBuffer()
FrameDecoder
internalBuffer
in class FrameDecoder
protected void checkpoint()
protected void checkpoint(T state)
protected T getState()
protected T setState(T newState)
protected abstract Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer, T state) throws Exception
ctx
- the context of this handlerchannel
- the current channelbuffer
- the cumulative buffer of received packets so far.
Note that the buffer might be empty, which means you
should not make an assumption that the buffer contains
at least one byte in your decoder implementation.state
- the current decoder state (null
if unused)Exception
protected Object decodeLast(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer, T state) throws Exception
ctx
- the context of this handlerchannel
- the current channelbuffer
- the cumulative buffer of received packets so far.
Note that the buffer might be empty, which means you
should not make an assumption that the buffer contains
at least one byte in your decoder implementation.state
- the current decoder state (null
if unused)Exception
protected final Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception
decode(ChannelHandlerContext, Channel, ChannelBuffer, Enum)
. This method
should be never used by ReplayingDecoder
itself. But to be safe we should handle it
anywaydecode
in class FrameDecoder
ctx
- the context of this handlerchannel
- the current channelbuffer
- the cumulative buffer of received packets so far.
Note that the buffer might be empty, which means you
should not make an assumption that the buffer contains
at least one byte in your decoder implementation.null
if there's not enough data in the buffer to decode a frame.Exception
protected final Object decodeLast(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception
FrameDecoder
decodeLast
in class FrameDecoder
ctx
- the context of this handlerchannel
- the current channelbuffer
- the cumulative buffer of received packets so far.
Note that the buffer might be empty, which means you
should not make an assumption that the buffer contains
at least one byte in your decoder implementation.null
if there's not enough data in the buffer to decode a frame.Exception
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception
SimpleChannelUpstreamHandler
ChannelBuffer
) was received
from a remote peer.messageReceived
in class FrameDecoder
Exception
protected void cleanup(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception
FrameDecoder
FrameDecoder.channelDisconnected(ChannelHandlerContext, ChannelStateEvent)
and
FrameDecoder.channelClosed(ChannelHandlerContext, ChannelStateEvent)
cleanup
in class FrameDecoder
Exception
Copyright © 2008-2013 The Netty Project. All Rights Reserved.