netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/NetconfEOMAggregator.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 package org.opendaylight.netconf.nettyutil.handler;
   9
  10 import com.google.common.annotations.VisibleForTesting;
  11 import io.netty.buffer.ByteBuf;
  12 import io.netty.channel.ChannelHandlerContext;
  13 import io.netty.handler.codec.ByteToMessageDecoder;
  14 import java.util.List;
  15 import org.slf4j.Logger;
  16 import org.slf4j.LoggerFactory;
  17
  18 public final class NetconfEOMAggregator extends ByteToMessageDecoder {
  19     private static final Logger LOG = LoggerFactory.getLogger(NetconfEOMAggregator.class);
  20
  21     // Cached for brevity and constantness
  22     private static final byte[] EOM = MessageParts.END_OF_MESSAGE;
  23     private static final int EOM_LENGTH = EOM.length;
  24
  25     // Number of input ByteBuf bytes known to not include the delimiter.
  26     private int bodyLength = 0;
  27
  28     @Override
  29     protected void decode(final ChannelHandlerContext ctx, final ByteBuf in, final List<Object> out) {
  30         ByteBuf frame;
  31         while ((frame = decodeFrame(ctx, in)) != null) {
  32             out.add(frame);
  33         }
  34     }
  35
  36     @VisibleForTesting
  37     int bodyLength() {
  38         return bodyLength;
  39     }
  40
  41     private ByteBuf decodeFrame(final ChannelHandlerContext ctx, final ByteBuf in) {
  42         // Cache the details of input ByteBuf as they are invariants
  43         final int readerIndex = in.readerIndex();
  44         final int writerIndex = in.writerIndex();
  45
  46         int searchIndex = readerIndex + bodyLength;
  47         while (true) {
  48             // Try to find the first EOM byte
  49             final int eomIndex = in.indexOf(searchIndex, writerIndex, EOM[0]);
  50             if (eomIndex == -1) {
  51                 // The first byte (']') is not present, everything in the buffer is part of the body
  52                 bodyLength = writerIndex - readerIndex;
  53                 return null;
  54             }
  55
  56             // a.k.a. in.readableBytes() from the first EOM byte
  57             final int readableBytes = writerIndex - eomIndex;
  58             if (readableBytes < EOM_LENGTH) {
  59                 // Not enough bytes to contain a delimiter, bail out
  60                 LOG.trace("Context {} buffer {} has only {} new bytes", ctx, in, readableBytes);
  61                 bodyLength = eomIndex - readerIndex;
  62                 return null;
  63             }
  64
  65             // Check for EOM match
  66             if (isEom(in, eomIndex)) {
  67                 final int frameLength = eomIndex - readerIndex;
  68                 LOG.debug("Context {} buffer {} frame detected: length {}", ctx, in, frameLength);
  69                 final var ret = in.readRetainedSlice(frameLength);
  70                 in.skipBytes(EOM_LENGTH);
  71                 bodyLength = 0;
  72                 return ret;
  73             }
  74
  75             // No match: move one byte past eomIndex and repeat
  76             searchIndex = eomIndex + 1;
  77             LOG.trace("Context {} buffer {} restart at {}", ctx, in, searchIndex);
  78         }
  79     }
  80
  81     private static boolean isEom(final ByteBuf in, final int index) {
  82         for (int i = 1; i < EOM_LENGTH; ++i) {
  83             if (in.getByte(index + i) != EOM[i]) {
  84                 return false;
  85             }
  86         }
  87         return true;
  88     }
  89 }