View Javadoc

    1   /*
    2    * Copyright 2009 Red Hat, Inc.
    3    *
    4    * Red Hat licenses this file to you under the Apache License, version 2.0
    5    * (the "License"); you may not use this file except in compliance with the
    6    * License.  You may obtain a copy of the License at:
    7    *
    8    *    http://www.apache.org/licenses/LICENSE-2.0
    9    *
    10   * Unless required by applicable law or agreed to in writing, software
    11   * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
    12   * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
    13   * License for the specific language governing permissions and limitations
    14   * under the License.
    15   */
    16  package org.jboss.netty.handler.codec.replay;
    17
    18  import java.net.SocketAddress;
    19  import java.util.concurrent.atomic.AtomicReference;
    20
    21  import org.jboss.netty.buffer.ChannelBuffer;
    22  import org.jboss.netty.buffer.ChannelBufferFactory;
    23  import org.jboss.netty.buffer.ChannelBuffers;
    24  import org.jboss.netty.channel.Channel;
    25  import org.jboss.netty.channel.ChannelHandler;
    26  import org.jboss.netty.channel.ChannelHandlerContext;
    27  import org.jboss.netty.channel.ChannelPipeline;
    28  import org.jboss.netty.channel.ChannelStateEvent;
    29  import org.jboss.netty.channel.Channels;
    30  import org.jboss.netty.channel.ExceptionEvent;
    31  import org.jboss.netty.channel.MessageEvent;
    32  import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
    33  import org.jboss.netty.handler.codec.frame.FrameDecoder;
    34
    35  /**
    36   * A specialized variation of {@link FrameDecoder} which enables implementation
    37   * of a non-blocking decoder in the blocking I/O paradigm.
    38   * <p>
    39   * The biggest difference between {@link ReplayingDecoder} and
    40   * {@link FrameDecoder} is that {@link ReplayingDecoder} allows you to
    41   * implement the {@code decode()} and {@code decodeLast()} methods just like
    42   * all required bytes were received already, rather than checking the
    43   * availability of the required bytes.  For example, the following
    44   * {@link FrameDecoder} implementation:
    45   * <pre>
    46   * public class IntegerHeaderFrameDecoder extends {@link FrameDecoder} {
    47   *
    48   *   {@code @Override}
    49   *   protected Object decode({@link ChannelHandlerContext} ctx,
    50   *                           {@link Channel} channel,
    51   *                           {@link ChannelBuffer} buf) throws Exception {
    52   *
    53   *     if (buf.readableBytes() &lt; 4) {
    54   *        return <strong>null</strong>;
    55   *     }
    56   *
    57   *     buf.markReaderIndex();
    58   *     int length = buf.readInt();
    59   *
    60   *     if (buf.readableBytes() &lt; length) {
    61   *        buf.resetReaderIndex();
    62   *        return <strong>null</strong>;
    63   *     }
    64   *
    65   *     return buf.readBytes(length);
    66   *   }
    67   * }
    68   * </pre>
    69   * is simplified like the following with {@link ReplayingDecoder}:
    70   * <pre>
    71   * public class IntegerHeaderFrameDecoder
    72   *      extends {@link ReplayingDecoder}&lt;{@link VoidEnum}&gt; {
    73   *
    74   *   protected Object decode({@link ChannelHandlerContext} ctx,
    75   *                           {@link Channel} channel,
    76   *                           {@link ChannelBuffer} buf,
    77   *                           {@link VoidEnum} state) throws Exception {
    78   *
    79   *     return buf.readBytes(buf.readInt());
    80   *   }
    81   * }
    82   * </pre>
    83   *
    84   * <h3>How does this work?</h3>
    85   * <p>
    86   * {@link ReplayingDecoder} passes a specialized {@link ChannelBuffer}
    87   * implementation which throws an {@link Error} of certain type when there's not
    88   * enough data in the buffer.  In the {@code IntegerHeaderFrameDecoder} above,
    89   * you just assumed that there will be 4 or more bytes in the buffer when
    90   * you call {@code buf.readInt()}.  If there's really 4 bytes in the buffer,
    91   * it will return the integer header as you expected.  Otherwise, the
    92   * {@link Error} will be raised and the control will be returned to
    93   * {@link ReplayingDecoder}.  If {@link ReplayingDecoder} catches the
    94   * {@link Error}, then it will rewind the {@code readerIndex} of the buffer
    95   * back to the 'initial' position (i.e. the beginning of the buffer) and call
    96   * the {@code decode(..)} method again when more data is received into the
    97   * buffer.
    98   * <p>
    99   * Please note that the overhead of throwing an {@link Error} is minimal unlike
    100  * throwing a new {@link Exception} in an ordinary way. {@link ReplayingDecoder}
    101  * reuses the same {@link Error} instance so that it does not need to fill its
    102  * stack trace, which takes most of {@link Exception} initialization time.
    103  *
    104  * <h3>Limitations</h3>
    105  * <p>
    106  * At the cost of the simplicity, {@link ReplayingDecoder} enforces you a few
    107  * limitations:
    108  * <ul>
    109  * <li>Some buffer operations are prohibited.</li>
    110  * <li>Performance can be worse if the network is slow and the message
    111  *     format is complicated unlike the example above.  In this case, your
    112  *     decoder might have to decode the same part of the message over and over
    113  *     again.</li>
    114  * <li>You must keep in mind that {@code decode(..)} method can be called many
    115  *     times to decode a single message.  For example, the following code will
    116  *     not work:
    117  * <pre> public class MyDecoder extends {@link ReplayingDecoder}&lt;{@link VoidEnum}&gt; {
    118  *
    119  *   private final Queue&lt;Integer&gt; values = new LinkedList&lt;Integer&gt;();
    120  *
    121  *   {@code @Override}
    122  *   public Object decode(.., {@link ChannelBuffer} buffer, ..) throws Exception {
    123  *
    124  *     // A message contains 2 integers.
    125  *     values.offer(buffer.readInt());
    126  *     values.offer(buffer.readInt());
    127  *
    128  *     // This assertion will fail intermittently since values.offer()
    129  *     // can be called more than two times!
    130  *     assert values.size() == 2;
    131  *     return values.poll() + values.poll();
    132  *   }
    133  * }</pre>
    134  *      The correct implementation looks like the following, and you can also
    135  *      utilize the 'checkpoint' feature which is explained in detail in the
    136  *      next section.
    137  * <pre> public class MyDecoder extends {@link ReplayingDecoder}&lt;{@link VoidEnum}&gt; {
    138  *
    139  *   private final Queue&lt;Integer&gt; values = new LinkedList&lt;Integer&gt;();
    140  *
    141  *   {@code @Override}
    142  *   public Object decode(.., {@link ChannelBuffer} buffer, ..) throws Exception {
    143  *
    144  *     // Revert the state of the variable that might have been changed
    145  *     // since the last partial decode.
    146  *     values.clear();
    147  *
    148  *     // A message contains 2 integers.
    149  *     values.offer(buffer.readInt());
    150  *     values.offer(buffer.readInt());
    151  *
    152  *     // Now we know this assertion will never fail.
    153  *     assert values.size() == 2;
    154  *     return values.poll() + values.poll();
    155  *   }
    156  * }</pre>
    157  *     </li>
    158  * </ul>
    159  *
    160  * <h3>Improving the performance</h3>
    161  * <p>
    162  * Fortunately, the performance of a complex decoder implementation can be
    163  * improved significantly with the {@code checkpoint()} method.  The
    164  * {@code checkpoint()} method updates the 'initial' position of the buffer so
    165  * that {@link ReplayingDecoder} rewinds the {@code readerIndex} of the buffer
    166  * to the last position where you called the {@code checkpoint()} method.
    167  *
    168  * <h4>Calling {@code checkpoint(T)} with an {@link Enum}</h4>
    169  * <p>
    170  * Although you can just use {@code checkpoint()} method and manage the state
    171  * of the decoder by yourself, the easiest way to manage the state of the
    172  * decoder is to create an {@link Enum} type which represents the current state
    173  * of the decoder and to call {@code checkpoint(T)} method whenever the state
    174  * changes.  You can have as many states as you want depending on the
    175  * complexity of the message you want to decode:
    176  *
    177  * <pre>
    178  * public enum MyDecoderState {
    179  *   READ_LENGTH,
    180  *   READ_CONTENT;
    181  * }
    182  *
    183  * public class IntegerHeaderFrameDecoder
    184  *      extends {@link ReplayingDecoder}&lt;<strong>MyDecoderState</strong>&gt; {
    185  *
    186  *   private int length;
    187  *
    188  *   public IntegerHeaderFrameDecoder() {
    189  *     // Set the initial state.
    190  *     <strong>super(MyDecoderState.READ_LENGTH);</strong>
    191  *   }
    192  *
    193  *   {@code @Override}
    194  *   protected Object decode({@link ChannelHandlerContext} ctx,
    195  *                           {@link Channel} channel,
    196  *                           {@link ChannelBuffer} buf,
    197  *                           <b>MyDecoderState</b> state) throws Exception {
    198  *     switch (state) {
    199  *     case READ_LENGTH:
    200  *       length = buf.readInt();
    201  *       <strong>checkpoint(MyDecoderState.READ_CONTENT);</strong>
    202  *     case READ_CONTENT:
    203  *       ChannelBuffer frame = buf.readBytes(length);
    204  *       <strong>checkpoint(MyDecoderState.READ_LENGTH);</strong>
    205  *       return frame;
    206  *     default:
    207  *       throw new Error("Shouldn't reach here.");
    208  *     }
    209  *   }
    210  * }
    211  * </pre>
    212  *
    213  * <h4>Calling {@code checkpoint()} with no parameter</h4>
    214  * <p>
    215  * An alternative way to manage the decoder state is to manage it by yourself.
    216  * <pre>
    217  * public class IntegerHeaderFrameDecoder
    218  *      extends {@link ReplayingDecoder}&lt;<strong>{@link VoidEnum}</strong>&gt; {
    219  *
    220  *   <strong>private boolean readLength;</strong>
    221  *   private int length;
    222  *
    223  *   {@code @Override}
    224  *   protected Object decode({@link ChannelHandlerContext} ctx,
    225  *                           {@link Channel} channel,
    226  *                           {@link ChannelBuffer} buf,
    227  *                           {@link VoidEnum} state) throws Exception {
    228  *     if (!readLength) {
    229  *       length = buf.readInt();
    230  *       <strong>readLength = true;</strong>
    231  *       <strong>checkpoint();</strong>
    232  *     }
    233  *
    234  *     if (readLength) {
    235  *       ChannelBuffer frame = buf.readBytes(length);
    236  *       <strong>readLength = false;</strong>
    237  *       <strong>checkpoint();</strong>
    238  *       return frame;
    239  *     }
    240  *   }
    241  * }
    242  * </pre>
    243  *
    244  * <h3>Replacing a decoder with another decoder in a pipeline</h3>
    245  * <p>
    246  * If you are going to write a protocol multiplexer, you will probably want to
    247  * replace a {@link ReplayingDecoder} (protocol detector) with another
    248  * {@link ReplayingDecoder} or {@link FrameDecoder} (actual protocol decoder).
    249  * It is not possible to achieve this simply by calling
    250  * {@link ChannelPipeline#replace(ChannelHandler, String, ChannelHandler)}, but
    251  * some additional steps are required:
    252  * <pre>
    253  * public class FirstDecoder extends {@link ReplayingDecoder}&lt;{@link VoidEnum}&gt; {
    254  *
    255  *     public FirstDecoder() {
    256  *         super(true); // Enable unfold
    257  *     }
    258  *
    259  *     {@code @Override}
    260  *     protected Object decode({@link ChannelHandlerContext} ctx,
    261  *                             {@link Channel} ch,
    262  *                             {@link ChannelBuffer} buf,
    263  *                             {@link VoidEnum} state) {
    264  *         ...
    265  *         // Decode the first message
    266  *         Object firstMessage = ...;
    267  *
    268  *         // Add the second decoder
    269  *         ctx.getPipeline().addLast("second", new SecondDecoder());
    270  *
    271  *         // Remove the first decoder (me)
    272  *         ctx.getPipeline().remove(this);
    273  *
    274  *         if (buf.readable()) {
    275  *             // Hand off the remaining data to the second decoder
    276  *             return new Object[] { firstMessage, buf.readBytes(<b>super.actualReadableBytes()</b>) };
    277  *         } else {
    278  *             // Nothing to hand off
    279  *             return firstMessage;
    280  *         }
    281  *     }
    282  * </pre>
    283  *
    284  * @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
    285  * @author <a href="http://gleamynode.net/">Trustin Lee</a>
    286  *
    287  * @version $Rev: 2358 $, $Date: 2010-08-30 15:03:31 +0900 (Mon, 30 Aug 2010) $
    288  *
    289  * @param <T>
    290  *        the state type; use {@link VoidEnum} if state management is unused
    291  *
    292  * @apiviz.landmark
    293  * @apiviz.has org.jboss.netty.handler.codec.replay.UnreplayableOperationException oneway - - throws
    294  */
    295 public abstract class ReplayingDecoder<T extends Enum<T>>
    296         extends SimpleChannelUpstreamHandler {
    297
    298
    299     private final AtomicReference<ChannelBuffer> cumulation =
    300         new AtomicReference<ChannelBuffer>();
    301     private final boolean unfold;
    302     private ReplayingDecoderBuffer replayable;
    303     private T state;
    304     private int checkpoint;
    305
    306     /**
    307      * Creates a new instance with no initial state (i.e: {@code null}).
    308      */
    309     protected ReplayingDecoder() {
    310         this(null);
    311     }
    312
    313     protected ReplayingDecoder(boolean unfold) {
    314         this(null, unfold);
    315     }
    316
    317     /**
    318      * Creates a new instance with the specified initial state.
    319      */
    320     protected ReplayingDecoder(T initialState) {
    321         this(initialState, false);
    322     }
    323
    324     protected ReplayingDecoder(T initialState, boolean unfold) {
    325         this.state = initialState;
    326         this.unfold = unfold;
    327     }
    328
    329     /**
    330      * Stores the internal cumulative buffer's reader position.
    331      */
    332     protected void checkpoint() {
    333         ChannelBuffer cumulation = this.cumulation.get();
    334         if (cumulation != null) {
    335             checkpoint = cumulation.readerIndex();
    336         } else {
    337             checkpoint = -1; // buffer not available (already cleaned up)
    338         }
    339     }
    340
    341     /**
    342      * Stores the internal cumulative buffer's reader position and updates
    343      * the current decoder state.
    344      */
    345     protected void checkpoint(T state) {
    346         checkpoint();
    347         setState(state);
    348     }
    349
    350     /**
    351      * Returns the current state of this decoder.
    352      * @return the current state of this decoder
    353      */
    354     protected T getState() {
    355         return state;
    356     }
    357
    358     /**
    359      * Sets the current state of this decoder.
    360      * @return the old state of this decoder
    361      */
    362     protected T setState(T newState) {
    363         T oldState = state;
    364         state = newState;
    365         return oldState;
    366     }
    367
    368     /**
    369      * Returns the actual number of readable bytes in the internal cumulative
    370      * buffer of this decoder.  You usually do not need to rely on this value
    371      * to write a decoder.  Use it only when you muse use it at your own risk.
    372      * This method is a shortcut to {@link #internalBuffer() internalBuffer().readableBytes()}.
    373      */
    374     protected int actualReadableBytes() {
    375         return internalBuffer().readableBytes();
    376     }
    377
    378     /**
    379      * Returns the internal cumulative buffer of this decoder.  You usually
    380      * do not need to access the internal buffer directly to write a decoder.
    381      * Use it only when you must use it at your own risk.
    382      */
    383     protected ChannelBuffer internalBuffer() {
    384         ChannelBuffer buf = cumulation.get();
    385         if (buf == null) {
    386             return ChannelBuffers.EMPTY_BUFFER;
    387         }
    388         return buf;
    389     }
    390
    391     /**
    392      * Decodes the received packets so far into a frame.
    393      *
    394      * @param ctx      the context of this handler
    395      * @param channel  the current channel
    396      * @param buffer   the cumulative buffer of received packets so far.
    397      *                 Note that the buffer might be empty, which means you
    398      *                 should not make an assumption that the buffer contains
    399      *                 at least one byte in your decoder implementation.
    400      * @param state    the current decoder state ({@code null} if unused)
    401      *
    402      * @return the decoded frame
    403      */
    404     protected abstract Object decode(ChannelHandlerContext ctx,
    405             Channel channel, ChannelBuffer buffer, T state) throws Exception;
    406
    407     /**
    408      * Decodes the received data so far into a frame when the channel is
    409      * disconnected.
    410      *
    411      * @param ctx      the context of this handler
    412      * @param channel  the current channel
    413      * @param buffer   the cumulative buffer of received packets so far.
    414      *                 Note that the buffer might be empty, which means you
    415      *                 should not make an assumption that the buffer contains
    416      *                 at least one byte in your decoder implementation.
    417      * @param state    the current decoder state ({@code null} if unused)
    418      *
    419      * @return the decoded frame
    420      */
    421     protected Object decodeLast(
    422             ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer, T state) throws Exception {
    423         return decode(ctx, channel, buffer, state);
    424     }
    425
    426     @Override
    427     public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
    428             throws Exception {
    429
    430         Object m = e.getMessage();
    431         if (!(m instanceof ChannelBuffer)) {
    432             ctx.sendUpstream(e);
    433             return;
    434         }
    435
    436         ChannelBuffer input = (ChannelBuffer) m;
    437         if (!input.readable()) {
    438             return;
    439         }
    440
    441         ChannelBuffer cumulation = cumulation(ctx);
    442         cumulation.discardReadBytes();
    443         cumulation.writeBytes(input);
    444         callDecode(ctx, e.getChannel(), cumulation, e.getRemoteAddress());
    445     }
    446
    447     @Override
    448     public void channelDisconnected(ChannelHandlerContext ctx,
    449             ChannelStateEvent e) throws Exception {
    450         cleanup(ctx, e);
    451     }
    452
    453     @Override
    454     public void channelClosed(ChannelHandlerContext ctx,
    455             ChannelStateEvent e) throws Exception {
    456         cleanup(ctx, e);
    457     }
    458
    459     @Override
    460     public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
    461             throws Exception {
    462         ctx.sendUpstream(e);
    463     }
    464
    465     private void callDecode(ChannelHandlerContext context, Channel channel, ChannelBuffer cumulation, SocketAddress remoteAddress) throws Exception {
    466         while (cumulation.readable()) {
    467             int oldReaderIndex = checkpoint = cumulation.readerIndex();
    468             Object result = null;
    469             T oldState = state;
    470             try {
    471                 result = decode(context, channel, replayable, state);
    472                 if (result == null) {
    473                     if (oldReaderIndex == cumulation.readerIndex() && oldState == state) {
    474                         throw new IllegalStateException(
    475                                 "null cannot be returned if no data is consumed and state didn't change.");
    476                     } else {
    477                         // Previous data has been discarded or caused state transition.
    478                         // Probably it is reading on.
    479                         continue;
    480                     }
    481                 }
    482             } catch (ReplayError replay) {
    483                 // Return to the checkpoint (or oldPosition) and retry.
    484                 int checkpoint = this.checkpoint;
    485                 if (checkpoint >= 0) {
    486                     cumulation.readerIndex(checkpoint);
    487                 } else {
    488                     // Called by cleanup() - no need to maintain the readerIndex
    489                     // anymore because the buffer has been released already.
    490                 }
    491             }
    492
    493             if (result == null) {
    494                 // Seems like more data is required.
    495                 // Let us wait for the next notification.
    496                 break;
    497             }
    498
    499             if (oldReaderIndex == cumulation.readerIndex() && oldState == state) {
    500                 throw new IllegalStateException(
    501                         "decode() method must consume at least one byte " +
    502                         "if it returned a decoded message (caused by: " +
    503                         getClass() + ")");
    504             }
    505
    506             // A successful decode
    507             unfoldAndfireMessageReceived(context, result, remoteAddress);
    508         }
    509     }
    510
    511     private void unfoldAndfireMessageReceived(
    512             ChannelHandlerContext context, Object result, SocketAddress remoteAddress) {
    513         if (unfold) {
    514             if (result instanceof Object[]) {
    515                 for (Object r: (Object[]) result) {
    516                     Channels.fireMessageReceived(context, r, remoteAddress);
    517                 }
    518             } else if (result instanceof Iterable<?>) {
    519                 for (Object r: (Iterable<?>) result) {
    520                     Channels.fireMessageReceived(context, r, remoteAddress);
    521                 }
    522             } else {
    523                 Channels.fireMessageReceived(context, result, remoteAddress);
    524             }
    525         } else {
    526             Channels.fireMessageReceived(context, result, remoteAddress);
    527         }
    528     }
    529
    530     private void cleanup(ChannelHandlerContext ctx, ChannelStateEvent e)
    531             throws Exception {
    532         try {
    533             ChannelBuffer cumulation = this.cumulation.getAndSet(null);
    534             if (cumulation == null) {
    535                 return;
    536             }
    537
    538             replayable.terminate();
    539
    540             if (cumulation.readable()) {
    541                 // Make sure all data was read before notifying a closed channel.
    542                 callDecode(ctx, e.getChannel(), cumulation, null);
    543             }
    544
    545             // Call decodeLast() finally.  Please note that decodeLast() is
    546             // called even if there's nothing more to read from the buffer to
    547             // notify a user that the connection was closed explicitly.
    548             Object partiallyDecoded = decodeLast(ctx, e.getChannel(), replayable, state);
    549             if (partiallyDecoded != null) {
    550                 unfoldAndfireMessageReceived(ctx, partiallyDecoded, null);
    551             }
    552         } catch (ReplayError replay) {
    553             // Ignore
    554         } finally {
    555             ctx.sendUpstream(e);
    556         }
    557     }
    558
    559     private ChannelBuffer cumulation(ChannelHandlerContext ctx) {
    560         ChannelBuffer buf = cumulation.get();
    561         if (buf == null) {
    562             ChannelBufferFactory factory = ctx.getChannel().getConfig().getBufferFactory();
    563             buf = new UnsafeDynamicChannelBuffer(factory);
    564             if (cumulation.compareAndSet(null, buf)) {
    565                 replayable = new ReplayingDecoderBuffer(buf);
    566             } else {
    567                 buf = cumulation.get();
    568             }
    569         }
    570         return buf;
    571     }
    572 }