cadi/core/src/main/java/org/onap/aaf/cadi/Capacitor.java

   1 /**
   2  * ============LICENSE_START====================================================
   3  * org.onap.aaf
   4  * ===========================================================================
   5  * Copyright (c) 2018 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
  22 package org.onap.aaf.cadi;
  23
  24 import java.nio.ByteBuffer;
  25 import java.util.ArrayList;
  26
  27 /**
  28  * Capacitor
  29  *
  30  * Storage mechanism for read data, specifically designed for InputStreams.
  31  *
  32  * The Standard BufferedInputStream requires a limit to be set for buffered reading, which is
  33  * impractical for reading SOAP headers, which can be quite large.
  34  * @author Jonathan
  35  *
  36  */
  37 public class Capacitor {
  38     private static final int DEFAULT_CHUNK = 256;
  39     private ArrayList<ByteBuffer> bbs = new ArrayList<>();
  40     private ByteBuffer curr = null;
  41     private int idx;
  42
  43     // Maintain a private RingBuffer for Memory, for efficiency
  44     private static ByteBuffer[] ring = new ByteBuffer[16];
  45     private static int start, end;
  46
  47
  48     public void put(byte b) {
  49         if (curr == null || curr.remaining()==0) { // ensure we have a "curr" buffer ready for data
  50             curr = ringGet();
  51             bbs.add(curr);
  52         }
  53         curr.put(b);
  54     }
  55
  56     public int read() {
  57         if (curr!=null) {
  58             if (curr.remaining()>0) { // have a buffer, use it!
  59                 return curr.get();
  60             } else if (idx<bbs.size()){ // Buffer not enough, get next one from array
  61                 curr=bbs.get(idx++);
  62                 return curr.get();
  63             }
  64         } // if no curr buffer, treat as end of stream
  65         return -1;
  66     }
  67
  68     /**
  69      * read into an array like Streams
  70      *
  71      * @param array
  72      * @param offset
  73      * @param length
  74      * @return
  75      */
  76     public int read(byte[] array, int offset, int length) {
  77         if (curr==null)return -1;
  78         int len;
  79         int count=0;
  80         while (length>0) { // loop through while there's data needed
  81             if ((len=curr.remaining())>length) { //  if enough data in curr buffer, use this code
  82                 curr.get(array,offset,length);
  83                 count+=length;
  84                 length=0;
  85             } else {  // get data from curr, mark how much is needed to fulfil, and loop for next curr.
  86                 curr.get(array,offset,len);
  87                 count+=len;
  88                 offset+=len;
  89                 length-=len;
  90                 if (idx<bbs.size()) {
  91                     curr=bbs.get(idx++);
  92                 } else {
  93                     length=0; // stop, and return the count of how many we were able to load
  94                 }
  95             }
  96         }
  97         return count;
  98     }
  99
 100     /**
 101      * Put an array of data into Capacitor
 102      *
 103      * @param array
 104      * @param offset
 105      * @param length
 106      */
 107     public void put(byte[] array, int offset, int length) {
 108         if (curr == null || curr.remaining()==0) {
 109             curr = ringGet();
 110             bbs.add(curr);
 111         }
 112
 113         int len;
 114         while (length>0) {
 115             if ((len=curr.remaining())>length) {
 116                 curr.put(array,offset,length);
 117                 length=0;
 118             } else {
 119 //                System.out.println(new String(array));
 120                 curr.put(array,offset,len);
 121                 length-=len;
 122                 offset+=len;
 123                 curr = ringGet();
 124                 bbs.add(curr);
 125             }
 126         }
 127     }
 128
 129     /**
 130      * Move state from Storage mode into Read mode, changing all internal buffers to read mode, etc
 131      */
 132     public void setForRead() {
 133         for (ByteBuffer bb : bbs) {
 134             bb.flip();
 135         }
 136         if (bbs.isEmpty()) {
 137             curr = null;
 138             idx = 0;
 139         } else {
 140             curr=bbs.get(0);
 141             idx=1;
 142         }
 143     }
 144
 145     /**
 146      * reuse all the buffers
 147      */
 148     public void done() {
 149         for (ByteBuffer bb : bbs) {
 150             ringPut(bb);
 151         }
 152         bbs.clear();
 153         curr = null;
 154     }
 155
 156     /**
 157      * Declare amount of data available to be read at once.
 158      *
 159      * @return
 160      */
 161     public int available() {
 162         int count = 0;
 163         for (ByteBuffer bb : bbs) {
 164             count+=bb.remaining();
 165         }
 166         return count;
 167     }
 168
 169     /**
 170      * Returns how many are left that were not skipped
 171      * @param n
 172      * @return
 173      */
 174     public long skip(long n) {
 175         long skipped=0L;
 176         int skip;
 177         if (curr==null) {
 178             return 0;
 179         }
 180         while (n>0) {
 181             if (n<(skip=curr.remaining())) {
 182                 curr.position(curr.position()+(int)n);
 183                 skipped+=skip;
 184                 n=0;
 185             } else {
 186                 curr.position(curr.limit());
 187
 188                 skipped-=skip;
 189                 if (idx<bbs.size()) {
 190                     curr=bbs.get(idx++);
 191                     n-=skip;
 192                 } else {
 193                     n=0;
 194                 }
 195             }
 196         }
 197         return skipped > 0 ? skipped : 0;
 198     }
 199     /**
 200      * Be able to re-read data that is stored that has already been re-read.  This is not a standard Stream behavior, but can be useful
 201      * in a standalone mode.
 202      */
 203     public void reset() {
 204         for (ByteBuffer bb : bbs) {
 205             bb.position(0);
 206         }
 207         if (bbs.isEmpty()) {
 208             curr = null;
 209             idx = 0;
 210         } else {
 211             curr=bbs.get(0);
 212             idx=1;
 213         }
 214     }
 215
 216     /*
 217      * Ring Functions.  Reuse allocated memory
 218      */
 219     private ByteBuffer ringGet() {
 220         ByteBuffer bb = null;
 221         synchronized(ring) {
 222             bb=ring[start];
 223             ring[start]=null;
 224             if (bb!=null && ++start>15)start=0;
 225         }
 226         if (bb==null) {
 227             bb=ByteBuffer.allocate(DEFAULT_CHUNK);
 228         } else {
 229             bb.clear();// refresh reused buffer
 230         }
 231         return bb;
 232     }
 233
 234     private void ringPut(ByteBuffer bb) {
 235         synchronized(ring) {
 236             ring[end]=bb; // if null or not, BB will just be Garbage collected
 237             if (++end>15)end=0;
 238         }
 239     }
 240
 241 }