View Javadoc

    1   /***
    2    *
    3    * Copyright 2004 Protique Ltd
    4    *
    5    * Licensed under the Apache License, Version 2.0 (the "License");
    6    * you may not use this file except in compliance with the License.
    7    * You may obtain a copy of the License at
    8    *
    9    * http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   **/
    18
    19  package org.codehaus.activemq.message.util;
    20  import java.util.LinkedList;
    21  import java.util.List;
    22  import org.apache.commons.logging.Log;
    23  import org.apache.commons.logging.LogFactory;
    24  import org.codehaus.activemq.message.Packet;
    25
    26  /***
    27   * MemoryBoundedQueue is a queue bounded by memory usage for Packets
    28   *
    29   * @version $Revision: 1.2 $
    30   */
    31  public class MemoryBoundedQueue implements BoundedPacketQueue {
    32      private MemoryBoundedQueueManager queueManager;
    33      private String name;
    34      private boolean stopped = false;
    35      private boolean closed = false;
    36      private long memoryUsedByThisQueue;
    37      private Object outLock = new Object();
    38      private Object inLock = new Object();
    39      private LinkedList internalList = new LinkedList();
    40      private static final int WAIT_TIMEOUT = 100;
    41      private static final Log log = LogFactory.getLog(MemoryBoundedQueueManager.class);
    42
    43      /***
    44       * Constructor
    45       *
    46       * @param name
    47       * @param manager
    48       */
    49      MemoryBoundedQueue(String name, MemoryBoundedQueueManager manager) {
    50          this.name = name;
    51          this.queueManager = manager;
    52      }
    53
    54      /***
    55       * @return the name of this MemoryBoundedQueue
    56       */
    57      public String getName() {
    58          return name;
    59      }
    60
    61      /***
    62       * @return a pretty print of this queue
    63       */
    64      public String toString() {
    65          return "" + name + " , cardinality = " + size() + " memory usage = " + memoryUsedByThisQueue;
    66      }
    67
    68      /***
    69       * @return the number of items held by this queue
    70       */
    71      public int size() {
    72          return internalList.size();
    73      }
    74
    75      /***
    76       * @return an aproximation the memory used by this queue
    77       */
    78      public long getLocalMemoryUsedByThisQueue() {
    79          return memoryUsedByThisQueue;
    80      }
    81
    82      /***
    83       * close and remove this queue from the MemoryBoundedQueueManager
    84       */
    85      public void close() {
    86          try {
    87              clear();
    88              closed = true;
    89              synchronized (outLock) {
    90                  outLock.notifyAll();
    91              }
    92              synchronized (inLock) {
    93                  inLock.notifyAll();
    94              }
    95          }
    96          catch (Throwable e) {
    97              e.printStackTrace();
    98          }
    99          finally {
    100             queueManager.removeMemoryBoundedQueue(getName());
    101         }
    102     }
    103
    104     /***
    105      * Enqueue a Packet without checking memory usage limits
    106      *
    107      * @param packet
    108      */
    109     public void enqueueNoBlock(Packet packet) {
    110         if (!closed) {
    111             synchronized (outLock) {
    112                 internalList.add(packet);
    113                 incrementMemoryUsed(packet);
    114                 outLock.notify();
    115             }
    116         }
    117     }
    118
    119     /***
    120      * Enqueue a Packet to this queue
    121      *
    122      * @param packet
    123      */
    124     public void enqueue(Packet packet) {
    125         if (!queueManager.isFull()) {
    126             enqueueNoBlock(packet);
    127         }
    128         else {
    129             synchronized (inLock) {
    130                 try {
    131                     while (queueManager.isFull() && !closed) {
    132                         inLock.wait(WAIT_TIMEOUT);
    133                     }
    134                 }
    135                 catch (InterruptedException ie) {
    136                 }
    137             }
    138             enqueueNoBlock(packet);
    139         }
    140     }
    141
    142     /***
    143      * Enqueue a packet to the head of the queue with total disregard for memory constraints
    144      *
    145      * @param packet
    146      */
    147     public final void enqueueFirstNoBlock(Packet packet) {
    148         if (!closed) {
    149             synchronized (outLock) {
    150                 internalList.addFirst(packet);
    151                 incrementMemoryUsed(packet);
    152                 outLock.notify();
    153             }
    154         }
    155     }
    156
    157     /***
    158      * Enqueue a Packet to the head of the queue
    159      *
    160      * @param packet
    161      * @throws InterruptedException
    162      */
    163     public void enqueueFirst(Packet packet) throws InterruptedException {
    164         if (!queueManager.isFull()) {
    165             enqueueFirstNoBlock(packet);
    166         }
    167         else {
    168             synchronized (inLock) {
    169                 while (queueManager.isFull() && !closed) {
    170                     inLock.wait(WAIT_TIMEOUT);
    171                 }
    172             }
    173             enqueueFirstNoBlock(packet);
    174         }
    175     }
    176
    177     /***
    178      * @return the first dequeued Packet or blocks until one is available
    179      * @throws InterruptedException
    180      */
    181     public Packet dequeue() throws InterruptedException {
    182         Packet result = null;
    183         synchronized (outLock) {
    184             while (internalList.isEmpty() && !closed) {
    185                 outLock.wait(WAIT_TIMEOUT);
    186             }
    187             result = dequeueNoWait();
    188         }
    189         return result;
    190     }
    191
    192     /***
    193      * Dequeues a Packet from the head of the queue
    194      *
    195      * @param timeInMillis time to wait for a Packet to be available
    196      * @return the first Packet or null if none available within <I>timeInMillis </I>
    197      * @throws InterruptedException
    198      */
    199     public Packet dequeue(long timeInMillis) throws InterruptedException {
    200         if (timeInMillis == 0) {
    201             return dequeue();
    202         }
    203         else {
    204             synchronized (outLock) {
    205                 // if timeInMillis is less than zero assume nowait
    206                 if (timeInMillis >= 0) {
    207                     if (internalList.isEmpty() && !closed) {
    208                         outLock.wait(timeInMillis);
    209                     }
    210                 }
    211                 return dequeueNoWait();
    212             }
    213         }
    214     }
    215
    216     /***
    217      * dequeues a Packet from the head of the queue
    218      *
    219      * @return the Packet at the head of the queue or null, if none is available
    220      * @throws InterruptedException
    221      */
    222     public Packet dequeueNoWait() throws InterruptedException {
    223         Packet packet = null;
    224         if (stopped) {
    225             synchronized (outLock) {
    226                 while (stopped && !closed) {
    227                     outLock.wait(WAIT_TIMEOUT);
    228                 }
    229             }
    230         }
    231         synchronized (outLock) {
    232             if (!internalList.isEmpty()) {
    233                 packet = (Packet) internalList.removeFirst();
    234                 decrementMemoryUsed(packet);
    235             }
    236         }
    237         return packet;
    238     }
    239
    240     /***
    241      * @return true if the queue is enabled for dequeing (default = true)
    242      */
    243     public boolean isStarted() {
    244         return stopped == false;
    245     }
    246
    247     /***
    248      * disable dequeueing
    249      */
    250     public void stop() {
    251         synchronized (outLock) {
    252             stopped = true;
    253         }
    254     }
    255
    256     /***
    257      * enable dequeueing
    258      */
    259     public void start() {
    260         stopped = false;
    261         synchronized (outLock) {
    262             outLock.notifyAll();
    263         }
    264         synchronized (inLock) {
    265             inLock.notifyAll();
    266         }
    267     }
    268
    269     /***
    270      * Remove a packet from the queue
    271      *
    272      * @param packet
    273      * @return true if the packet was found
    274      */
    275     public boolean remove(Packet packet) {
    276         boolean result = false;
    277         synchronized (inLock) {
    278             if (!internalList.isEmpty()) {
    279                 result = internalList.remove(packet);
    280             }
    281             if (result) {
    282                 decrementMemoryUsed(packet);
    283                 if (!queueManager.isFull()) {
    284                     inLock.notify();
    285                 }
    286             }
    287         }
    288         return result;
    289     }
    290
    291     /***
    292      * remove any Packets in the queue
    293      */
    294     public void clear() {
    295         synchronized (inLock) {
    296             while (!internalList.isEmpty()) {
    297                 Packet packet = (Packet) internalList.removeFirst();
    298                 decrementMemoryUsed(packet);
    299             }
    300             inLock.notify();
    301         }
    302     }
    303
    304     /***
    305      * @return true if the queue is empty
    306      */
    307     public boolean isEmpty() {
    308         return internalList.isEmpty();
    309     }
    310
    311     /***
    312      * retrieve a Packet at an indexed position in the queue
    313      *
    314      * @param index
    315      * @return
    316      */
    317     public Packet get(int index) {
    318         return (Packet) internalList.get(index);
    319     }
    320
    321     /***
    322      * Retrieve a shallow copy of the contents as a list
    323      *
    324      * @return a list containing the bounded queue contents
    325      */
    326     public List getContents() {
    327         return (List) internalList.clone();
    328     }
    329
    330     private void incrementMemoryUsed(Packet packet) {
    331         if (packet != null) {
    332             memoryUsedByThisQueue += queueManager.incrementMemoryUsed(packet);
    333         }
    334     }
    335
    336     private void decrementMemoryUsed(Packet packet) {
    337         if (packet != null) {
    338             memoryUsedByThisQueue -= queueManager.decrementMemoryUsed(packet);
    339         }
    340     }
    341 }