models-sim/models-sim-dmaap/src/main/java/org/onap/policy/models/sim/dmaap/rest/CambriaMessageBodyHandler.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * ONAP Policy Models
   4  * ================================================================================
   5  * Copyright (C) 2019 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  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.onap.policy.models.sim.dmaap.rest;
  22
  23 import java.io.BufferedReader;
  24 import java.io.EOFException;
  25 import java.io.IOException;
  26 import java.io.InputStream;
  27 import java.io.InputStreamReader;
  28 import java.io.Reader;
  29 import java.lang.annotation.Annotation;
  30 import java.lang.reflect.Type;
  31 import java.nio.charset.StandardCharsets;
  32 import java.util.LinkedList;
  33 import java.util.List;
  34 import javax.ws.rs.Consumes;
  35 import javax.ws.rs.core.MediaType;
  36 import javax.ws.rs.core.MultivaluedMap;
  37 import javax.ws.rs.ext.MessageBodyReader;
  38 import javax.ws.rs.ext.Provider;
  39 import org.apache.commons.io.IOUtils;
  40
  41 /**
  42  * Provider that decodes "application/cambria" messages.
  43  */
  44 @Provider
  45 @Consumes(CambriaMessageBodyHandler.MEDIA_TYPE_APPLICATION_CAMBRIA)
  46 public class CambriaMessageBodyHandler implements MessageBodyReader<Object> {
  47     public static final String MEDIA_TYPE_APPLICATION_CAMBRIA = "application/cambria";
  48
  49     /**
  50      * Maximum length of a message or partition.
  51      */
  52     private static final int MAX_LEN = 10000000;
  53
  54     /**
  55      * Maximum digits in a length field.
  56      */
  57     private static final int MAX_DIGITS = 10;
  58
  59     @Override
  60     public boolean isReadable(Class<?> type, Type genericType, Annotation[] annotations, MediaType mediaType) {
  61         return (mediaType != null && MEDIA_TYPE_APPLICATION_CAMBRIA.equals(mediaType.toString()));
  62     }
  63
  64     @Override
  65     public List<Object> readFrom(Class<Object> type, Type genericType, Annotation[] annotations, MediaType mediaType,
  66                     MultivaluedMap<String, String> httpHeaders, InputStream entityStream) throws IOException {
  67
  68         try (BufferedReader bufferedReader =
  69                         new BufferedReader(new InputStreamReader(entityStream, StandardCharsets.UTF_8))) {
  70             List<Object> messages = new LinkedList<>();
  71             String msg;
  72             while ((msg = readMessage(bufferedReader)) != null) {
  73                 messages.add(msg);
  74             }
  75
  76             return messages;
  77         }
  78     }
  79
  80     /**
  81      * Reads a message.
  82      *
  83      * @param reader source from which to read
  84      * @return the message that was read, or {@code null} if there are no more messages
  85      * @throws IOException if an error occurs
  86      */
  87     private String readMessage(Reader reader) throws IOException {
  88         if (!skipWhitespace(reader)) {
  89             return null;
  90         }
  91
  92         int partlen = readLength(reader);
  93         if (partlen > MAX_LEN) {
  94             throw new IOException("invalid partition length");
  95         }
  96
  97         int msglen = readLength(reader);
  98         if (msglen > MAX_LEN) {
  99             throw new IOException("invalid message length");
 100         }
 101
 102         // skip over the partition
 103         reader.skip(partlen);
 104
 105         return readString(reader, msglen);
 106     }
 107
 108     /**
 109      * Skips whitespace.
 110      *
 111      * @param reader source from which to read
 112      * @return {@code true} if there is another character after the whitespace,
 113      *         {@code false} if the end of the stream has been reached
 114      * @throws IOException if an error occurs
 115      */
 116     private boolean skipWhitespace(Reader reader) throws IOException {
 117         int chr;
 118
 119         do {
 120             reader.mark(1);
 121             if ((chr = reader.read()) < 0) {
 122                 return false;
 123             }
 124         } while (Character.isWhitespace(chr));
 125
 126         // push the last character back onto the reader
 127         reader.reset();
 128
 129         return true;
 130     }
 131
 132     /**
 133      * Reads a length field, which is a number followed by ".".
 134      *
 135      * @param reader source from which to read
 136      * @return the length, or -1 if EOF has been reached
 137      * @throws IOException if an error occurs
 138      */
 139     private int readLength(Reader reader) throws IOException {
 140         StringBuilder bldr = new StringBuilder(MAX_DIGITS);
 141
 142         int chr;
 143         for (int x = 0; x < MAX_DIGITS; ++x) {
 144             if ((chr = reader.read()) < 0) {
 145                 throw new EOFException("missing '.' in 'length' field");
 146             }
 147
 148             if (chr == '.') {
 149                 String text = bldr.toString().trim();
 150                 return (text.isEmpty() ? 0 : Integer.valueOf(text));
 151             }
 152
 153             if (!Character.isDigit(chr)) {
 154                 throw new IOException("invalid character in 'length' field");
 155             }
 156
 157             bldr.append((char) chr);
 158         }
 159
 160         throw new IOException("too many digits in 'length' field");
 161     }
 162
 163     /**
 164      * Reads a string.
 165      *
 166      * @param reader source from which to read
 167      * @param len length of the string (i.e., number of characters to read)
 168      * @return the string that was read
 169      * @throws IOException if an error occurs
 170      */
 171     private String readString(Reader reader, int len) throws IOException {
 172         char[] buf = new char[len];
 173         IOUtils.readFully(reader, buf);
 174
 175         return new String(buf);
 176     }
 177 }