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, 2021 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 (var bufferedReader = new BufferedReader(new InputStreamReader(entityStream, StandardCharsets.UTF_8))) {
  69             List<Object> messages = new LinkedList<>();
  70             String msg;
  71             while ((msg = readMessage(bufferedReader)) != null) {
  72                 messages.add(msg);
  73             }
  74
  75             return messages;
  76         }
  77     }
  78
  79     /**
  80      * Reads a message.
  81      *
  82      * @param reader source from which to read
  83      * @return the message that was read, or {@code null} if there are no more messages
  84      * @throws IOException if an error occurs
  85      */
  86     private String readMessage(Reader reader) throws IOException {
  87         if (!skipWhitespace(reader)) {
  88             return null;
  89         }
  90
  91         int partlen = readLength(reader);
  92         if (partlen > MAX_LEN) {
  93             throw new IOException("invalid partition length");
  94         }
  95
  96         int msglen = readLength(reader);
  97         if (msglen > MAX_LEN) {
  98             throw new IOException("invalid message length");
  99         }
 100
 101         // skip over the partition
 102         reader.skip(partlen);
 103
 104         return readString(reader, msglen);
 105     }
 106
 107     /**
 108      * Skips whitespace.
 109      *
 110      * @param reader source from which to read
 111      * @return {@code true} if there is another character after the whitespace,
 112      *         {@code false} if the end of the stream has been reached
 113      * @throws IOException if an error occurs
 114      */
 115     private boolean skipWhitespace(Reader reader) throws IOException {
 116         int chr;
 117
 118         do {
 119             reader.mark(1);
 120             if ((chr = reader.read()) < 0) {
 121                 return false;
 122             }
 123         } while (Character.isWhitespace(chr));
 124
 125         // push the last character back onto the reader
 126         reader.reset();
 127
 128         return true;
 129     }
 130
 131     /**
 132      * Reads a length field, which is a number followed by ".".
 133      *
 134      * @param reader source from which to read
 135      * @return the length, or -1 if EOF has been reached
 136      * @throws IOException if an error occurs
 137      */
 138     private int readLength(Reader reader) throws IOException {
 139         var bldr = new StringBuilder(MAX_DIGITS);
 140
 141         int chr;
 142         for (var x = 0; x < MAX_DIGITS; ++x) {
 143             if ((chr = reader.read()) < 0) {
 144                 throw new EOFException("missing '.' in 'length' field");
 145             }
 146
 147             if (chr == '.') {
 148                 String text = bldr.toString().trim();
 149                 return (text.isEmpty() ? 0 : Integer.valueOf(text));
 150             }
 151
 152             if (!Character.isDigit(chr)) {
 153                 throw new IOException("invalid character in 'length' field");
 154             }
 155
 156             bldr.append((char) chr);
 157         }
 158
 159         throw new IOException("too many digits in 'length' field");
 160     }
 161
 162     /**
 163      * Reads a string.
 164      *
 165      * @param reader source from which to read
 166      * @param len length of the string (i.e., number of characters to read)
 167      * @return the string that was read
 168      * @throws IOException if an error occurs
 169      */
 170     private String readString(Reader reader, int len) throws IOException {
 171         var buf = new char[len];
 172         IOUtils.readFully(reader, buf);
 173
 174         return new String(buf);
 175     }
 176 }