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 }