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