Source code: org/activemq/ra/CircularQueue.java

   package org.activemq.ra;
   
   import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
   
   /**
    */
   public class CircularQueue {
   
       private final int size;
  
      private final SynchronizedBoolean stopping;
  
  
      // For pooling objects
      private final Object[] contents;
      final private Object mutex = new Object();
      //where the next worker to be supplied currently is.
      private int start=0;
      //where the next worker to be inserted will go
      private int end=0;
  
      public CircularQueue(int size, SynchronizedBoolean stopping) {
          this.size = size;
          contents = new Object[size];
          this.stopping = stopping;
      }
  
      public Object get() {
        synchronized(mutex) {
          while( true ) {
              Object ew = contents[start];
                  if (ew != null) {
                      start++;
              if(start == contents.length) {
                start=0;
              }
              return ew;
            } else {
              try {
              mutex.wait();
              if(stopping.get()) {
                return null;
              }
            } catch (InterruptedException e) {
              return null;
            }
            }
          }
        }
      }
  
      public void returnObject(Object worker) {
        synchronized(mutex) {
          contents[end++] = worker;
              if( end == contents.length) {
                  end=0;
              }
          mutex.notify();
        }
      }
  
      public int size() {
          return contents.length;
      }
  
      public void drain() {
          int i = 0;
          while (i < size) {
              if (get() != null) {
                  i++;
              }
          }
      }
  
  
      public void notifyWaiting() {
          synchronized(mutex) {
          mutex.notifyAll();
        }
      }
  
  }