Refactoring updates (job queue changes, enqueue collections..)

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/broker-queue-refactor@660490 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 81 insertions, 20 deletions

diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java b/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
index f1a7d4970a..c60b9ee0cb 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
@@ -464,15 +464,49 @@ public final class AMQShortString implements CharSequence, Comparable<AMQShortSt
             return false;
         }
 
-        if ((_hashCode != 0) && (otherString._hashCode != 0) && (_hashCode != otherString._hashCode))
+        final int hashCode = _hashCode;
+
+        final int otherHashCode = otherString._hashCode;
+
+        if ((hashCode != 0) && (otherHashCode != 0) && (hashCode != otherHashCode))
         {
             return false;
         }
 
+        final int length = _length;
+
+        if(length != otherString._length)
+        {
+            return false;
+        }
 
 
-        return (_offset == 0 && otherString._offset == 0 && _length == _data.length && otherString._length == otherString._data.length && Arrays.equals(_data,otherString._data))
-                || Arrays.equals(getBytes(),otherString.getBytes());
+        final byte[] data = _data;
+
+        final byte[] otherData = otherString._data;
+
+        final int offset = _offset;
+
+        final int otherOffset = otherString._offset;
+
+        if(offset == 0 && otherOffset == 0 && length == data.length && length == otherData.length)
+        {
+            return Arrays.equals(data, otherData);
+        }
+        else
+        {
+            int thisIdx = offset;
+            int otherIdx = otherOffset;
+            for(int i = length;  i-- != 0; )
+            {
+                if(!(data[thisIdx++] == otherData[otherIdx++]))
+                {
+                    return false;
+                }
+            }
+        }
+
+        return true;
 
     }
 
diff --git a/java/common/src/main/java/org/apache/qpid/pool/ReadWriteJobQueue.java b/java/common/src/main/java/org/apache/qpid/pool/ReadWriteJobQueue.java
index 05141aea7b..8de0f93ce9 100644
--- a/java/common/src/main/java/org/apache/qpid/pool/ReadWriteJobQueue.java
+++ b/java/common/src/main/java/org/apache/qpid/pool/ReadWriteJobQueue.java
@@ -41,16 +41,16 @@ public class ReadWriteJobQueue extends AbstractQueue<Runnable> implements Blocki
 
     private final ReentrantLock _putLock = new ReentrantLock();
 
-    private final ConcurrentLinkedQueue<Job> _readJobQueue = new ConcurrentLinkedQueue<Job>();
+    private final ConcurrentLinkedQueue<ReadWriteRunnable> _readJobQueue = new ConcurrentLinkedQueue<ReadWriteRunnable>();
 
-    private final ConcurrentLinkedQueue<Job> _writeJobQueue = new ConcurrentLinkedQueue<Job>();
+    private final ConcurrentLinkedQueue<ReadWriteRunnable> _writeJobQueue = new ConcurrentLinkedQueue<ReadWriteRunnable>();
 
 
     private class ReadWriteJobIterator implements Iterator<Runnable>
     {
 
         private boolean _onReads;
-        private Iterator<Job> _iter = _writeJobQueue.iterator();
+        private Iterator<ReadWriteRunnable> _iter = _writeJobQueue.iterator();
 
         public boolean hasNext()
         {
@@ -112,12 +112,12 @@ public class ReadWriteJobQueue extends AbstractQueue<Runnable> implements Blocki
 
     public boolean offer(final Runnable runnable)
     {
-        final Job job = (Job) runnable;
+        final ReadWriteRunnable job = (ReadWriteRunnable) runnable;
         final ReentrantLock putLock = _putLock;
         putLock.lock();
         try
         {
-            if(job.isReadJob())
+            if(job.isRead())
             {
                 _readJobQueue.offer(job);
             }
@@ -147,13 +147,13 @@ public class ReadWriteJobQueue extends AbstractQueue<Runnable> implements Blocki
 
     public void put(final Runnable runnable) throws InterruptedException
     {
-        final Job job = (Job) runnable;
+        final ReadWriteRunnable job = (ReadWriteRunnable) runnable;
         final ReentrantLock putLock = _putLock;
         putLock.lock();
 
         try
         {
-            if(job.isReadJob())
+            if(job.isRead())
             {
                 _readJobQueue.offer(job);
             }
@@ -185,13 +185,13 @@ public class ReadWriteJobQueue extends AbstractQueue<Runnable> implements Blocki
 
     public boolean offer(final Runnable runnable, final long timeout, final TimeUnit unit) throws InterruptedException
     {
-        final Job job = (Job) runnable;
+        final ReadWriteRunnable job = (ReadWriteRunnable) runnable;
         final ReentrantLock putLock = _putLock;
         putLock.lock();
 
         try
         {
-            if(job.isReadJob())
+            if(job.isRead())
             {
                 _readJobQueue.offer(job);
             }
@@ -240,7 +240,7 @@ public class ReadWriteJobQueue extends AbstractQueue<Runnable> implements Blocki
                 throw ie;
             }
 
-            Job job = _writeJobQueue.poll();
+            ReadWriteRunnable job = _writeJobQueue.poll();
             if(job == null)
             {
                 job = _readJobQueue.poll();
@@ -266,7 +266,7 @@ public class ReadWriteJobQueue extends AbstractQueue<Runnable> implements Blocki
         final AtomicInteger count = _count;
         long nanos = unit.toNanos(timeout);
         takeLock.lockInterruptibly();
-        Job job = null;
+        ReadWriteRunnable job = null;
         try
         {
 
@@ -322,7 +322,7 @@ public class ReadWriteJobQueue extends AbstractQueue<Runnable> implements Blocki
         _takeLock.lock();
         try
         {
-            Job job;
+            ReadWriteRunnable job;
             while((job = _writeJobQueue.peek())!= null)
             {
                 c.add(job);
@@ -356,7 +356,7 @@ public class ReadWriteJobQueue extends AbstractQueue<Runnable> implements Blocki
         _takeLock.lock();
         try
         {
-            Job job;
+            ReadWriteRunnable job;
             while(total<=maxElements && (job = _writeJobQueue.peek())!= null)
             {
                 c.add(job);
@@ -391,7 +391,7 @@ public class ReadWriteJobQueue extends AbstractQueue<Runnable> implements Blocki
         {
             if(_count.get() > 0)
             {
-                Job job = _writeJobQueue.poll();
+                ReadWriteRunnable job = _writeJobQueue.poll();
                 if(job == null)
                 {
                     job = _readJobQueue.poll();
@@ -417,7 +417,7 @@ public class ReadWriteJobQueue extends AbstractQueue<Runnable> implements Blocki
         takeLock.lock();
         try
         {
-            Job job = _writeJobQueue.peek();
+            ReadWriteRunnable job = _writeJobQueue.peek();
             if(job == null)
             {
                 job = _readJobQueue.peek();
diff --git a/java/common/src/main/java/org/apache/qpid/pool/ReadWriteRunnable.java b/java/common/src/main/java/org/apache/qpid/pool/ReadWriteRunnable.java
new file mode 100644
index 0000000000..ad04a923e1
--- /dev/null
+++ b/java/common/src/main/java/org/apache/qpid/pool/ReadWriteRunnable.java
@@ -0,0 +1,27 @@
+package org.apache.qpid.pool;
+
+/*
+*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License.  You may obtain a copy of the License at
+*
+*   http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied.  See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*
+*/
+public interface ReadWriteRunnable extends Runnable
+{
+    boolean isRead();
+    boolean isWrite();
+}
diff --git a/java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java b/java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java
index d99973cffb..ce9c6ae4cb 100644
--- a/java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java
+++ b/java/common/src/main/java/org/apache/qpid/pool/ReferenceCountingExecutorService.java
@@ -110,7 +110,7 @@ public class ReferenceCountingExecutorService
      *
      * @return An executor service.
      */
-    ExecutorService acquireExecutorService()
+    public ExecutorService acquireExecutorService()
     {
         synchronized (_lock)
         {
@@ -140,7 +140,7 @@ public class ReferenceCountingExecutorService
      * Releases a reference to a shared executor service, decrementing the reference count. If the refence count falls
      * to zero, the executor service is shut down.
      */
-    void releaseExecutorService()
+    public void releaseExecutorService()
     {
         synchronized (_lock)
         {