veslibrary/ves_javalibrary/evel_javalib2/src/evel_javalibrary/att/com/RingBuffer.java

   1 package evel_javalibrary.att.com;\r
   2 /**************************************************************************//**\r
   3  * @file\r
   4  * RingBuffer class\r
   5  *\r
   6   * This file implements internal Ringbuffer for storing and\r
   7   *  forwarding events to Collector.\r
   8  *\r
   9  * License\r
  10  * -------\r
  11  * Unless otherwise specified, all software contained herein is\r
  12  * Licensed under the Apache License, Version 2.0 (the "License");\r
  13  * you may not use this file except in compliance with the License.\r
  14  * You may obtain a copy of the License at\r
  15  *        http://www.apache.org/licenses/LICENSE-2.0\r
  16  *\r
  17  * Unless required by applicable law or agreed to in writing, software\r
  18  * distributed under the License is distributed on an "AS IS" BASIS,\r
  19  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\r
  20  * See the License for the specific language governing permissions and\r
  21  * limitations under the License.\r
  22  *****************************************************************************/\r
  23 \r
  24 import java.util.concurrent.Semaphore;\r
  25 /*\r
  26  * Ringbuffer to store and Forward http(s) POST requests\r
  27  */\r
  28 public class RingBuffer {\r
  29 \r
  30             // message count semaphore\r
  31             public static Semaphore countsem;\r
  32             // space semaphore\r
  33             public static Semaphore spacesem;\r
  34             // lock semaphore\r
  35             public static Semaphore lock;\r
  36 \r
  37             public Object[] elements = null;\r
  38 \r
  39             public int capacity  = 0;\r
  40             public int writePos  = 0;\r
  41             public int available = 0;\r
  42 \r
  43             /*\r
  44              * Constructs Ringbuffer of specified capacity\r
  45              */\r
  46             public RingBuffer(int capacity) {\r
  47                 this.capacity = capacity;\r
  48                 this.elements = new Object[capacity];\r
  49                 countsem = new Semaphore(1);\r
  50                 spacesem = new Semaphore(capacity);\r
  51                 lock = new Semaphore(1);\r
  52             }\r
  53 \r
  54             //resets the positions\r
  55             public void reset() {\r
  56                 this.writePos = 0;\r
  57                 this.available = 0;\r
  58             }\r
  59 \r
  60             //returns available capacity\r
  61             public int remainingCapacity() {\r
  62                 return this.capacity - this.available;\r
  63             }\r
  64 \r
  65 \r
  66 \r
  67             //Puts Java object into ringbuffer\r
  68             public boolean put(Object element){\r
  69                 \r
  70                 boolean ret = false;\r
  71                 //acquire locks\r
  72                 try {\r
  73                                 spacesem.acquire();\r
  74                                 lock.acquire();\r
  75                         } catch (InterruptedException e) {\r
  76                                 // TODO Auto-generated catch block\r
  77                                 e.printStackTrace();\r
  78                         }\r
  79                 \r
  80 \r
  81                 //store object\r
  82                 if(available < capacity){\r
  83                     if(writePos >= capacity){\r
  84                         writePos = 0;\r
  85                     }\r
  86                     elements[writePos] = element;\r
  87                     writePos++;\r
  88                     available++;\r
  89                     ret = true;\r
  90                 }\r
  91                 \r
  92                 //release Locks\r
  93                     lock.release();\r
  94                         countsem.release();\r
  95 \r
  96 \r
  97                 return ret;\r
  98             }\r
  99 \r
 100             public int put(Object[] newElements){\r
 101                 return put(newElements, newElements.length);\r
 102             }\r
 103 \r
 104             public int put(Object[] newElements, int length){\r
 105                 //Acquire locks\r
 106                 try {\r
 107                                 spacesem.acquire();\r
 108                                 lock.acquire();\r
 109                         } catch (InterruptedException e) {\r
 110                                 // TODO Auto-generated catch block\r
 111                                 e.printStackTrace();\r
 112                         }\r
 113                 \r
 114                 int readPos = 0;\r
 115                 if(this.writePos > this.available){\r
 116                     //space above writePos is all empty\r
 117 \r
 118                     if(length <= this.capacity - this.writePos){\r
 119                         //space above writePos is sufficient to insert batch\r
 120 \r
 121                         for(;  readPos < length; readPos++){\r
 122                             this.elements[this.writePos++] = newElements[readPos];\r
 123                         }\r
 124                         this.available += readPos;\r
 125                         //release\r
 126                             lock.release();\r
 127                                 countsem.release();\r
 128                         return length;\r
 129 \r
 130                     } else {\r
 131                         //both space above writePos and below writePos is necessary to use\r
 132                         //to insert batch.\r
 133 \r
 134                         int lastEmptyPos = writePos - available;\r
 135 \r
 136                         for(; this.writePos < this.capacity; this.writePos++){\r
 137                             this.elements[this.writePos] = newElements[readPos++];\r
 138                         }\r
 139 \r
 140                         //fill into bottom of array too.\r
 141                         this.writePos = 0;\r
 142 \r
 143                         int endPos = Math.min(length - readPos, capacity - available - readPos);\r
 144                         for(;this.writePos < endPos; this.writePos++){\r
 145                             this.elements[this.writePos] = newElements[readPos++];\r
 146                         }\r
 147                         this.available += readPos;\r
 148                         //release\r
 149                             lock.release();\r
 150                                 countsem.release();\r
 151                         return readPos;\r
 152                     }\r
 153                 } else {\r
 154                     int endPos = this.capacity - this.available + this.writePos;\r
 155 \r
 156                     for(; this.writePos < endPos; this.writePos++){\r
 157                         this.elements[this.writePos] = newElements[readPos++];\r
 158                     }\r
 159                     this.available += readPos;\r
 160                     //release\r
 161                             lock.release();\r
 162                                 countsem.release();\r
 163 \r
 164                     return readPos;\r
 165                 }\r
 166 \r
 167             }\r
 168 \r
 169             /*\r
 170              * Takes a stored object in Ringbuffer and releases the space\r
 171              */\r
 172 \r
 173             public Object take() {\r
 174                 \r
 175                 Object nextObj;\r
 176                 //acquire lock\r
 177                 try {\r
 178                                 countsem.acquire();\r
 179                                 lock.acquire();\r
 180                         } catch (InterruptedException e) {\r
 181                                 // TODO Auto-generated catch block\r
 182                                 e.printStackTrace();\r
 183                         }\r
 184                 \r
 185                 if(available == 0){\r
 186                     nextObj = null;\r
 187                 }\r
 188                 else {\r
 189                   int nextSlot = writePos - available;\r
 190                   if(nextSlot < 0){\r
 191                     nextSlot += capacity;\r
 192                   }\r
 193                   nextObj = elements[nextSlot];\r
 194                   available--;\r
 195                 }\r
 196                 //releases object\r
 197                     lock.release();\r
 198                         spacesem.release();\r
 199                 \r
 200                 return nextObj;\r
 201             }\r
 202 \r
 203 \r
 204             public int take(Object[] into){\r
 205                 return take(into, into.length);\r
 206             }\r
 207 \r
 208 \r
 209             public int take(Object[] into, int length){\r
 210                 int intoPos = 0;\r
 211                 \r
 212                 //acquire lock\r
 213                 try {\r
 214                                 countsem.acquire();\r
 215                                 lock.acquire();\r
 216                         } catch (InterruptedException e) {\r
 217                                 // TODO Auto-generated catch block\r
 218                                 e.printStackTrace();\r
 219                         }\r
 220 \r
 221                 if(available <= writePos){\r
 222                     int nextPos= writePos - available;\r
 223                     int endPos   = nextPos + Math.min(available, length);\r
 224 \r
 225                     for(;nextPos < endPos; nextPos++){\r
 226                         into[intoPos++] = this.elements[nextPos];\r
 227                     }\r
 228                     this.available -= intoPos;\r
 229                     \r
 230                     //release\r
 231                             lock.release();\r
 232                                 countsem.release();\r
 233                                 \r
 234                     return intoPos;\r
 235                 } else {\r
 236                     int nextPos = writePos - available + capacity;\r
 237 \r
 238                     int leftInTop = capacity - nextPos;\r
 239                     if(length <= leftInTop){\r
 240                         //copy directly\r
 241                         for(; intoPos < length; intoPos++){\r
 242                             into[intoPos] = this.elements[nextPos++];\r
 243                         }\r
 244                         this.available -= length;\r
 245                             //release\r
 246                                     lock.release();\r
 247                                         countsem.release();\r
 248                         return length;\r
 249 \r
 250                     } else {\r
 251                         //copy top\r
 252                         for(; nextPos < capacity; nextPos++){\r
 253                             into[intoPos++] = this.elements[nextPos];\r
 254                         }\r
 255 \r
 256                         //copy bottom - from 0 to writePos\r
 257                         nextPos = 0;\r
 258                         int leftToCopy = length - intoPos;\r
 259                         int endPos = Math.min(writePos, leftToCopy);\r
 260 \r
 261                         for(;nextPos < endPos; nextPos++){\r
 262                             into[intoPos++] = this.elements[nextPos];\r
 263                         }\r
 264 \r
 265                         this.available -= intoPos;\r
 266                         \r
 267                             //release\r
 268                                     lock.release();\r
 269                                         countsem.release();\r
 270 \r
 271                         return intoPos;\r
 272                     }\r
 273                 }\r
 274             }\r
 275         \r
 276 }\r