qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rgodf...@apache.org
Subject svn commit: r1564703 [1/4] - in /qpid/branches/java-broker-amqp-1-0-management/java: broker-core/src/main/java/org/apache/qpid/server/consumer/ broker-core/src/main/java/org/apache/qpid/server/logging/actors/ broker-core/src/main/java/org/apache/qpid/s...
Date Wed, 05 Feb 2014 10:29:57 GMT
Author: rgodfrey
Date: Wed Feb  5 10:29:55 2014
New Revision: 1564703

URL: http://svn.apache.org/r1564703
Log:
rename subscription to consumer

Added:
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/
      - copied from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AbstractSubscriptionTarget.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/Consumer.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/Subscription.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/SubscriptionTarget.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedConsumerMessageGroupManager.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedSubscriptionMessageGroupManager.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueSubscription.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerList.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SubscriptionList.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/consumer/
      - copied from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/subscription/
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/subscription/MockSubscription.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/ConsumerMessagesTest.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/SubscriptionMessagesTest.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/queue/ConsumerListTest.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/queue/SubscriptionListTest.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
      - copied, changed from r1564601, qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/SubscriptionTarget_0_10.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/SubscriptionTarget_0_8.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SubscriptionTarget_1_0.java
    qpid/branches/java-broker-amqp-1-0-management/java/systests/src/main/java/org/apache/qpid/server/logging/ConsumerLoggingTest.java
      - copied, changed from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
Removed:
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractSubscriptionTarget.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/Subscription.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/SubscriptionTarget.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedSubscriptionMessageGroupManager.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueSubscription.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SubscriptionList.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/consumer/MockSubscription.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/SubscriptionMessagesTest.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/queue/SubscriptionListTest.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/subscription/
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/SubscriptionTarget_0_10.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/SubscriptionTarget_0_8.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SubscriptionTarget_1_0.java
    qpid/branches/java-broker-amqp-1-0-management/java/systests/src/main/java/org/apache/qpid/server/logging/SubscriptionLoggingTest.java
Modified:
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstance.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/Session.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/DefinedGroupMessageGroupManager.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/MessageGroupManager.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntryImpl.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueRunner.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SimpleAMQQueue.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SortedQueue.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SubFlushRunner.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/logging/messages/VirtualHostMessagesTest.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/queue/AMQPriorityQueueTest.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockAMQQueue.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/queue/MockQueueEntry.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/queue/QueueEntryImplTestBase.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/test/java/org/apache/qpid/server/queue/SimpleAMQQueueTest.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ExplicitAcceptDispositionChangeListener.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ImplicitAcceptDispositionChangeListener.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/MessageAcceptCompletionListener.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerConnectionDelegate.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQProtocolEngine.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ClientDeliveryMethod.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ExtractResendAndRequeue.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/RecordDeliveryMethod.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicConsumeMethodHandler.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/handler/BasicGetMethodHandler.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/AckTest.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/ExtractResendAndRequeueTest.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/InternalTestProtocolSession.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-0-8-protocol/src/test/java/org/apache/qpid/server/protocol/v0_8/QueueBrowserUsesNoAckTest.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/SendingLink_1_0.java
    qpid/branches/java-broker-amqp-1-0-management/java/broker-plugins/management-http/src/main/java/org/apache/qpid/server/management/plugin/servlet/rest/MessageServlet.java

Copied: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java (from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AbstractSubscriptionTarget.java)
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java?p2=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java&p1=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AbstractSubscriptionTarget.java&r1=1564581&r2=1564703&rev=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/AbstractSubscriptionTarget.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java Wed Feb  5 10:29:55 2014
@@ -18,20 +18,20 @@
  * under the License.
  *
  */
-package org.apache.qpid.server.subscription;
+package org.apache.qpid.server.consumer;
 
 import org.apache.qpid.server.util.StateChangeListener;
 
 import java.util.concurrent.atomic.AtomicReference;
 
-public abstract class AbstractSubscriptionTarget implements SubscriptionTarget
+public abstract class AbstractConsumerTarget implements ConsumerTarget
 {
 
     private final AtomicReference<State> _state;
-    private final AtomicReference<StateChangeListener<SubscriptionTarget, State>> _stateListener =
-            new AtomicReference<StateChangeListener<SubscriptionTarget, State>>();
+    private final AtomicReference<StateChangeListener<ConsumerTarget, State>> _stateListener =
+            new AtomicReference<StateChangeListener<ConsumerTarget, State>>();
 
-    protected AbstractSubscriptionTarget(final State initialState)
+    protected AbstractConsumerTarget(final State initialState)
     {
         _state = new AtomicReference<State>(initialState);
     }
@@ -46,7 +46,7 @@ public abstract class AbstractSubscripti
     {
         if(_state.compareAndSet(from, to))
         {
-            StateChangeListener<SubscriptionTarget, State> listener = _stateListener.get();
+            StateChangeListener<ConsumerTarget, State> listener = _stateListener.get();
             if(listener != null)
             {
                 listener.stateChanged(this, from, to);
@@ -60,12 +60,12 @@ public abstract class AbstractSubscripti
     }
 
 
-    public final void setStateListener(StateChangeListener<SubscriptionTarget, State> listener)
+    public final void setStateListener(StateChangeListener<ConsumerTarget, State> listener)
     {
         _stateListener.set(listener);
     }
 
-    public final StateChangeListener<SubscriptionTarget, State> getStateListener()
+    public final StateChangeListener<ConsumerTarget, State> getStateListener()
     {
         return _stateListener.get();
     }

Copied: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/Consumer.java (from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/Subscription.java)
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/Consumer.java?p2=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/Consumer.java&p1=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/Subscription.java&r1=1564581&r2=1564703&rev=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/Subscription.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/Consumer.java Wed Feb  5 10:29:55 2014
@@ -18,7 +18,7 @@
  * under the License.
  *
  */
-package org.apache.qpid.server.subscription;
+package org.apache.qpid.server.consumer;
 
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.qpid.AMQException;
@@ -27,7 +27,7 @@ import org.apache.qpid.server.message.Me
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.util.StateChangeListener;
 
-public interface Subscription
+public interface Consumer
 {
     AtomicLong SUB_ID_GENERATOR = new AtomicLong(0);
 
@@ -63,11 +63,11 @@ public interface Subscription
 
     AMQSessionModel getSessionModel();
 
-    MessageInstance.SubscriptionAcquiredState getOwningState();
+    MessageInstance.ConsumerAcquiredState getOwningState();
 
     void setNoLocal(boolean noLocal);
 
-    long getSubscriptionID();
+    long getId();
 
     boolean isSuspended();
 
@@ -101,7 +101,7 @@ public interface Subscription
 
     void restoreCredit(final MessageInstance queueEntry);
 
-    void setStateListener(final StateChangeListener<? extends Subscription, State> listener);
+    void setStateListener(final StateChangeListener<? extends Consumer, State> listener);
 
     public State getState();
 

Copied: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java (from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/SubscriptionTarget.java)
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java?p2=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java&p1=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/SubscriptionTarget.java&r1=1564581&r2=1564703&rev=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/subscription/SubscriptionTarget.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java Wed Feb  5 10:29:55 2014
@@ -18,7 +18,7 @@
  * under the License.
  *
  */
-package org.apache.qpid.server.subscription;
+package org.apache.qpid.server.consumer;
 
 import org.apache.qpid.AMQException;
 import org.apache.qpid.server.message.MessageInstance;
@@ -26,7 +26,7 @@ import org.apache.qpid.server.message.Se
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.util.StateChangeListener;
 
-public interface SubscriptionTarget
+public interface ConsumerTarget
 {
 
 
@@ -37,11 +37,11 @@ public interface SubscriptionTarget
 
     State getState();
 
-    void subscriptionRegistered(Subscription sub);
+    void consumerAdded(Consumer sub);
 
-    void subscriptionRemoved(Subscription sub);
+    void consumerRemoved(Consumer sub);
 
-    void setStateListener(StateChangeListener<SubscriptionTarget, State> listener);
+    void setStateListener(StateChangeListener<ConsumerTarget, State> listener);
 
     long getUnacknowledgedBytes();
 

Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java?rev=1564703&r1=1564702&r2=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/logging/actors/CurrentActor.java Wed Feb  5 10:29:55 2014
@@ -35,7 +35,6 @@ import org.apache.qpid.server.logging.Lo
  * 2) We can set new actors at the point we have enough information. i.e.
  * - Set a low level ConnectionActor when processing bytes from the wire.
  * - Set a ChannelActor when we are processing the frame
- * - Set a SubscriptionActor when we are handling the subscription.
  * <p/>
  * The code performing the logging need not worry about what type of actor is
  * currently set so can perform its logging. The resulting log entry though will

Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstance.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstance.java?rev=1564703&r1=1564702&r2=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstance.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstance.java Wed Feb  5 10:29:55 2014
@@ -25,7 +25,7 @@ import org.apache.qpid.AMQException;
 import org.apache.qpid.server.filter.Filterable;
 import org.apache.qpid.server.queue.QueueEntry;
 import org.apache.qpid.server.store.TransactionLogResource;
-import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.consumer.Consumer;
 import org.apache.qpid.server.txn.ServerTransaction;
 import org.apache.qpid.server.util.Action;
 import org.apache.qpid.server.util.StateChangeListener;
@@ -49,25 +49,25 @@ public interface MessageInstance
 
     boolean removeStateChangeListener(StateChangeListener<QueueEntry, State> listener);
 
-    boolean acquiredBySubscription();
+    boolean acquiredByConsumer();
 
-    boolean isAcquiredBy(Subscription subscription);
+    boolean isAcquiredBy(Consumer consumer);
 
     void setRedelivered();
 
     boolean isRedelivered();
 
-    Subscription getDeliveredSubscription();
+    Consumer getDeliveredConsumer();
 
     void reject();
 
-    boolean isRejectedBy(Subscription subscription);
+    boolean isRejectedBy(Consumer consumer);
 
     boolean getDeliveredToConsumer();
 
     boolean expired() throws AMQException;
 
-    boolean acquire(Subscription sub);
+    boolean acquire(Consumer sub);
 
     int getMaximumDeliveryCount();
 
@@ -148,7 +148,7 @@ public interface MessageInstance
         }
     }
 
-    public final class NonSubscriptionAcquiredState extends EntryState
+    public final class NonConsumerAcquiredState extends EntryState
     {
         public State getState()
         {
@@ -161,13 +161,13 @@ public interface MessageInstance
         }
     }
 
-    public final class SubscriptionAcquiredState extends EntryState
+    public final class ConsumerAcquiredState extends EntryState
     {
-        private final Subscription _subscription;
+        private final Consumer _consumer;
 
-        public SubscriptionAcquiredState(Subscription subscription)
+        public ConsumerAcquiredState(Consumer consumer)
         {
-            _subscription = subscription;
+            _consumer = consumer;
         }
 
 
@@ -176,14 +176,14 @@ public interface MessageInstance
             return State.ACQUIRED;
         }
 
-        public Subscription getSubscription()
+        public Consumer getConsumer()
         {
-            return _subscription;
+            return _consumer;
         }
 
         public String toString()
         {
-            return "{" + getState().name() + " : " + _subscription +"}";
+            return "{" + getState().name() + " : " + _consumer +"}";
         }
     }
 
@@ -191,7 +191,7 @@ public interface MessageInstance
     final static EntryState AVAILABLE_STATE = new AvailableState();
     final static EntryState DELETED_STATE = new DeletedState();
     final static EntryState DEQUEUED_STATE = new DequeuedState();
-    final static EntryState NON_SUBSCRIPTION_ACQUIRED_STATE = new NonSubscriptionAcquiredState();
+    final static EntryState NON_CONSUMER_ACQUIRED_STATE = new NonConsumerAcquiredState();
 
     boolean isAvailable();
 

Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/Session.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/Session.java?rev=1564703&r1=1564702&r2=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/Session.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/Session.java Wed Feb  5 10:29:55 2014
@@ -77,6 +77,6 @@ public interface Session extends Configu
                                                              CHANNEL_ID,
                                                              PRODUCER_FLOW_BLOCKED));
 
-    Collection<Consumer> getSubscriptions();
+    Collection<Consumer> getConsumers();
     Collection<Publisher> getPublishers();
 }

Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java?rev=1564703&r1=1564702&r2=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/ConsumerAdapter.java Wed Feb  5 10:29:55 2014
@@ -22,33 +22,32 @@ package org.apache.qpid.server.model.ada
 
 import java.util.Map;
 import org.apache.qpid.server.model.ConfiguredObject;
-import org.apache.qpid.server.model.Consumer;
 import org.apache.qpid.server.model.LifetimePolicy;
 import org.apache.qpid.server.model.State;
 import org.apache.qpid.server.model.Statistics;
 import org.apache.qpid.server.model.UUIDGenerator;
-import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.consumer.Consumer;
 
 import java.security.AccessControlException;
 import java.util.Collection;
 import java.util.Collections;
 
-public class ConsumerAdapter extends AbstractAdapter implements Consumer
+public class ConsumerAdapter extends AbstractAdapter implements org.apache.qpid.server.model.Consumer
 {
-    private final Subscription _subscription;
+    private final Consumer _consumer;
     private final QueueAdapter _queue;
     private final SessionAdapter _session;
     private final ConsumerStatistics _statistics;
 
     public ConsumerAdapter(final QueueAdapter queueAdapter, final SessionAdapter sessionAdapter,
-                           final Subscription subscription)
+                           final Consumer consumer)
     {
         super(UUIDGenerator.generateConsumerUUID(queueAdapter.getVirtualHost().getName(),
                                                queueAdapter.getName(),
-                                               subscription.getSessionModel().getConnectionModel().getRemoteAddressString(),
-                                               String.valueOf(subscription.getSessionModel().getChannelId()),
-                                               subscription.getName()), queueAdapter.getTaskExecutor());
-        _subscription = subscription;
+                                               consumer.getSessionModel().getConnectionModel().getRemoteAddressString(),
+                                               String.valueOf(consumer.getSessionModel().getChannelId()),
+                                               consumer.getName()), queueAdapter.getTaskExecutor());
+        _consumer = consumer;
         _queue = queueAdapter;
         _session = sessionAdapter;
         _statistics = new ConsumerStatistics();
@@ -57,7 +56,7 @@ public class ConsumerAdapter extends Abs
 
     public String getName()
     {
-        return _subscription.getName();
+        return _consumer.getName();
     }
 
     public String setName(final String currentName, final String desiredName)
@@ -107,7 +106,7 @@ public class ConsumerAdapter extends Abs
     @Override
     public Collection<String> getAttributeNames()
     {
-        return Consumer.AVAILABLE_ATTRIBUTES;
+        return org.apache.qpid.server.model.Consumer.AVAILABLE_ATTRIBUTES;
     }
 
     @Override
@@ -147,7 +146,7 @@ public class ConsumerAdapter extends Abs
         }
         else if(DISTRIBUTION_MODE.equals(name))
         {
-            return _subscription.acquires() ? "MOVE" : "COPY";
+            return _consumer.acquires() ? "MOVE" : "COPY";
         }
         else if(SETTLEMENT_MODE.equals(name))
         {
@@ -197,11 +196,11 @@ public class ConsumerAdapter extends Abs
         {
             if(name.equals(BYTES_OUT))
             {
-                return _subscription.getBytesOut();
+                return _consumer.getBytesOut();
             }
             else if(name.equals(MESSAGES_OUT))
             {
-                return _subscription.getMessagesOut();
+                return _consumer.getMessagesOut();
             }
             else if(name.equals(STATE_CHANGED))
             {
@@ -209,11 +208,11 @@ public class ConsumerAdapter extends Abs
             }
             else if(name.equals(UNACKNOWLEDGED_BYTES))
             {
-                return _subscription.getUnacknowledgedBytes();
+                return _consumer.getUnacknowledgedBytes();
             }
             else if(name.equals(UNACKNOWLEDGED_MESSAGES))
             {
-                return _subscription.getUnacknowledgedMessages();
+                return _consumer.getUnacknowledgedMessages();
             }
             return null;  // TODO - Implement
         }

Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java?rev=1564703&r1=1564702&r2=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/QueueAdapter.java Wed Feb  5 10:29:55 2014
@@ -35,7 +35,6 @@ import org.apache.qpid.server.binding.Bi
 import org.apache.qpid.server.configuration.IllegalConfigurationException;
 import org.apache.qpid.server.model.ConfiguredObject;
 import org.apache.qpid.server.model.ConfiguredObjectFinder;
-import org.apache.qpid.server.model.Consumer;
 import org.apache.qpid.server.model.Exchange;
 import org.apache.qpid.server.model.IllegalStateTransitionException;
 import org.apache.qpid.server.model.LifetimePolicy;
@@ -47,10 +46,11 @@ import org.apache.qpid.server.protocol.A
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.queue.*;
 import org.apache.qpid.server.store.DurableConfigurationStoreHelper;
-import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.consumer.Consumer;
 import org.apache.qpid.server.util.MapValueConverter;
 
-final class QueueAdapter extends AbstractAdapter implements Queue, AMQQueue.SubscriptionRegistrationListener, AMQQueue.NotificationListener
+final class QueueAdapter extends AbstractAdapter implements Queue,
+                                                            AMQQueue.ConsumerRegistrationListener, AMQQueue.NotificationListener
 {
     @SuppressWarnings("serial")
     static final Map<String, Type> ATTRIBUTE_TYPES = Collections.unmodifiableMap(new HashMap<String, Type>(){{
@@ -69,8 +69,8 @@ final class QueueAdapter extends Abstrac
     private final AMQQueue _queue;
     private final Map<Binding, BindingAdapter> _bindingAdapters =
             new HashMap<Binding, BindingAdapter>();
-    private Map<org.apache.qpid.server.subscription.Subscription, ConsumerAdapter> _consumerAdapters =
-            new HashMap<org.apache.qpid.server.subscription.Subscription, ConsumerAdapter>();
+    private Map<Consumer, ConsumerAdapter> _consumerAdapters =
+            new HashMap<Consumer, ConsumerAdapter>();
 
 
     private final VirtualHostAdapter _vhost;
@@ -84,7 +84,7 @@ final class QueueAdapter extends Abstrac
         addParent(org.apache.qpid.server.model.VirtualHost.class, virtualHostAdapter);
 
         _queue = queue;
-        _queue.addSubscriptionRegistrationListener(this);
+        _queue.addConsumerRegistrationListener(this);
         populateConsumers();
         _statistics = new QueueStatisticsAdapter(queue);
         _queue.setNotificationListener(this);
@@ -124,21 +124,21 @@ final class QueueAdapter extends Abstrac
 
     private void populateConsumers()
     {
-        Collection<org.apache.qpid.server.subscription.Subscription> actualSubscriptions = _queue.getConsumers();
+        Collection<Consumer> actualConsumers = _queue.getConsumers();
 
         synchronized (_consumerAdapters)
         {
-            Iterator<org.apache.qpid.server.subscription.Subscription> iter = _consumerAdapters.keySet().iterator();
-            for(org.apache.qpid.server.subscription.Subscription subscription : actualSubscriptions)
+            Iterator<Consumer> iter = _consumerAdapters.keySet().iterator();
+            for(Consumer consumer : actualConsumers)
             {
-                if(!_consumerAdapters.containsKey(subscription))
+                if(!_consumerAdapters.containsKey(consumer))
                 {
-                    SessionAdapter sessionAdapter = getSessionAdapter(subscription.getSessionModel());
-                    ConsumerAdapter adapter = new ConsumerAdapter(this, sessionAdapter, subscription);
-                    _consumerAdapters.put(subscription, adapter);
+                    SessionAdapter sessionAdapter = getSessionAdapter(consumer.getSessionModel());
+                    ConsumerAdapter adapter = new ConsumerAdapter(this, sessionAdapter, consumer);
+                    _consumerAdapters.put(consumer, adapter);
                     if (sessionAdapter != null)
                     { // Register ConsumerAdapter with the SessionAdapter.
-                        sessionAdapter.subscriptionRegistered(subscription, adapter);
+                        sessionAdapter.consumerRegistered(consumer, adapter);
                     }
                 }
             }
@@ -153,11 +153,11 @@ final class QueueAdapter extends Abstrac
         }
     }
 
-    public Collection<Consumer> getConsumers()
+    public Collection<org.apache.qpid.server.model.Consumer> getConsumers()
     {
         synchronized (_consumerAdapters)
         {
-            return new ArrayList<Consumer>(_consumerAdapters.values());
+            return new ArrayList<org.apache.qpid.server.model.Consumer>(_consumerAdapters.values());
         }
 
     }
@@ -502,7 +502,7 @@ final class QueueAdapter extends Abstrac
     @Override
     public <C extends ConfiguredObject> Collection<C> getChildren(Class<C> clazz)
     {
-        if(clazz == Consumer.class)
+        if(clazz == org.apache.qpid.server.model.Consumer.class)
         {
             return (Collection<C>) getConsumers();
         }
@@ -587,19 +587,19 @@ final class QueueAdapter extends Abstrac
         return _queue;
     }
 
-    public void subscriptionRegistered(final AMQQueue queue, final Subscription subscription)
+    public void consumerAdded(final AMQQueue queue, final Consumer consumer)
     {
         ConsumerAdapter adapter = null;
         synchronized (_consumerAdapters)
         {
-            if(!_consumerAdapters.containsKey(subscription))
+            if(!_consumerAdapters.containsKey(consumer))
             {
-                SessionAdapter sessionAdapter = getSessionAdapter(subscription.getSessionModel());
-                adapter = new ConsumerAdapter(this, sessionAdapter, subscription);
-                _consumerAdapters.put(subscription, adapter);
+                SessionAdapter sessionAdapter = getSessionAdapter(consumer.getSessionModel());
+                adapter = new ConsumerAdapter(this, sessionAdapter, consumer);
+                _consumerAdapters.put(consumer, adapter);
                 if (sessionAdapter != null)
                 { // Register ConsumerAdapter with the SessionAdapter.
-                    sessionAdapter.subscriptionRegistered(subscription, adapter);
+                    sessionAdapter.consumerRegistered(consumer, adapter);
                 }
             }
         }
@@ -609,20 +609,20 @@ final class QueueAdapter extends Abstrac
         }
     }
 
-    public void subscriptionUnregistered(final AMQQueue queue, final Subscription subscription)
+    public void consumerRemoved(final AMQQueue queue, final Consumer consumer)
     {
         ConsumerAdapter adapter = null;
 
         synchronized (_consumerAdapters)
         {
-            adapter = _consumerAdapters.remove(subscription);
+            adapter = _consumerAdapters.remove(consumer);
         }
         if(adapter != null)
         {
-            SessionAdapter sessionAdapter = getSessionAdapter(subscription.getSessionModel());
+            SessionAdapter sessionAdapter = getSessionAdapter(consumer.getSessionModel());
             if (sessionAdapter != null)
             { // Unregister ConsumerAdapter with the SessionAdapter.
-                sessionAdapter.subscriptionUnregistered(subscription);
+                sessionAdapter.consumerUnregistered(consumer);
             }
             childRemoved(adapter);
         }

Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java?rev=1564703&r1=1564702&r2=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/model/adapter/SessionAdapter.java Wed Feb  5 10:29:55 2014
@@ -34,9 +34,8 @@ import org.apache.qpid.server.model.Publ
 import org.apache.qpid.server.model.Session;
 import org.apache.qpid.server.model.State;
 import org.apache.qpid.server.model.Statistics;
-import org.apache.qpid.server.model.Consumer;
 import org.apache.qpid.server.model.UUIDGenerator;
-import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.consumer.Consumer;
 import org.apache.qpid.server.configuration.updater.TaskExecutor;
 import org.apache.qpid.server.protocol.AMQSessionModel;
 
@@ -47,7 +46,7 @@ final class SessionAdapter extends Abstr
 
     private AMQSessionModel _session;
     private SessionStatistics _statistics;
-    private Map<Subscription, ConsumerAdapter> _consumerAdapters = new HashMap<Subscription, ConsumerAdapter>();
+    private Map<Consumer, ConsumerAdapter> _consumerAdapters = new HashMap<Consumer, ConsumerAdapter>();
 
     public SessionAdapter(final AMQSessionModel session, TaskExecutor taskExecutor)
     {
@@ -56,11 +55,11 @@ final class SessionAdapter extends Abstr
         _statistics = new SessionStatistics();
     }
 
-    public Collection<Consumer> getSubscriptions()
+    public Collection<org.apache.qpid.server.model.Consumer> getConsumers()
     {
         synchronized (_consumerAdapters)
         {
-            return new ArrayList<Consumer>(_consumerAdapters.values());
+            return new ArrayList<org.apache.qpid.server.model.Consumer>(_consumerAdapters.values());
         }
     }
 
@@ -119,29 +118,29 @@ final class SessionAdapter extends Abstr
     }
 
     /**
-     * Register a ConsumerAdapter (Subscription) with this Session keyed by the Subscription.
-     * @param subscription the org.apache.qpid.server.subscription.Subscription used to key the ConsumerAdapter.
+     * Register a ConsumerAdapter with this Session keyed by the Consumer.
+     * @param consumer the org.apache.qpid.server.consumer.Consumer used to key the ConsumerAdapter.
      * @param adapter the registered ConsumerAdapter.
      */
-    void subscriptionRegistered(Subscription subscription, ConsumerAdapter adapter)
+    void consumerRegistered(Consumer consumer, ConsumerAdapter adapter)
     {
         synchronized (_consumerAdapters)
         {
-            _consumerAdapters.put(subscription, adapter);
+            _consumerAdapters.put(consumer, adapter);
         }
         childAdded(adapter);
     }
 
     /**
-     * Unregister a ConsumerAdapter (Subscription) with this Session keyed by the Subscription.
-     * @param subscription the org.apache.qpid.server.subscription.Subscription used to key the ConsumerAdapter.
+     * Unregister a ConsumerAdapter  with this Session keyed by the Consumer.
+     * @param consumer the org.apache.qpid.server.consumer.Consumer used to key the ConsumerAdapter.
      */
-    void subscriptionUnregistered(Subscription subscription)
+    void consumerUnregistered(Consumer consumer)
     {
         ConsumerAdapter adapter = null;
         synchronized (_consumerAdapters)
         {
-            adapter = _consumerAdapters.remove(subscription);
+            adapter = _consumerAdapters.remove(consumer);
         }
         if (adapter != null)
         {
@@ -188,9 +187,9 @@ final class SessionAdapter extends Abstr
     @Override
     public <C extends ConfiguredObject> Collection<C> getChildren(Class<C> clazz)
     {
-        if(clazz == Consumer.class)
+        if(clazz == org.apache.qpid.server.model.Consumer.class)
         {
-            return (Collection<C>) getSubscriptions();
+            return (Collection<C>) getConsumers();
         }
         else if(clazz == Publisher.class)
         {

Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java?rev=1564703&r1=1564702&r2=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AMQQueue.java Wed Feb  5 10:29:55 2014
@@ -31,8 +31,8 @@ import org.apache.qpid.server.message.Se
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.security.AuthorizationHolder;
 import org.apache.qpid.server.store.TransactionLogResource;
-import org.apache.qpid.server.subscription.Subscription;
-import org.apache.qpid.server.subscription.SubscriptionTarget;
+import org.apache.qpid.server.consumer.Consumer;
+import org.apache.qpid.server.consumer.ConsumerTarget;
 import org.apache.qpid.server.util.Action;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 
@@ -83,27 +83,27 @@ public interface AMQQueue extends Compar
 
     VirtualHost getVirtualHost();
 
-    Subscription registerSubscription(final SubscriptionTarget target, final FilterManager filters,
-                                      final Class<? extends ServerMessage> messageClass,
-                                      final String consumerName, EnumSet<Subscription.Option> options) throws AMQException;
+    Consumer addConsumer(final ConsumerTarget target, final FilterManager filters,
+                         final Class<? extends ServerMessage> messageClass,
+                         final String consumerName, EnumSet<Consumer.Option> options) throws AMQException;
 
-    Collection<Subscription> getConsumers();
+    Collection<Consumer> getConsumers();
 
-    interface SubscriptionRegistrationListener
+    interface ConsumerRegistrationListener
     {
-        void subscriptionRegistered(AMQQueue queue, Subscription subscription);
-        void subscriptionUnregistered(AMQQueue queue, Subscription subscription);
+        void consumerAdded(AMQQueue queue, Consumer consumer);
+        void consumerRemoved(AMQQueue queue, Consumer consumer);
     }
 
-    void addSubscriptionRegistrationListener(SubscriptionRegistrationListener listener);
-    void removeSubscriptionRegistrationListener(SubscriptionRegistrationListener listener);
+    void addConsumerRegistrationListener(ConsumerRegistrationListener listener);
+    void removeConsumerRegistrationListener(ConsumerRegistrationListener listener);
 
 
     int getConsumerCount();
 
     int getActiveConsumerCount();
 
-    boolean hasExclusiveSubscriber();
+    boolean hasExclusiveConsumer();
 
     boolean isUnused();
 
@@ -126,11 +126,11 @@ public interface AMQQueue extends Compar
 
     void requeue(QueueEntry entry);
 
-    void dequeue(QueueEntry entry, Subscription sub);
+    void dequeue(QueueEntry entry, Consumer sub);
 
     void decrementUnackedMsgCount(QueueEntry queueEntry);
 
-    boolean resend(final QueueEntry entry, final Subscription subscription) throws AMQException;
+    boolean resend(final QueueEntry entry, final Consumer consumer) throws AMQException;
 
     void addQueueDeleteTask(Action<AMQQueue> task);
     void removeQueueDeleteTask(Action<AMQQueue> task);
@@ -209,7 +209,7 @@ public interface AMQQueue extends Compar
 
     Set<NotificationCheck> getNotificationChecks();
 
-    void flushSubscription(final Subscription sub) throws AMQException;
+    void flushConsumer(final Consumer sub) throws AMQException;
 
     void deliverAsync();
 
@@ -227,43 +227,43 @@ public interface AMQQueue extends Compar
     void checkCapacity(AMQSessionModel channel);
 
     /**
-     * ExistingExclusiveSubscription signals a failure to create a subscription, because an exclusive subscription
+     * ExistingExclusiveConsumer signals a failure to create a consumer, because an exclusive consumer
      * already exists.
      *
      * <p/><table id="crc"><caption>CRC Card</caption>
      * <tr><th> Responsibilities <th> Collaborations
-     * <tr><td> Represent failure to create a subscription, because an exclusive subscription already exists.
+     * <tr><td> Represent failure to create a consumer, because an exclusive consumer already exists.
      * </table>
      *
      * @todo Not an AMQP exception as no status code.
      *
      * @todo Move to top level, used outside this class.
      */
-    static final class ExistingExclusiveSubscription extends AMQException
+    static final class ExistingExclusiveConsumer extends AMQException
     {
 
-        public ExistingExclusiveSubscription()
+        public ExistingExclusiveConsumer()
         {
             super("");
         }
     }
 
     /**
-     * ExistingSubscriptionPreventsExclusive signals a failure to create an exclusive subscription, as a subscription
+     * ExistingConsumerPreventsExclusive signals a failure to create an exclusive consumer, as a consumer
      * already exists.
      *
      * <p/><table id="crc"><caption>CRC Card</caption>
      * <tr><th> Responsibilities <th> Collaborations
-     * <tr><td> Represent failure to create an exclusive subscription, as a subscription already exists.
+     * <tr><td> Represent failure to create an exclusive consumer, as a consumer already exists.
      * </table>
      *
      * @todo Not an AMQP exception as no status code.
      *
      * @todo Move to top level, used outside this class.
      */
-    static final class ExistingSubscriptionPreventsExclusive extends AMQException
+    static final class ExistingConsumerPreventsExclusive extends AMQException
     {
-        public ExistingSubscriptionPreventsExclusive()
+        public ExistingConsumerPreventsExclusive()
         {
             super("");
         }

Copied: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedConsumerMessageGroupManager.java (from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedSubscriptionMessageGroupManager.java)
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedConsumerMessageGroupManager.java?p2=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedConsumerMessageGroupManager.java&p1=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedSubscriptionMessageGroupManager.java&r1=1564581&r2=1564703&rev=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedSubscriptionMessageGroupManager.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/AssignedConsumerMessageGroupManager.java Wed Feb  5 10:29:55 2014
@@ -20,7 +20,7 @@
  */
 package org.apache.qpid.server.queue;
 
-import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.consumer.Consumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -28,16 +28,16 @@ import java.util.Iterator;
 import java.util.concurrent.ConcurrentHashMap;
 
 
-public class AssignedSubscriptionMessageGroupManager implements MessageGroupManager
+public class AssignedConsumerMessageGroupManager implements MessageGroupManager
 {
-    private static final Logger _logger = LoggerFactory.getLogger(AssignedSubscriptionMessageGroupManager.class);
+    private static final Logger _logger = LoggerFactory.getLogger(AssignedConsumerMessageGroupManager.class);
 
 
     private final String _groupId;
-    private final ConcurrentHashMap<Integer, QueueSubscription> _groupMap = new ConcurrentHashMap<Integer, QueueSubscription>();
+    private final ConcurrentHashMap<Integer, QueueConsumer> _groupMap = new ConcurrentHashMap<Integer, QueueConsumer>();
     private final int _groupMask;
 
-    public AssignedSubscriptionMessageGroupManager(final String groupId, final int maxGroups)
+    public AssignedConsumerMessageGroupManager(final String groupId, final int maxGroups)
     {
         _groupId = groupId;
         _groupMask = pow2(maxGroups)-1;
@@ -53,13 +53,13 @@ public class AssignedSubscriptionMessage
         return val;
     }
 
-    public QueueSubscription getAssignedSubscription(final QueueEntry entry)
+    public QueueConsumer getAssignedConsumer(final QueueEntry entry)
     {
         Object groupVal = entry.getMessage().getMessageHeader().getHeader(_groupId);
         return groupVal == null ? null : _groupMap.get(groupVal.hashCode() & _groupMask);
     }
 
-    public boolean acceptMessage(QueueSubscription sub, QueueEntry entry)
+    public boolean acceptMessage(QueueConsumer sub, QueueEntry entry)
     {
         if(assignMessage(sub, entry))
         {
@@ -71,7 +71,7 @@ public class AssignedSubscriptionMessage
         }
     }
 
-    private boolean assignMessage(QueueSubscription sub, QueueEntry entry)
+    private boolean assignMessage(QueueConsumer sub, QueueEntry entry)
     {
         Object groupVal = entry.getMessage().getMessageHeader().getHeader(_groupId);
         if(groupVal == null)
@@ -81,7 +81,7 @@ public class AssignedSubscriptionMessage
         else
         {
             Integer group = groupVal.hashCode() & _groupMask;
-            QueueSubscription assignedSub = _groupMap.get(group);
+            QueueConsumer assignedSub = _groupMap.get(group);
             if(assignedSub == sub)
             {
                 return true;
@@ -105,7 +105,7 @@ public class AssignedSubscriptionMessage
         }
     }
     
-    public QueueEntry findEarliestAssignedAvailableEntry(QueueSubscription sub)
+    public QueueEntry findEarliestAssignedAvailableEntry(QueueConsumer sub)
     {
         EntryFinder visitor = new EntryFinder(sub);
         sub.getQueue().visit(visitor);
@@ -115,9 +115,9 @@ public class AssignedSubscriptionMessage
     private class EntryFinder implements QueueEntryVisitor
     {
         private QueueEntry _entry;
-        private QueueSubscription _sub;
+        private QueueConsumer _sub;
 
-        public EntryFinder(final QueueSubscription sub)
+        public EntryFinder(final QueueConsumer sub)
         {
             _sub = sub;
         }
@@ -136,7 +136,7 @@ public class AssignedSubscriptionMessage
             }
 
             Integer group = groupId.hashCode() & _groupMask;
-            Subscription assignedSub = _groupMap.get(group);
+            Consumer assignedSub = _groupMap.get(group);
             if(assignedSub == _sub)
             {
                 _entry = entry;
@@ -154,9 +154,9 @@ public class AssignedSubscriptionMessage
         }
     }
 
-    public void clearAssignments(QueueSubscription sub)
+    public void clearAssignments(QueueConsumer sub)
     {
-        Iterator<QueueSubscription> subIter = _groupMap.values().iterator();
+        Iterator<QueueConsumer> subIter = _groupMap.values().iterator();
         while(subIter.hasNext())
         {
             if(subIter.next() == sub)

Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/DefinedGroupMessageGroupManager.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/DefinedGroupMessageGroupManager.java?rev=1564703&r1=1564702&r2=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/DefinedGroupMessageGroupManager.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/DefinedGroupMessageGroupManager.java Wed Feb  5 10:29:55 2014
@@ -20,7 +20,7 @@
  */
 package org.apache.qpid.server.queue;
 
-import org.apache.qpid.server.subscription.Subscription;
+import org.apache.qpid.server.consumer.Consumer;
 import org.apache.qpid.server.util.StateChangeListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,23 +38,23 @@ public class DefinedGroupMessageGroupMan
     private final String _groupId;
     private final String _defaultGroup;
     private final Map<Object, Group> _groupMap = new HashMap<Object, Group>();
-    private final SubscriptionResetHelper _resetHelper;
+    private final ConsumerResetHelper _resetHelper;
 
     private final class Group
     {
         private final Object _group;
-        private QueueSubscription _subscription;
+        private QueueConsumer _consumer;
         private int _activeCount;
 
-        private Group(final Object key, final QueueSubscription subscription)
+        private Group(final Object key, final QueueConsumer consumer)
         {
             _group = key;
-            _subscription = subscription;
+            _consumer = consumer;
         }
         
         public boolean add()
         {
-            if(_subscription != null)
+            if(_consumer != null)
             {
                 _activeCount++;
                 return true;
@@ -69,8 +69,8 @@ public class DefinedGroupMessageGroupMan
         {
             if(--_activeCount == 0)
             {
-                _resetHelper.resetSubPointersForGroups(_subscription, false);
-                _subscription = null;
+                _resetHelper.resetSubPointersForGroups(_consumer, false);
+                _consumer = null;
                 _groupMap.remove(_group);
             }
         }
@@ -100,12 +100,12 @@ public class DefinedGroupMessageGroupMan
 
         public boolean isValid()
         {
-            return !(_subscription == null || (_activeCount == 0 && _subscription.isClosed()));
+            return !(_consumer == null || (_activeCount == 0 && _consumer.isClosed()));
         }
 
-        public QueueSubscription getSubscription()
+        public QueueConsumer getConsumer()
         {
-            return _subscription;
+            return _consumer;
         }
 
         @Override
@@ -113,28 +113,28 @@ public class DefinedGroupMessageGroupMan
         {
             return "Group{" +
                     "_group=" + _group +
-                    ", _subscription=" + _subscription +
+                    ", _consumer=" + _consumer +
                     ", _activeCount=" + _activeCount +
                     '}';
         }
     }
 
-    public DefinedGroupMessageGroupManager(final String groupId, String defaultGroup, SubscriptionResetHelper resetHelper)
+    public DefinedGroupMessageGroupManager(final String groupId, String defaultGroup, ConsumerResetHelper resetHelper)
     {
         _groupId = groupId;
         _defaultGroup = defaultGroup;
         _resetHelper = resetHelper;
     }
     
-    public synchronized QueueSubscription getAssignedSubscription(final QueueEntry entry)
+    public synchronized QueueConsumer getAssignedConsumer(final QueueEntry entry)
     {
         Object groupId = getKey(entry);
 
         Group group = _groupMap.get(groupId);
-        return group == null || !group.isValid() ? null : group.getSubscription();
+        return group == null || !group.isValid() ? null : group.getConsumer();
     }
 
-    public synchronized boolean acceptMessage(final QueueSubscription sub, final QueueEntry entry)
+    public synchronized boolean acceptMessage(final QueueConsumer sub, final QueueEntry entry)
     {
         if(assignMessage(sub, entry))
         {
@@ -146,7 +146,7 @@ public class DefinedGroupMessageGroupMan
         }
     }
 
-    private boolean assignMessage(final QueueSubscription sub, final QueueEntry entry)
+    private boolean assignMessage(final QueueConsumer sub, final QueueEntry entry)
     {
         Object groupId = getKey(entry);
         Group group = _groupMap.get(groupId);
@@ -158,15 +158,15 @@ public class DefinedGroupMessageGroupMan
             _groupMap.put(groupId, group);
 
             // there's a small change that the group became empty between the point at which getNextAvailable() was
-            // called on the subscription, and when accept message is called... in that case we want to avoid delivering
+            // called on the consumer, and when accept message is called... in that case we want to avoid delivering
             // out of order
-            if(_resetHelper.isEntryAheadOfSubscription(entry, sub))
+            if(_resetHelper.isEntryAheadOfConsumer(entry, sub))
             {
                 return false;
             }
         }
 
-        Subscription assignedSub = group.getSubscription();
+        Consumer assignedSub = group.getConsumer();
 
         if(assignedSub == sub)
         {
@@ -179,7 +179,7 @@ public class DefinedGroupMessageGroupMan
         }
     }
 
-    public synchronized QueueEntry findEarliestAssignedAvailableEntry(final QueueSubscription sub)
+    public synchronized QueueEntry findEarliestAssignedAvailableEntry(final QueueConsumer sub)
     {
         EntryFinder visitor = new EntryFinder(sub);
         sub.getQueue().visit(visitor);
@@ -189,9 +189,9 @@ public class DefinedGroupMessageGroupMan
     private class EntryFinder implements QueueEntryVisitor
     {
         private QueueEntry _entry;
-        private QueueSubscription _sub;
+        private QueueConsumer _sub;
 
-        public EntryFinder(final QueueSubscription sub)
+        public EntryFinder(final QueueConsumer sub)
         {
             _sub = sub;
         }
@@ -206,7 +206,7 @@ public class DefinedGroupMessageGroupMan
             Object groupId = getKey(entry);
 
             Group group = _groupMap.get(groupId);
-            if(group != null && group.getSubscription() == _sub)
+            if(group != null && group.getConsumer() == _sub)
             {
                 _entry = entry;
                 return true;
@@ -224,7 +224,7 @@ public class DefinedGroupMessageGroupMan
     }
 
     
-    public void clearAssignments(final QueueSubscription sub)
+    public void clearAssignments(final QueueConsumer sub)
     {
     }
     

Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/MessageGroupManager.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/MessageGroupManager.java?rev=1564703&r1=1564702&r2=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/MessageGroupManager.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/MessageGroupManager.java Wed Feb  5 10:29:55 2014
@@ -20,22 +20,20 @@
  */
 package org.apache.qpid.server.queue;
 
-import org.apache.qpid.server.subscription.Subscription;
-
 public interface MessageGroupManager
 {
-    public interface SubscriptionResetHelper
+    public interface ConsumerResetHelper
     {
-        public void resetSubPointersForGroups(QueueSubscription subscription, boolean clearAssignments);
+        public void resetSubPointersForGroups(QueueConsumer consumer, boolean clearAssignments);
 
-        boolean isEntryAheadOfSubscription(QueueEntry entry, QueueSubscription sub);
+        boolean isEntryAheadOfConsumer(QueueEntry entry, QueueConsumer sub);
     }
 
-    QueueSubscription getAssignedSubscription(QueueEntry entry);
+    QueueConsumer getAssignedConsumer(QueueEntry entry);
 
-    boolean acceptMessage(QueueSubscription sub, QueueEntry entry);
+    boolean acceptMessage(QueueConsumer sub, QueueEntry entry);
 
-    QueueEntry findEarliestAssignedAvailableEntry(QueueSubscription sub);
+    QueueEntry findEarliestAssignedAvailableEntry(QueueConsumer sub);
 
-    void clearAssignments(QueueSubscription sub);
+    void clearAssignments(QueueConsumer sub);
 }

Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java?rev=1564703&r1=1564702&r2=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/OutOfOrderQueue.java Wed Feb  5 10:29:55 2014
@@ -20,8 +20,6 @@
  */
 package org.apache.qpid.server.queue;
 
-import org.apache.qpid.server.subscription.Subscription;
-import org.apache.qpid.server.queue.SubscriptionList;
 import org.apache.qpid.server.virtualhost.VirtualHost;
 
 import java.util.Map;
@@ -38,16 +36,16 @@ public abstract class OutOfOrderQueue ex
     }
 
     @Override
-    protected void checkSubscriptionsNotAheadOfDelivery(final QueueEntry entry)
+    protected void checkConsumersNotAheadOfDelivery(final QueueEntry entry)
     {
-        // check that all subscriptions are not in advance of the entry
-        SubscriptionList.SubscriptionNodeIterator subIter = getSubscriptionList().iterator();
+        // check that all consumers are not in advance of the entry
+        QueueConsumerList.ConsumerNodeIterator subIter = getConsumerList().iterator();
         while(subIter.advance() && !entry.isAcquired())
         {
-            final QueueSubscription subscription = subIter.getNode().getSubscription();
-            if(!subscription.isClosed())
+            final QueueConsumer consumer = subIter.getNode().getConsumer();
+            if(!consumer.isClosed())
             {
-                QueueContext context = subscription.getQueueContext();
+                QueueContext context = consumer.getQueueContext();
                 if(context != null)
                 {
                     QueueEntry released = context.getReleasedEntry();

Copied: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java (from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueSubscription.java)
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java?p2=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java&p1=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueSubscription.java&r1=1564581&r2=1564703&rev=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueSubscription.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java Wed Feb  5 10:29:55 2014
@@ -33,8 +33,8 @@ import org.apache.qpid.server.message.Me
 import org.apache.qpid.server.message.ServerMessage;
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.protocol.MessageConverterRegistry;
-import org.apache.qpid.server.subscription.Subscription;
-import org.apache.qpid.server.subscription.SubscriptionTarget;
+import org.apache.qpid.server.consumer.Consumer;
+import org.apache.qpid.server.consumer.ConsumerTarget;
 import org.apache.qpid.server.util.StateChangeListener;
 
 import java.text.MessageFormat;
@@ -47,16 +47,16 @@ import java.util.concurrent.locks.Reentr
 
 import static org.apache.qpid.server.logging.subjects.LogSubjectFormat.SUBSCRIPTION_FORMAT;
 
-class QueueSubscription<T extends SubscriptionTarget> implements Subscription
+class QueueConsumer<T extends ConsumerTarget> implements Consumer
 {
-    private static final Logger _logger = Logger.getLogger(QueueSubscription.class);
+    private static final Logger _logger = Logger.getLogger(QueueConsumer.class);
     private final AtomicBoolean _targetClosed = new AtomicBoolean(false);
     private final AtomicBoolean _closed = new AtomicBoolean(false);
-    private final long _subscriptionID;
+    private final long _id;
     private final AtomicReference<State> _state = new AtomicReference<State>(State.ACTIVE);
     private final Lock _stateChangeLock = new ReentrantLock();
     private final long _createTime = System.currentTimeMillis();
-    private final QueueEntry.SubscriptionAcquiredState _owningState = new QueueEntry.SubscriptionAcquiredState(this);
+    private final MessageInstance.ConsumerAcquiredState _owningState = new MessageInstance.ConsumerAcquiredState(this);
     private final boolean _acquires;
     private final boolean _seesRequeues;
     private final String _consumerName;
@@ -69,39 +69,39 @@ class QueueSubscription<T extends Subscr
     private SimpleAMQQueue _queue;
     private GenericActor _logActor;
 
-    static final EnumMap<SubscriptionTarget.State, State> STATE_MAP =
-            new EnumMap<SubscriptionTarget.State, State>(SubscriptionTarget.State.class);
+    static final EnumMap<ConsumerTarget.State, State> STATE_MAP =
+            new EnumMap<ConsumerTarget.State, State>(ConsumerTarget.State.class);
 
     static
     {
-        STATE_MAP.put(SubscriptionTarget.State.ACTIVE, State.ACTIVE);
-        STATE_MAP.put(SubscriptionTarget.State.SUSPENDED, State.SUSPENDED);
-        STATE_MAP.put(SubscriptionTarget.State.CLOSED, State.CLOSED);
+        STATE_MAP.put(ConsumerTarget.State.ACTIVE, State.ACTIVE);
+        STATE_MAP.put(ConsumerTarget.State.SUSPENDED, State.SUSPENDED);
+        STATE_MAP.put(ConsumerTarget.State.CLOSED, State.CLOSED);
     }
 
     private final T _target;
     private final SubFlushRunner _runner = new SubFlushRunner(this);
     private volatile QueueContext _queueContext;
-    private StateChangeListener<? extends Subscription, State> _stateListener = new StateChangeListener<Subscription, State>()
+    private StateChangeListener<? extends Consumer, State> _stateListener = new StateChangeListener<Consumer, State>()
     {
-        public void stateChanged(Subscription sub, State oldState, State newState)
+        public void stateChanged(Consumer sub, State oldState, State newState)
         {
             CurrentActor.get().message(SubscriptionMessages.STATE(newState.toString()));
         }
     };
     private boolean _noLocal;
 
-    QueueSubscription(final FilterManager filters,
-                      final Class<? extends ServerMessage> messageClass,
-                      final boolean acquires,
-                      final boolean seesRequeues,
-                      final String consumerName,
-                      final boolean isTransient,
-                      T target)
+    QueueConsumer(final FilterManager filters,
+                  final Class<? extends ServerMessage> messageClass,
+                  final boolean acquires,
+                  final boolean seesRequeues,
+                  final String consumerName,
+                  final boolean isTransient,
+                  T target)
     {
         _messageClass = messageClass;
         _sessionReference = target.getSessionModel().getConnectionReference();
-        _subscriptionID = SUB_ID_GENERATOR.getAndIncrement();
+        _id = SUB_ID_GENERATOR.getAndIncrement();
         _filters = filters;
         _acquires = acquires;
         _seesRequeues = seesRequeues;
@@ -109,23 +109,23 @@ class QueueSubscription<T extends Subscr
         _isTransient = isTransient;
         _target = target;
         _target.setStateListener(
-                new StateChangeListener<SubscriptionTarget, SubscriptionTarget.State>()
+                new StateChangeListener<ConsumerTarget, ConsumerTarget.State>()
                     {
                         @Override
-                        public void stateChanged(final SubscriptionTarget object,
-                                                 final SubscriptionTarget.State oldState,
-                                                 final SubscriptionTarget.State newState)
+                        public void stateChanged(final ConsumerTarget object,
+                                                 final ConsumerTarget.State oldState,
+                                                 final ConsumerTarget.State newState)
                         {
                             targetStateChanged(oldState, newState);
                         }
                     });
     }
 
-    private void targetStateChanged(final SubscriptionTarget.State oldState, final SubscriptionTarget.State newState)
+    private void targetStateChanged(final ConsumerTarget.State oldState, final ConsumerTarget.State newState)
     {
         if(oldState != newState)
         {
-            if(newState == SubscriptionTarget.State.CLOSED)
+            if(newState == ConsumerTarget.State.CLOSED)
             {
                 if(_targetClosed.compareAndSet(false,true))
                 {
@@ -138,7 +138,7 @@ class QueueSubscription<T extends Subscr
             }
         }
 
-        if(newState == SubscriptionTarget.State.CLOSED && oldState != newState && !_closed.get())
+        if(newState == ConsumerTarget.State.CLOSED && oldState != newState && !_closed.get())
         {
             try
             {
@@ -146,12 +146,12 @@ class QueueSubscription<T extends Subscr
             }
             catch (AMQException e)
             {
-                _logger.error("Unable to remove to remove subscription", e);
+                _logger.error("Unable to remove to remove consumer", e);
                 throw new RuntimeException(e);
             }
         }
-        final StateChangeListener<Subscription, State> stateListener =
-                (StateChangeListener<Subscription, State>) getStateListener();
+        final StateChangeListener<Consumer, State> stateListener =
+                (StateChangeListener<Consumer, State>) getStateListener();
         if(stateListener != null)
         {
             stateListener.stateChanged(this, STATE_MAP.get(oldState), STATE_MAP.get(newState));
@@ -202,8 +202,8 @@ class QueueSubscription<T extends Subscr
             try
             {
                 _target.close();
-                _target.subscriptionRemoved(this);
-                _queue.unregisterSubscription(this);
+                _target.consumerRemoved(this);
+                _queue.unregisterConsumer(this);
             }
             finally
             {
@@ -258,13 +258,13 @@ class QueueSubscription<T extends Subscr
     {
         if(getQueue() != null)
         {
-            throw new IllegalStateException("Attempt to set queue for subscription " + this + " to " + queue + "when already set to " + getQueue());
+            throw new IllegalStateException("Attempt to set queue for consumer " + this + " to " + queue + "when already set to " + getQueue());
         }
         _queue = queue;
 
         String queueString = new QueueLogSubject(_queue).toLogString();
 
-        _logActor = new GenericActor("[" + MessageFormat.format(SUBSCRIPTION_FORMAT, getSubscriptionID())
+        _logActor = new GenericActor("[" + MessageFormat.format(SUBSCRIPTION_FORMAT, getId())
                              + "("
                              // queueString is [vh(/{0})/qu({1}) ] so need to trim
                              //                ^                ^^
@@ -295,7 +295,7 @@ class QueueSubscription<T extends Subscr
     @Override
     public final void flush() throws AMQException
     {
-        getQueue().flushSubscription(this);
+        getQueue().flushConsumer(this);
     }
 
     @Override
@@ -309,17 +309,17 @@ class QueueSubscription<T extends Subscr
         return _runner;
     }
 
-    public final long getSubscriptionID()
+    public final long getId()
     {
-        return _subscriptionID;
+        return _id;
     }
 
-    public final StateChangeListener<? extends Subscription, State> getStateListener()
+    public final StateChangeListener<? extends Consumer, State> getStateListener()
     {
         return _stateListener;
     }
 
-    public final void setStateListener(StateChangeListener<? extends Subscription, State> listener)
+    public final void setStateListener(StateChangeListener<? extends Consumer, State> listener)
     {
         _stateListener = listener;
     }
@@ -430,7 +430,7 @@ class QueueSubscription<T extends Subscr
         return _createTime;
     }
 
-    public final MessageInstance.SubscriptionAcquiredState getOwningState()
+    public final MessageInstance.ConsumerAcquiredState getOwningState()
     {
         return _owningState;
     }

Copied: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerList.java (from r1564581, qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SubscriptionList.java)
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerList.java?p2=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerList.java&p1=qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SubscriptionList.java&r1=1564581&r2=1564703&rev=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/SubscriptionList.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerList.java Wed Feb  5 10:29:55 2014
@@ -24,28 +24,28 @@ import java.util.concurrent.atomic.Atomi
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 
-class SubscriptionList
+class QueueConsumerList
 {
-    private final SubscriptionNode _head = new SubscriptionNode();
+    private final ConsumerNode _head = new ConsumerNode();
 
-    private final AtomicReference<SubscriptionNode> _tail = new AtomicReference<SubscriptionNode>(_head);
-    private final AtomicReference<SubscriptionNode> _subNodeMarker = new AtomicReference<SubscriptionNode>(_head);
+    private final AtomicReference<ConsumerNode> _tail = new AtomicReference<ConsumerNode>(_head);
+    private final AtomicReference<ConsumerNode> _subNodeMarker = new AtomicReference<ConsumerNode>(_head);
     private final AtomicInteger _size = new AtomicInteger();
 
-    public static final class SubscriptionNode
+    public static final class ConsumerNode
     {
         private final AtomicBoolean _deleted = new AtomicBoolean();
-        private final AtomicReference<SubscriptionNode> _next = new AtomicReference<SubscriptionNode>();
-        private final QueueSubscription _sub;
+        private final AtomicReference<ConsumerNode> _next = new AtomicReference<ConsumerNode>();
+        private final QueueConsumer _sub;
 
-        public SubscriptionNode()
+        public ConsumerNode()
         {
             //used for sentinel head and dummy node construction
             _sub = null;
             _deleted.set(true);
         }
 
-        public SubscriptionNode(final QueueSubscription sub)
+        public ConsumerNode(final QueueConsumer sub)
         {
             //used for regular node construction
             _sub = sub;
@@ -57,12 +57,12 @@ class SubscriptionList
          *
          * @return the next non-deleted node, or null if none was found.
          */
-        public SubscriptionNode findNext()
+        public ConsumerNode findNext()
         {
-            SubscriptionNode next = nextNode();
+            ConsumerNode next = nextNode();
             while(next != null && next.isDeleted())
             {
-                final SubscriptionNode newNext = next.nextNode();
+                final ConsumerNode newNext = next.nextNode();
                 if(newNext != null)
                 {
                     //try to move our _next reference forward to the 'newNext'
@@ -86,7 +86,7 @@ class SubscriptionList
          *
          * @return the immediately next node in the structure, or null if at the tail.
          */
-        protected SubscriptionNode nextNode()
+        protected ConsumerNode nextNode()
         {
             return _next.get();
         }
@@ -94,10 +94,10 @@ class SubscriptionList
         /**
          * Used to initialise the 'next' reference. Will only succeed if the reference was not previously set.
          *
-         * @param node the SubscriptionNode to set as 'next'
+         * @param node the ConsumerNode to set as 'next'
          * @return whether the operation succeeded
          */
-        private boolean setNext(final SubscriptionNode node)
+        private boolean setNext(final ConsumerNode node)
         {
             return _next.compareAndSet(null, node);
         }
@@ -112,18 +112,18 @@ class SubscriptionList
             return _deleted.compareAndSet(false,true);
         }
 
-        public QueueSubscription getSubscription()
+        public QueueConsumer getConsumer()
         {
             return _sub;
         }
     }
 
-    private void insert(final SubscriptionNode node, final boolean count)
+    private void insert(final ConsumerNode node, final boolean count)
     {
         for (;;)
         {
-            SubscriptionNode tail = _tail.get();
-            SubscriptionNode next = tail.nextNode();
+            ConsumerNode tail = _tail.get();
+            ConsumerNode next = tail.nextNode();
             if (tail == _tail.get())
             {
                 if (next == null)
@@ -146,35 +146,35 @@ class SubscriptionList
         }
     }
 
-    public void add(final QueueSubscription sub)
+    public void add(final QueueConsumer sub)
     {
-        SubscriptionNode node = new SubscriptionNode(sub);
+        ConsumerNode node = new ConsumerNode(sub);
         insert(node, true);
     }
 
-    public boolean remove(final QueueSubscription sub)
+    public boolean remove(final QueueConsumer sub)
     {
-        SubscriptionNode prevNode = _head;
-        SubscriptionNode node = _head.nextNode();
+        ConsumerNode prevNode = _head;
+        ConsumerNode node = _head.nextNode();
 
         while(node != null)
         {
-            if(sub.equals(node.getSubscription()) && node.delete())
+            if(sub.equals(node.getConsumer()) && node.delete())
             {
                 _size.decrementAndGet();
 
-                SubscriptionNode tail = _tail.get();
+                ConsumerNode tail = _tail.get();
                 if(node == tail)
                 {
                     //we cant remove the last node from the structure for
                     //correctness reasons, however we have just 'deleted'
                     //the tail. Inserting an empty dummy node after it will
-                    //let us scavenge the node containing the Subscription.
-                    insert(new SubscriptionNode(), false);
+                    //let us scavenge the node containing the Consumer.
+                    insert(new ConsumerNode(), false);
                 }
 
                 //advance the next node reference in the 'prevNode' to scavenge
-                //the newly 'deleted' node for the Subscription.
+                //the newly 'deleted' node for the Consumer.
                 prevNode.findNext();
 
                 nodeMarkerCleanup(node);
@@ -189,9 +189,9 @@ class SubscriptionList
         return false;
     }
 
-    private void nodeMarkerCleanup(final SubscriptionNode node)
+    private void nodeMarkerCleanup(final ConsumerNode node)
     {
-        SubscriptionNode markedNode = _subNodeMarker.get();
+        ConsumerNode markedNode = _subNodeMarker.get();
         if(node == markedNode)
         {
             //if the marked node is the one we are removing, then
@@ -200,7 +200,7 @@ class SubscriptionList
             //into the list and find the next node to use.
             //Because we inserted a dummy if node was the
             //tail, markedNode.nextNode() can never be null.
-            SubscriptionNode dummy = new SubscriptionNode();
+            ConsumerNode dummy = new ConsumerNode();
             dummy.setNext(markedNode.nextNode());
 
             //if the CAS fails the marked node has changed, thus
@@ -219,53 +219,53 @@ class SubscriptionList
         }
     }
 
-    public boolean updateMarkedNode(final SubscriptionNode expected, final SubscriptionNode nextNode)
+    public boolean updateMarkedNode(final ConsumerNode expected, final ConsumerNode nextNode)
     {
         return _subNodeMarker.compareAndSet(expected, nextNode);
     }
 
     /**
-     * Get the current marked SubscriptionNode. This should only be used only to index into the list and find the next node
+     * Get the current marked ConsumerNode. This should only be used only to index into the list and find the next node
      * after the mark, since if the previously marked node was subsequently deleted the item returned may be a dummy node
      * with reference to the next node.
      *
      * @return the previously marked node (or a dummy if it was subsequently deleted)
      */
-    public SubscriptionNode getMarkedNode()
+    public ConsumerNode getMarkedNode()
     {
         return _subNodeMarker.get();
     }
 
 
-    public static class SubscriptionNodeIterator
+    public static class ConsumerNodeIterator
     {
-        private SubscriptionNode _lastNode;
+        private ConsumerNode _lastNode;
 
-        SubscriptionNodeIterator(SubscriptionNode startNode)
+        ConsumerNodeIterator(ConsumerNode startNode)
         {
             _lastNode = startNode;
         }
 
-        public SubscriptionNode getNode()
+        public ConsumerNode getNode()
         {
             return _lastNode;
         }
 
         public boolean advance()
         {
-            SubscriptionNode nextNode = _lastNode.findNext();
+            ConsumerNode nextNode = _lastNode.findNext();
             _lastNode = nextNode;
 
             return _lastNode != null;
         }
     }
 
-    public SubscriptionNodeIterator iterator()
+    public ConsumerNodeIterator iterator()
     {
-        return new SubscriptionNodeIterator(_head);
+        return new ConsumerNodeIterator(_head);
     }
 
-    public SubscriptionNode getHead()
+    public ConsumerNode getHead()
     {
         return _head;
     }

Modified: qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java
URL: http://svn.apache.org/viewvc/qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java?rev=1564703&r1=1564702&r2=1564703&view=diff
==============================================================================
--- qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java (original)
+++ qpid/branches/java-broker-amqp-1-0-management/java/broker-core/src/main/java/org/apache/qpid/server/queue/QueueEntry.java Wed Feb  5 10:29:55 2014
@@ -20,13 +20,7 @@
 */
 package org.apache.qpid.server.queue;
 
-import org.apache.qpid.AMQException;
-import org.apache.qpid.server.filter.Filterable;
 import org.apache.qpid.server.message.MessageInstance;
-import org.apache.qpid.server.subscription.Subscription;
-import org.apache.qpid.server.txn.ServerTransaction;
-import org.apache.qpid.server.util.Action;
-import org.apache.qpid.server.util.StateChangeListener;
 
 public interface QueueEntry extends MessageInstance, Comparable<QueueEntry>
 {



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message