tests/honeynode/2.2.1/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/NetconfChunkAggregator.java

   1 /*
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.netconf.nettyutil.handler;
  10
  11 import io.netty.buffer.ByteBuf;
  12 import io.netty.buffer.CompositeByteBuf;
  13 import io.netty.buffer.Unpooled;
  14 import io.netty.channel.ChannelHandlerContext;
  15 import io.netty.handler.codec.ByteToMessageDecoder;
  16 import java.util.List;
  17 import org.slf4j.Logger;
  18 import org.slf4j.LoggerFactory;
  19
  20 public class NetconfChunkAggregator extends ByteToMessageDecoder {
  21     private static final Logger LOG = LoggerFactory.getLogger(NetconfChunkAggregator.class);
  22     private static final String GOT_PARAM_WHILE_WAITING_FOR_PARAM = "Got byte {} while waiting for {}";
  23     private static final String GOT_PARAM_WHILE_WAITING_FOR_PARAM_PARAM = "Got byte {} while waiting for {}-{}";
  24     public static final int DEFAULT_MAXIMUM_CHUNK_SIZE = 16 * 1024 * 1024;
  25
  26     private enum State {
  27         HEADER_ONE, // \n
  28         HEADER_TWO, // #
  29         HEADER_LENGTH_FIRST, // [1-9]
  30         HEADER_LENGTH_OTHER, // [0-9]*\n
  31         DATA,
  32         FOOTER_ONE, // \n
  33         FOOTER_TWO, // #
  34         FOOTER_THREE, // #
  35         FOOTER_FOUR, // \n
  36     }
  37
  38     private final int maxChunkSize = DEFAULT_MAXIMUM_CHUNK_SIZE;
  39     private State state = State.HEADER_ONE;
  40     private long chunkSize;
  41     private CompositeByteBuf chunk;
  42
  43     private static void checkNewLine(final byte byteToCheck, final String errorMessage) {
  44         if (byteToCheck != '\n') {
  45             LOG.debug(GOT_PARAM_WHILE_WAITING_FOR_PARAM, byteToCheck, (byte)'\n');
  46             throw new IllegalStateException(errorMessage);
  47         }
  48     }
  49
  50     private static void checkHash(final byte byteToCheck, final String errorMessage) {
  51         if (byteToCheck != '#') {
  52             LOG.debug(GOT_PARAM_WHILE_WAITING_FOR_PARAM, byteToCheck, (byte)'#');
  53             throw new IllegalStateException(errorMessage);
  54         }
  55     }
  56
  57     private void checkChunkSize() {
  58         if (chunkSize > maxChunkSize) {
  59             LOG.debug("Parsed chunk size {}, maximum allowed is {}", chunkSize, maxChunkSize);
  60             throw new IllegalStateException("Maximum chunk size exceeded");
  61         }
  62     }
  63
  64     @Override
  65     protected void decode(final ChannelHandlerContext ctx,
  66                           final ByteBuf in, final List<Object> out) throws IllegalStateException {
  67         while (in.isReadable()) {
  68             switch (state) {
  69                 case HEADER_ONE:
  70                 {
  71                     final byte b = in.readByte();
  72                     checkNewLine(b, "Malformed chunk header encountered (byte 0)");
  73                     state = State.HEADER_TWO;
  74                     initChunk();
  75                     break;
  76                 }
  77                 case HEADER_TWO:
  78                 {
  79                     final byte b = in.readByte();
  80                     checkHash(b, "Malformed chunk header encountered (byte 1)");
  81                     state = State.HEADER_LENGTH_FIRST;
  82                     break;
  83                 }
  84                 case HEADER_LENGTH_FIRST:
  85                 {
  86                     final byte b = in.readByte();
  87                     chunkSize = processHeaderLengthFirst(b);
  88                     state = State.HEADER_LENGTH_OTHER;
  89                     break;
  90                 }
  91                 case HEADER_LENGTH_OTHER:
  92                 {
  93                     final byte b = in.readByte();
  94                     if (b == '\n') {
  95                         state = State.DATA;
  96                         break;
  97                     }
  98                     if (b < '0' || b > '9') {
  99                         LOG.debug(GOT_PARAM_WHILE_WAITING_FOR_PARAM_PARAM, b, (byte)'0', (byte)'9');
 100                         throw new IllegalStateException("Invalid chunk size encountered");
 101                     }
 102                     chunkSize *= 10;
 103                     chunkSize += b - '0';
 104                     checkChunkSize();
 105                     break;
 106                 }
 107                 case DATA:
 108                     /*
 109                      * FIXME: this gathers all data into one big chunk before passing
 110                      *        it on. Make sure the pipeline can work with partial data
 111                      *        and then change this piece to pass the data on as it
 112                      *        comes through.
 113                      */
 114                     if (in.readableBytes() < chunkSize) {
 115                         LOG.debug("Buffer has {} bytes, need {} to complete chunk", in.readableBytes(), chunkSize);
 116                         in.discardReadBytes();
 117                         return;
 118                     }
 119                     aggregateChunks(in.readBytes((int) chunkSize));
 120                     state = State.FOOTER_ONE;
 121                     break;
 122                 case FOOTER_ONE:
 123                 {
 124                     final byte b = in.readByte();
 125                     checkNewLine(b,"Malformed chunk footer encountered (byte 0)");
 126                     state = State.FOOTER_TWO;
 127                     chunkSize = 0;
 128                     break;
 129                 }
 130                 case FOOTER_TWO:
 131                 {
 132                     final byte b = in.readByte();
 133                     checkHash(b,"Malformed chunk footer encountered (byte 1)");
 134                     state = State.FOOTER_THREE;
 135                     break;
 136                 }
 137                 case FOOTER_THREE:
 138                 {
 139                     final byte b = in.readByte();
 140                     // In this state, either header-of-new-chunk or message-end is expected
 141                     // Depends on the next character
 142                     extractNewChunkOrMessageEnd(b);
 143                     break;
 144                 }
 145                 case FOOTER_FOUR:
 146                 {
 147                     final byte b = in.readByte();
 148                     checkNewLine(b,"Malformed chunk footer encountered (byte 3)");
 149                     state = State.HEADER_ONE;
 150                     out.add(chunk);
 151                     chunk = null;
 152                     break;
 153                 }
 154                 default :
 155                 {
 156                     LOG.info("Unknown state.");
 157                 }
 158             }
 159         }
 160
 161         in.discardReadBytes();
 162     }
 163
 164     private void extractNewChunkOrMessageEnd(final byte byteToCheck) {
 165         if (isHeaderLengthFirst(byteToCheck)) {
 166             // Extract header length#1 from new chunk
 167             chunkSize = processHeaderLengthFirst(byteToCheck);
 168             // Proceed with next chunk processing
 169             state = State.HEADER_LENGTH_OTHER;
 170         } else if (byteToCheck == '#') {
 171             state = State.FOOTER_FOUR;
 172         } else {
 173             LOG.debug(GOT_PARAM_WHILE_WAITING_FOR_PARAM_PARAM, byteToCheck, (byte) '#', (byte) '1', (byte) '9');
 174             throw new IllegalStateException("Malformed chunk footer encountered (byte 2)");
 175         }
 176     }
 177
 178     private void initChunk() {
 179         chunk = Unpooled.compositeBuffer();
 180     }
 181
 182     private void aggregateChunks(final ByteBuf newChunk) {
 183         chunk.addComponent(chunk.numComponents(), newChunk);
 184
 185         // Update writer index, addComponent does not update it
 186         chunk.writerIndex(chunk.writerIndex() + newChunk.readableBytes());
 187     }
 188
 189     private static int processHeaderLengthFirst(final byte byteToCheck) {
 190         if (!isHeaderLengthFirst(byteToCheck)) {
 191             LOG.debug(GOT_PARAM_WHILE_WAITING_FOR_PARAM_PARAM, byteToCheck, (byte)'1', (byte)'9');
 192             throw new IllegalStateException("Invalid chunk size encountered (byte 0)");
 193         }
 194
 195         return byteToCheck - '0';
 196     }
 197
 198     private static boolean isHeaderLengthFirst(final byte byteToCheck) {
 199         return byteToCheck >= '1' && byteToCheck <= '9';
 200     }
 201 }