ovsdb/src/main/java/org/opendaylight/ovsdb/lib/jsonrpc/JsonRpcDecoder.java

   1 /*
   2  * Copyright (C) 2013 EBay Software Foundation
   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  * Authors : Ashwin Raveendran, Madhu Venugopal
   9  */
  10 package org.opendaylight.ovsdb.lib.jsonrpc;
  11
  12
  13 import io.netty.buffer.ByteBuf;
  14 import io.netty.buffer.ByteBufInputStream;
  15 import io.netty.channel.ChannelHandlerContext;
  16 import io.netty.handler.codec.ByteToMessageDecoder;
  17 import io.netty.handler.codec.TooLongFrameException;
  18
  19 import java.io.IOException;
  20 import java.util.List;
  21
  22 import org.slf4j.Logger;
  23 import org.slf4j.LoggerFactory;
  24
  25 import com.fasterxml.jackson.core.JsonEncoding;
  26 import com.fasterxml.jackson.core.JsonFactory;
  27 import com.fasterxml.jackson.core.JsonParser;
  28 import com.fasterxml.jackson.core.io.IOContext;
  29 import com.fasterxml.jackson.core.json.ByteSourceJsonBootstrapper;
  30 import com.fasterxml.jackson.core.util.BufferRecycler;
  31 import com.fasterxml.jackson.databind.JsonNode;
  32 import com.fasterxml.jackson.databind.MappingJsonFactory;
  33
  34 /**
  35  * JSON RPC 1.0 compatible decoder capable of decoding JSON messages from a TCP stream.
  36  * The stream is framed first by inspecting the json for valid end marker (left curly)
  37  * and is passed to a Json parser (jackson) for converting into an object model.
  38  *
  39  * There are no JSON parsers that I am aware of that does non blocking parsing.
  40  * This approach avoids having to run json parser over and over again on the entire
  41  * stream waiting for input. Parser is invoked only when we know of a full JSON message
  42  * in the stream.
  43  */
  44 public class JsonRpcDecoder extends ByteToMessageDecoder {
  45
  46     protected static final Logger logger = LoggerFactory.getLogger(JsonRpcDecoder.class);
  47
  48     private int maxFrameLength;
  49
  50     private JsonFactory jacksonJsonFactory = new MappingJsonFactory();
  51
  52     private IOContext jacksonIOContext = new IOContext(new BufferRecycler(), null, false);
  53
  54     // context for the previously read incomplete records
  55     private int lastRecordBytes = 0;
  56     private int leftCurlies = 0;
  57     private int rightCurlies = 0;
  58     private boolean inS = false;
  59
  60     private int recordsRead;
  61
  62     public JsonRpcDecoder(int maxFrameLength) {
  63         this.maxFrameLength = maxFrameLength;
  64     }
  65
  66     @Override
  67     protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) throws Exception {
  68
  69         logger.trace("readable bytes {}, records read {}, incomplete record bytes {}",
  70                 buf.readableBytes(), recordsRead, lastRecordBytes);
  71
  72         if (lastRecordBytes == 0) {
  73             if (buf.readableBytes() < 4) {
  74                 return; //wait for more data
  75             }
  76
  77             skipSpaces(buf);
  78
  79             byte[] buff = new byte[4];
  80             buf.getBytes(buf.readerIndex(), buff);
  81             ByteSourceJsonBootstrapper strapper = new ByteSourceJsonBootstrapper(jacksonIOContext, buff, 0, 4);
  82             JsonEncoding jsonEncoding = strapper.detectEncoding();
  83             if (!JsonEncoding.UTF8.equals(jsonEncoding)) {
  84                 throw new InvalidEncodingException(jsonEncoding.getJavaName(), "currently only UTF-8 is supported");
  85             }
  86         }
  87
  88         int i = lastRecordBytes + buf.readerIndex();
  89
  90         for (; i < buf.writerIndex(); i++) {
  91             switch (buf.getByte(i)) {
  92                 case '{':
  93                     if (!inS) leftCurlies++;
  94                     break;
  95                 case '}':
  96                     if (!inS) rightCurlies++;
  97                     break;
  98                 case '"': {
  99                     if (buf.getByte(i - 1) != '\\') inS = !inS;
 100                     break;
 101                 }
 102             }
 103
 104             if (leftCurlies != 0 && leftCurlies == rightCurlies && !inS) {
 105                 ByteBuf slice = buf.readSlice(1 + i - buf.readerIndex());
 106                 JsonParser jp = jacksonJsonFactory.createParser(new ByteBufInputStream(slice));
 107                 JsonNode root = jp.readValueAsTree();
 108                 out.add(root);
 109                 leftCurlies = rightCurlies = lastRecordBytes = 0;
 110                 recordsRead++;
 111                 break;
 112             }
 113
 114             if (i - buf.readerIndex() >= maxFrameLength) {
 115                 fail(ctx, i - buf.readerIndex());
 116             }
 117         }
 118
 119         // end of stream, save the incomplete record index to avoid reexamining the whole on next run
 120         if (i >= buf.writerIndex()) {
 121             lastRecordBytes = buf.readableBytes();
 122             return;
 123         }
 124     }
 125
 126     public int getRecordsRead() {
 127         return recordsRead;
 128     }
 129
 130     private static void skipSpaces(ByteBuf b) throws IOException {
 131         while (b.isReadable()) {
 132             int ch = b.getByte(b.readerIndex()) & 0xFF;
 133             if (!(ch == ' ' || ch == '\r' || ch == '\n' || ch == '\t')) {
 134                 return;
 135             } else {
 136                 b.readByte(); //move the read index
 137             }
 138         }
 139     }
 140
 141
 142     private void print(ByteBuf buf, String message) {
 143         print(buf, buf.readerIndex(), buf.readableBytes(), message == null ? "buff" : message);
 144     }
 145
 146     private void print(ByteBuf buf, int startPos, int chars, String message) {
 147         if (null == message) message = "";
 148         if (startPos > buf.writerIndex()) {
 149             logger.trace("startPos out of bounds");
 150         }
 151         byte[] b = new byte[startPos + chars <= buf.writerIndex() ? chars : buf.writerIndex() - startPos];
 152         buf.getBytes(startPos, b);
 153         logger.trace("{} ={}", message, new String(b));
 154     }
 155
 156     // copied from Netty decoder
 157     private void fail(ChannelHandlerContext ctx, long frameLength) {
 158         if (frameLength > 0) {
 159             ctx.fireExceptionCaught(
 160                     new TooLongFrameException(
 161                             "frame length exceeds " + maxFrameLength +
 162                                     ": " + frameLength + " - discarded"));
 163         } else {
 164             ctx.fireExceptionCaught(
 165                     new TooLongFrameException(
 166                             "frame length exceeds " + maxFrameLength +
 167                                     " - discarding"));
 168         }
 169     }
 170 }