1 /*******************************************************************************
2 * ============LICENSE_START=======================================================
4 * ================================================================================
5 * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
6 * ================================================================================
7 * Licensed under the Apache License, Version 2.0 (the "License");
8 * you may not use this file except in compliance with the License.
9 * You may obtain a copy of the License at
10 * http://www.apache.org/licenses/LICENSE-2.0
12 * Unless required by applicable law or agreed to in writing, software
13 * distributed under the License is distributed on an "AS IS" BASIS,
14 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15 * See the License for the specific language governing permissions and
16 * limitations under the License.
17 * ============LICENSE_END=========================================================
19 * ECOMP is a trademark and service mark of AT&T Intellectual Property.
21 *******************************************************************************/
22 package org.onap.dmaap.messagerouter.msgrtr.nsa.cambria.resources.streamReaders;
24 import java.io.IOException;
25 import java.io.InputStream;
27 import javax.servlet.http.HttpServletResponse;
29 import org.onap.dmaap.messagerouter.msgrtr.nsa.cambria.CambriaApiException;
30 import org.onap.dmaap.messagerouter.msgrtr.nsa.cambria.backends.Publisher.message;
31 import org.onap.dmaap.messagerouter.msgrtr.nsa.cambria.beans.LogDetails;
32 import org.onap.dmaap.messagerouter.msgrtr.nsa.cambria.resources.CambriaEventSet.reader;
35 * Read an optionally chunked stream in the Cambria app format. This format
36 * allows for speedier server-side message parsing than pure JSON. It's looks
39 * <keyLength>.<msgLength>.<key><message><br/>
41 * Whitespace before/after each entry is ignored, so messages can be delivered
42 * with newlines between them, or not.
47 public class CambriaStreamReader implements reader {
49 * constructor initializing InputStream with fStream
52 * @throws CambriaApiException
54 public CambriaStreamReader(InputStream senderStream) throws CambriaApiException {
55 fStream = senderStream;
60 * next method iterates through msg length
62 * throws CambriaApiException
65 public message next() throws IOException, CambriaApiException {
66 final int keyLen = readLength();
70 final int msgLen = readLength();
71 final String keyPart = readString(keyLen);
72 final String msgPart = readString(msgLen);
74 return new msg(keyPart, msgPart);
77 private static class msg implements message {
79 * constructor initialization
84 public msg(String key, String msg) {
85 // if no key, use the current time. This allows the message to be
87 // in any order without forcing it into a single partition as empty
89 if (key.length() < 1) {
90 key = "" + System.currentTimeMillis();
101 public String getKey() {
107 * returns the message in String type object
109 public String getMessage() {
113 private final String fKey;
114 private final String fMsg;
115 private LogDetails logDetails;
116 private boolean transactionEnabled;
119 * returns boolean value which
120 * indicates whether transaction is enabled
122 public boolean isTransactionEnabled() {
123 return transactionEnabled;
127 * sets boolean value which
128 * indicates whether transaction is enabled
130 public void setTransactionEnabled(boolean transactionEnabled) {
131 this.transactionEnabled = transactionEnabled;
136 * set log details in logDetails variable
138 public void setLogDetails(LogDetails logDetails) {
139 this.logDetails = logDetails;
144 * get the log details
146 public LogDetails getLogDetails() {
147 return this.logDetails;
152 private final InputStream fStream;
155 * max cambria length indicates message length
157 // This limit is here to prevent the server from spinning on a long string of numbers
158 // that is delivered with 'application/cambria' as the format. The limit needs to be
159 // large enough to support the max message length (currently 1MB, the default Kafka
163 private static final int kMaxCambriaLength = 4*1000*1024;
169 * @throws IOException
170 * @throws CambriaApiException
172 private int readLength() throws IOException, CambriaApiException {
173 // always ignore leading whitespace
174 int c = fStream.read();
175 while (Character.isWhitespace(c)) {
184 while (Character.isDigit(c)) {
185 result = (result * 10) + (c - '0');
186 if (result > kMaxCambriaLength) {
187 throw new CambriaApiException(HttpServletResponse.SC_BAD_REQUEST, "Expected . after length.");
193 throw new CambriaApiException(HttpServletResponse.SC_BAD_REQUEST, "Expected . after length.");
203 * @throws IOException
204 * @throws CambriaApiException
206 private String readString(int len) throws IOException, CambriaApiException {
207 final byte[] buffer = new byte[len];
209 final long startMs = System.currentTimeMillis();
210 final long timeoutMs = startMs + 30000; // FIXME configurable
213 while (readTotal < len) {
214 final int read = fStream.read(buffer, readTotal, len - readTotal);
215 if (read == -1 || System.currentTimeMillis() > timeoutMs) {
222 if (readTotal < len) {
223 throw new CambriaApiException(HttpServletResponse.SC_BAD_REQUEST,
224 "End of stream while reading " + len + " bytes");
227 return new String(buffer);