svn commit: r661324 - in /incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server: AMQChannel.java queue/AMQQueue.java queue/SimpleAMQQueue.java

[rgodfrey]

Author: rgodfrey
Date: Thu May 29 05:54:06 2008
New Revision: 661324

URL: http://svn.apache.org/viewvc?rev=661324&view=rev
Log:
Temp fix out of order issue with async(sub)

Modified:
    
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
    
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
    
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java

Modified: 
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java?rev=661324&r1=661323&r2=661324&view=diff
==============================================================================
--- 
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
 (original)
+++ 
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/AMQChannel.java
 Thu May 29 05:54:06 2008
@@ -831,7 +831,7 @@
                 // may need to deliver queued messages
                 for (Subscription s : _tag2SubscriptionMap.values())
                 {
-                    s.getQueue().deliverAsync(s);
+                    s.getQueue().deliverAsync();
                 }
             }
         }

Modified: 
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java?rev=661324&r1=661323&r2=661324&view=diff
==============================================================================
--- 
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
 (original)
+++ 
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
 Thu May 29 05:54:06 2008
@@ -152,6 +152,8 @@
 
     void deliverAsync(final Subscription sub);
 
+    void deliverAsync();
+
 
     /**
      * ExistingExclusiveSubscription signals a failure to create a 
subscription, because an exclusive subscription

Modified: 
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
URL: 
http://svn.apache.org/viewvc/incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java?rev=661324&r1=661323&r2=661324&view=diff
==============================================================================
--- 
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
 (original)
+++ 
incubator/qpid/branches/broker-queue-refactor/java/broker/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
 Thu May 29 05:54:06 2008
@@ -245,12 +245,19 @@
             throw new ExistingExclusiveSubscription();
         }
 
-        if(exclusive && getConsumerCount() != 0)
+        if(exclusive)
         {
-            throw new ExistingSubscriptionPreventsExclusive();
+            if(getConsumerCount() != 0)
+            {
+                throw new ExistingSubscriptionPreventsExclusive();
+            }
+            else
+            {
+                _exclusiveSubscriber = subscription;
+
+            }
         }
 
-        setExclusiveSubscriber(subscription);
 
         _activeSubscriberCount.incrementAndGet();
         subscription.setStateListener(this);
@@ -271,7 +278,7 @@
         }
 
 
-        deliverAsync(subscription);
+        deliverAsync();
 
     }
 
@@ -765,7 +772,7 @@
                 _activeSubscriberCount.incrementAndGet();
 
             }
-            deliverAsync(sub);
+            deliverAsync();
         }
     }
 
@@ -1655,7 +1662,7 @@
         public void stateChanged(QueueEntry entry, QueueEntry.State oldSate, 
QueueEntry.State newState)
         {
             entry.removeStateChangeListener(this);
-            deliverAsync(_sub);
+            deliverAsync();
         }
     }
 }
\ No newline at end of file