2 * Copyright (C) 2013 EBay Software Foundation
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
8 * Authors : Ashwin Raveendran, Madhu Venugopal
10 package org.opendaylight.ovsdb.lib.jsonrpc;
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;
19 import java.io.IOException;
20 import java.util.List;
22 import org.slf4j.Logger;
23 import org.slf4j.LoggerFactory;
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;
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.
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
44 public class JsonRpcDecoder extends ByteToMessageDecoder {
46 protected static final Logger logger = LoggerFactory.getLogger(JsonRpcDecoder.class);
48 private int maxFrameLength;
50 private JsonFactory jacksonJsonFactory = new MappingJsonFactory();
52 private IOContext jacksonIOContext = new IOContext(new BufferRecycler(), null, false);
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;
60 private int recordsRead;
62 public JsonRpcDecoder(int maxFrameLength) {
63 this.maxFrameLength = maxFrameLength;
67 protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) throws Exception {
69 logger.trace("readable bytes {}, records read {}, incomplete record bytes {}",
70 buf.readableBytes(), recordsRead, lastRecordBytes);
72 if (lastRecordBytes == 0) {
73 if (buf.readableBytes() < 4) {
74 return; //wait for more data
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");
88 int i = lastRecordBytes + buf.readerIndex();
90 for (; i < buf.writerIndex(); i++) {
91 switch (buf.getByte(i)) {
93 if (!inS) leftCurlies++;
96 if (!inS) rightCurlies++;
99 if (buf.getByte(i - 1) != '\\') inS = !inS;
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();
109 leftCurlies = rightCurlies = lastRecordBytes = 0;
114 if (i - buf.readerIndex() >= maxFrameLength) {
115 fail(ctx, i - buf.readerIndex());
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();
126 public int getRecordsRead() {
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')) {
136 b.readByte(); //move the read index
142 private void print(ByteBuf buf, String message) {
143 print(buf, buf.readerIndex(), buf.readableBytes(), message == null ? "buff" : message);
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");
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));
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"));
164 ctx.fireExceptionCaught(
165 new TooLongFrameException(
166 "frame length exceeds " + maxFrameLength +