diff --git a/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java b/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java
index 9fa88fb..b21f948 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/consumer/AbstractConsumerTarget.java
@@ -21,27 +21,33 @@
 package org.apache.qpid.server.consumer;
 
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.qpid.server.logging.LogSubject;
+import org.apache.qpid.server.logging.messages.SubscriptionMessages;
 import org.apache.qpid.server.message.MessageInstance;
 import org.apache.qpid.server.message.MessageInstanceConsumer;
+import org.apache.qpid.server.model.Consumer;
+import org.apache.qpid.server.queue.SuspendedConsumerLoggingTicker;
 import org.apache.qpid.server.transport.AMQPConnection;
 import org.apache.qpid.server.util.StateChangeListener;
 
-public abstract class AbstractConsumerTarget implements ConsumerTarget
+public abstract class AbstractConsumerTarget implements ConsumerTarget, LogSubject
 {
     private static final Logger LOGGER = LoggerFactory.getLogger(AbstractConsumerTarget.class);
+    protected static final String PULL_ONLY_CONSUMER = "x-pull-only";
     private final AtomicReference<State> _state;
 
     private final Set<StateChangeListener<ConsumerTarget, State>> _stateChangeListeners = new
@@ -49,12 +55,39 @@
 
     private final Lock _stateChangeLock = new ReentrantLock();
     private final AtomicInteger _stateActivates = new AtomicInteger();
+    private final boolean _isMultiQueue;
+    private final SuspendedConsumerLoggingTicker _suspendedConsumerLoggingTicker;
     private ConcurrentLinkedQueue<ConsumerMessageInstancePair> _queue = new ConcurrentLinkedQueue();
+    private final List<MessageInstanceConsumer> _consumers = new CopyOnWriteArrayList<>();
+
+    private final boolean _isPullOnly;
+    private Iterator<MessageInstanceConsumer> _pullIterator;
 
 
-    protected AbstractConsumerTarget(final State initialState)
+    protected AbstractConsumerTarget(final State initialState,
+                                     final boolean isPullOnly,
+                                     final boolean isMultiQueue,
+                                     final AMQPConnection<?> amqpConnection)
     {
         _state = new AtomicReference<State>(initialState);
+        _isPullOnly = isPullOnly;
+        _isMultiQueue = isMultiQueue;
+        _suspendedConsumerLoggingTicker = isMultiQueue
+                ? new SuspendedConsumerLoggingTicker(amqpConnection.getContextValue(Long.class, Consumer.SUSPEND_NOTIFICATION_PERIOD))
+                {
+                    @Override
+                    protected void log(final long period)
+                    {
+                        amqpConnection.getEventLogger().message(AbstractConsumerTarget.this, SubscriptionMessages.STATE(period));
+                    }
+                }
+                : null;
+
+    }
+
+    public boolean isMultiQueue()
+    {
+        return _isMultiQueue;
     }
 
     @Override
@@ -64,9 +97,8 @@
         {
             return false;
         }
-        if(hasMessagesToSend())
+        if(sendNextMessage())
         {
-            sendNextMessage();
             return true;
         }
         else
@@ -92,6 +124,28 @@
     protected abstract void processClosed();
 
     @Override
+    public void consumerAdded(final MessageInstanceConsumer sub)
+    {
+        _consumers.add(sub);
+    }
+
+    @Override
+    public void consumerRemoved(final MessageInstanceConsumer sub)
+    {
+        _consumers.remove(sub);
+        if(_consumers.isEmpty())
+        {
+            close();
+        }
+    }
+
+    public List<MessageInstanceConsumer> getConsumers()
+    {
+        return _consumers;
+    }
+
+
+    @Override
     public final boolean isSuspended()
     {
         return getSessionModel().getAMQPConnection().isMessageAssignmentSuspended() || isFlowSuspended();
@@ -144,9 +198,24 @@
             }
             else
             {
-                for (StateChangeListener<ConsumerTarget, State> listener : _stateChangeListeners)
+                if(!_stateChangeListeners.isEmpty())
                 {
-                    listener.stateChanged(this, from, to);
+                    for (StateChangeListener<ConsumerTarget, State> listener : _stateChangeListeners)
+                    {
+                        listener.stateChanged(this, from, to);
+                    }
+                }
+            }
+            if(_suspendedConsumerLoggingTicker != null)
+            {
+                if (to == State.SUSPENDED)
+                {
+                    _suspendedConsumerLoggingTicker.setStartTime(System.currentTimeMillis());
+                    getSessionModel().addTicker(_suspendedConsumerLoggingTicker);
+                }
+                else
+                {
+                    getSessionModel().removeTicker(_suspendedConsumerLoggingTicker);
                 }
             }
             return true;
@@ -178,6 +247,11 @@
         _stateChangeListeners.remove(listener);
     }
 
+    public boolean isPullOnly()
+    {
+        return _isPullOnly;
+    }
+
     public final boolean trySendLock()
     {
         return _stateChangeLock.tryLock();
@@ -208,12 +282,39 @@
     @Override
     public boolean hasMessagesToSend()
     {
-        return !_queue.isEmpty();
+        return !_queue.isEmpty() || (isPullOnly() && messagesAvailable());
+    }
+
+    private boolean messagesAvailable()
+    {
+        if(hasCredit())
+        {
+            for (MessageInstanceConsumer consumer : _consumers)
+            {
+                if (consumer.hasAvailableMessages())
+                {
+                    return true;
+                }
+            }
+        }
+        return false;
     }
 
     @Override
-    public void sendNextMessage()
+    public boolean sendNextMessage()
     {
+        if(isPullOnly())
+        {
+            if(_pullIterator == null || !_pullIterator.hasNext())
+            {
+                _pullIterator = getConsumers().iterator();
+            }
+            if(_pullIterator.hasNext())
+            {
+                MessageInstanceConsumer consumer = _pullIterator.next();
+                consumer.pullMessage();
+            }
+        }
         ConsumerMessageInstancePair consumerMessage = _queue.poll();
         if (consumerMessage != null)
         {
@@ -234,7 +335,13 @@
             {
                 consumerMessage.release();
             }
+            return true;
         }
+        else
+        {
+            return false;
+        }
+
 
     }
 
@@ -268,12 +375,26 @@
             releaseSendLock();
         }
 
-        afterCloseInternal();
+        for (MessageInstanceConsumer consumer : _consumers)
+        {
+            consumer.close();
+        }
+        if(_suspendedConsumerLoggingTicker != null)
+        {
+            getSessionModel().removeTicker(_suspendedConsumerLoggingTicker);
+        }
+
         return closed;
 
     }
 
-    protected abstract void afterCloseInternal();
+    @Override
+    public String toLogString()
+    {
+
+        return "[(** Multi-Queue **)] ";
+    }
+
 
     protected abstract void doCloseInternal();
 }
diff --git a/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java b/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java
index 86fa893..9d76a39 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/consumer/ConsumerTarget.java
@@ -42,6 +42,8 @@
 
     boolean hasCredit();
 
+    boolean isMultiQueue();
+
     enum State
     {
         ACTIVE, SUSPENDED, CLOSED
@@ -67,7 +69,7 @@
 
     boolean hasMessagesToSend();
 
-    void sendNextMessage();
+    boolean sendNextMessage();
 
     void flushBatched();
 
@@ -89,4 +91,6 @@
 
     void releaseSendLock();
 
+    boolean isPullOnly();
+
 }
diff --git a/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstanceConsumer.java b/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstanceConsumer.java
index 5e06526..4d645d4 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstanceConsumer.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/message/MessageInstanceConsumer.java
@@ -36,4 +36,8 @@
     void externalStateChange();
 
     Object getIdentifier();
+
+    boolean hasAvailableMessages();
+
+    void pullMessage();
 }
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
index 2987761..9ce9eb2 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
@@ -291,6 +291,7 @@
     private final ConcurrentMap<String, Callable<MessageFilter>> _defaultFiltersMap = new ConcurrentHashMap<>();
     private final List<HoldMethod> _holdMethods = new CopyOnWriteArrayList<>();
     private Map<String, String> _mimeTypeToFileExtension = Collections.emptyMap();
+    private volatile boolean _hasPullOnlyConsumers;
 
     private interface HoldMethod
     {
@@ -886,6 +887,10 @@
 
         if (!isDeleted())
         {
+            if(consumer.isPullOnly())
+            {
+                _hasPullOnlyConsumers = true;
+            }
             _consumerList.add(consumer);
 
             if (isDeleted())
@@ -901,7 +906,14 @@
         childAdded(consumer);
         consumer.addChangeListener(_deletedChildListener);
 
-        deliverAsync();
+        if(consumer.isPullOnly())
+        {
+            consumer.getSessionModel().getAMQPConnection().notifyWork();
+        }
+        else
+        {
+            deliverAsync();
+        }
 
         return consumer;
     }
@@ -1016,6 +1028,18 @@
                 resetSubPointersForGroups(consumer);
             }
 
+            if(consumer.isPullOnly())
+            {
+                boolean hasOnlyPushConsumers = true;
+                ConsumerNode consumerNode = _consumerList.getHead().findNext();
+                while (consumerNode != null && hasOnlyPushConsumers)
+                {
+                    hasOnlyPushConsumers = !consumerNode.getConsumer().isPullOnly();
+                    consumerNode = consumerNode.findNext();
+                }
+                _hasPullOnlyConsumers = !hasOnlyPushConsumers;
+            }
+
             // auto-delete queues must be deleted if there are no remaining subscribers
 
             if(!consumer.isTransient()
@@ -1086,7 +1110,7 @@
                 updateSubRequeueEntry(sub, entry);
             }
         }
-
+        notifyPullOnlyConsumers();
         deliverAsync();
     }
 
@@ -1239,7 +1263,7 @@
             if (entry.isAvailable())
             {
                 checkConsumersNotAheadOfDelivery(entry);
-
+                notifyPullOnlyConsumers();
                 deliverAsync();
             }
 
@@ -1533,7 +1557,7 @@
                 updateSubRequeueEntry(sub, entry);
             }
         }
-
+        notifyPullOnlyConsumers();
         deliverAsync();
 
     }
@@ -1728,9 +1752,16 @@
             if (oldState != State.ACTIVE)
             {
                 _activeSubscriberCount.incrementAndGet();
+                if(sub.isPullOnly())
+                {
+                    sub.getSessionModel().getAMQPConnection().notifyWork();
+                }
 
             }
-            deliverAsync();
+            if(!sub.isPullOnly())
+            {
+                deliverAsync();
+            }
         }
     }
 
@@ -2204,6 +2235,23 @@
 
     }
 
+    void notifyPullOnlyConsumers()
+    {
+        if(_hasPullOnlyConsumers)
+        {
+            ConsumerNode consumerNode = _consumerList.getHead().findNext();
+            while (consumerNode != null)
+            {
+                QueueConsumer<?> consumer = consumerNode.getConsumer();
+                if (consumer.isActive() && consumer.isPullOnly() && getNextAvailableEntry(consumer) != null)
+                {
+                    consumer.getSessionModel().getAMQPConnection().notifyWork();
+                }
+                consumerNode = consumerNode.findNext();
+            }
+        }
+    }
+
     void flushConsumer(QueueConsumer<?> sub)
     {
 
@@ -2450,6 +2498,12 @@
         }
     }
 
+
+    boolean hasAvailableMessages(final QueueConsumer queueConsumer)
+    {
+        return getNextAvailableEntry(queueConsumer) != null;
+    }
+
     /**
      * Used by queue Runners to asynchronously deliver messages to consumers.
      *
@@ -2516,44 +2570,46 @@
 
                 QueueConsumer<?> sub = consumerNodeIterator.getNode().getConsumer();
 
-                sub.getSendLock();
-
-                try
+                if(!sub.isPullOnly())
                 {
-                    for(int i = 0 ; i < perSub; i++)
+                    sub.getSendLock();
+
+                    try
                     {
-                        //attempt delivery. returns true if no further delivery currently possible to this sub
-                        consumerDone = attemptDelivery(sub, true);
-                        if (consumerDone)
+                        for (int i = 0; i < perSub; i++)
                         {
-                            sub.flushBatched();
-                            boolean noMore = getNextAvailableEntry(sub) == null;
-                            if (lastLoop && noMore)
-                            {
-                                sub.queueEmpty();
-                            }
-                            break;
-                        }
-                        else
-                        {
-                            //this consumer can accept additional deliveries, so we must
-                            //keep going after this (if iteration slicing allows it)
-                            allConsumersDone = false;
-                            lastLoop = false;
-                            if(--iterations == 0)
+                            //attempt delivery. returns true if no further delivery currently possible to this sub
+                            consumerDone = attemptDelivery(sub, true);
+                            if (consumerDone)
                             {
                                 sub.flushBatched();
+                                boolean noMore = getNextAvailableEntry(sub) == null;
+                                if (lastLoop && noMore)
+                                {
+                                    sub.queueEmpty();
+                                }
                                 break;
                             }
+                            else
+                            {
+                                //this consumer can accept additional deliveries, so we must
+                                //keep going after this (if iteration slicing allows it)
+                                allConsumersDone = false;
+                                lastLoop = false;
+                                if (--iterations == 0)
+                                {
+                                    sub.flushBatched();
+                                    break;
+                                }
+                            }
                         }
 
+                        sub.flushBatched();
                     }
-
-                    sub.flushBatched();
-                }
-                finally
-                {
-                    sub.releaseSendLock();
+                    finally
+                    {
+                        sub.releaseSendLock();
+                    }
                 }
             }
 
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java b/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java
index d8db052..d8d959e 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumer.java
@@ -55,4 +55,6 @@
     void awaitCredit(QueueEntry entry);
 
     boolean hasCredit();
+
+    boolean isPullOnly();
 }
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java b/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
index 7f91f29..d13f1a3 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/QueueConsumerImpl.java
@@ -120,7 +120,8 @@
     private int _priority;
 
     QueueConsumerImpl(final AbstractQueue<?> queue,
-                      ConsumerTarget target, final String consumerName,
+                      ConsumerTarget target,
+                      final String consumerName,
                       final FilterManager filters,
                       final Class<? extends ServerMessage> messageClass,
                       EnumSet<ConsumerOption> optionSet,
@@ -155,14 +156,16 @@
         };
         _target.addStateListener(_listener);
 
-        _suspendedConsumerLoggingTicker = new SuspendedConsumerLoggingTicker(queue.getContextValue(Long.class, Consumer.SUSPEND_NOTIFICATION_PERIOD))
-        {
-            @Override
-            protected void log(final long period)
-            {
-                getEventLogger().message(getLogSubject(), SubscriptionMessages.STATE(period));
-            }
-        };
+        _suspendedConsumerLoggingTicker = target.isMultiQueue()
+                ? null
+                : new SuspendedConsumerLoggingTicker(queue.getContextValue(Long.class, Consumer.SUSPEND_NOTIFICATION_PERIOD))
+                    {
+                        @Override
+                        protected void log(final long period)
+                        {
+                            getEventLogger().message(getLogSubject(), SubscriptionMessages.STATE(period));
+                        }
+                    };
 
 
     }
@@ -220,15 +223,17 @@
                 }
             }
 
-
-            if(newState == ConsumerTarget.State.SUSPENDED)
+            if(_suspendedConsumerLoggingTicker != null)
             {
-                _suspendedConsumerLoggingTicker.setStartTime(System.currentTimeMillis());
-                getSessionModel().addTicker(_suspendedConsumerLoggingTicker);
-            }
-            else
-            {
-                getSessionModel().removeTicker(_suspendedConsumerLoggingTicker);
+                if (newState == ConsumerTarget.State.SUSPENDED)
+                {
+                    _suspendedConsumerLoggingTicker.setStartTime(System.currentTimeMillis());
+                    getSessionModel().addTicker(_suspendedConsumerLoggingTicker);
+                }
+                else
+                {
+                    getSessionModel().removeTicker(_suspendedConsumerLoggingTicker);
+                }
             }
         }
 
@@ -264,7 +269,20 @@
     @Override
     public void externalStateChange()
     {
-        _queue.deliverAsync();
+        if(isPullOnly())
+        {
+            getSessionModel().getAMQPConnection().notifyWork();
+        }
+        else
+        {
+            _queue.deliverAsync();
+        }
+    }
+
+    @Override
+    public boolean hasAvailableMessages()
+    {
+        return !_queue.isEmpty() && _queue.hasAvailableMessages(this);
     }
 
     @Override
@@ -309,6 +327,10 @@
                 _target.consumerRemoved(this);
                 _target.removeStateChangeListener(_listener);
                 _queue.unregisterConsumer(this);
+                if(_suspendedConsumerLoggingTicker != null)
+                {
+                    getSessionModel().removeTicker(_suspendedConsumerLoggingTicker);
+                }
                 deleted();
             }
             finally
@@ -330,6 +352,12 @@
         _target.flushBatched();
     }
 
+    @Override
+    public boolean isPullOnly()
+    {
+        return _target.isPullOnly();
+    }
+
     public void queueDeleted()
     {
         _target.queueDeleted();
@@ -392,6 +420,22 @@
 
     }
 
+    @Override
+    public void pullMessage()
+    {
+        AMQPConnection<?> connection = _target.getSessionModel().getAMQPConnection();
+        try
+        {
+            connection.alwaysAllowMessageAssignmentInThisThreadIfItIsIOThread(true);
+            _queue.flushConsumer(this, 1);
+        }
+        finally
+        {
+            connection.alwaysAllowMessageAssignmentInThisThreadIfItIsIOThread(false);
+        }
+
+    }
+
     public boolean resend(final QueueEntry entry)
     {
         boolean messageWasResent = getQueue().resend(entry, this);
@@ -668,7 +712,14 @@
             entry.addStateChangeListener(this);
             if(!entry.isAvailable())
             {
-                _queue.deliverAsync();
+                if(isPullOnly())
+                {
+                    getSessionModel().getAMQPConnection().notifyWork();
+                }
+                else
+                {
+                    _queue.deliverAsync();
+                }
                 remove();
             }
         }
@@ -688,7 +739,14 @@
         {
             entry.removeStateChangeListener(this);
             _entry.compareAndSet(entry, null);
-            _queue.deliverAsync();
+            if(isPullOnly())
+            {
+                getSessionModel().getAMQPConnection().notifyWork();
+            }
+            else
+            {
+                _queue.deliverAsync();
+            }
         }
 
     }
diff --git a/broker-core/src/main/java/org/apache/qpid/server/transfer/TransferQueueConsumer.java b/broker-core/src/main/java/org/apache/qpid/server/transfer/TransferQueueConsumer.java
index 9d0dc6e..0547fee 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/transfer/TransferQueueConsumer.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/transfer/TransferQueueConsumer.java
@@ -196,6 +196,18 @@
         return _identifier;
     }
 
+    @Override
+    public boolean hasAvailableMessages()
+    {
+        return _transferQueue.hasAvailableMessages(this);
+    }
+
+    @Override
+    public void pullMessage()
+    {
+
+    }
+
     public void send(final TransferQueueEntry entry)
     {
         _target.send(entry);
diff --git a/broker-core/src/main/java/org/apache/qpid/server/transfer/TransferQueueImpl.java b/broker-core/src/main/java/org/apache/qpid/server/transfer/TransferQueueImpl.java
index 6b9ad95..ce81bb7 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/transfer/TransferQueueImpl.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/transfer/TransferQueueImpl.java
@@ -189,40 +189,14 @@
         _consumers.add(consumer);
         consumer.notifyWork();
 
-
-  /*      consumer.setStateListener(this);
-        QueueContext queueContext;
-        if(filters == null || !filters.startAtTail())
-        {
-            queueContext = new QueueContext(getEntries().getHead());
-        }
-        else
-        {
-            queueContext = new QueueContext(getEntries().getTail());
-        }
-        consumer.setQueueContext(queueContext);
-
-        if (!isDeleted())
-        {
-            _consumerList.add(consumer);
-
-            if (isDeleted())
-            {
-                consumer.queueDeleted();
-            }
-        }
-        else
-        {
-            // TODO
-        }
-
-        consumer.addChangeListener(_deletedChildListener);
-
-        deliverAsync();
-*/
         return consumer;
     }
 
+    boolean hasAvailableMessages(final TransferQueueConsumer queueConsumer)
+    {
+        return getNextAvailableEntry(queueConsumer) != null;
+    }
+
 
     void setLastSeenEntry(final TransferQueueConsumer sub, final TransferQueueEntry entry)
     {
@@ -396,6 +370,11 @@
         return MessageDurability.DEFAULT;
     }
 
+    public void flushConsumer(final TransferQueueConsumer consumer, final int count)
+    {
+
+    }
+
     private static class EnqueueRequest
     {
         private final MessageReference<?> _message;
diff --git a/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java b/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
index 5b6415d..2f96140 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/transport/AMQPConnection.java
@@ -28,11 +28,12 @@
 import javax.security.auth.Subject;
 
 import org.apache.qpid.protocol.AMQConstant;
+import org.apache.qpid.server.logging.EventLoggerProvider;
 import org.apache.qpid.server.model.Connection;
 import org.apache.qpid.server.protocol.AMQSessionModel;
 import org.apache.qpid.server.util.Deletable;
 
-public interface AMQPConnection<C extends AMQPConnection<C>> extends Connection<C>, Deletable<C>
+public interface AMQPConnection<C extends AMQPConnection<C>> extends Connection<C>, Deletable<C>, EventLoggerProvider
 {
     boolean isMessageAssignmentSuspended();
 
diff --git a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
index 3227214..935ab60 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractSystemMessageSource.java
@@ -149,6 +149,17 @@
             return _target;
         }
 
+        public boolean hasAvailableMessages()
+        {
+            return !_queue.isEmpty();
+        }
+
+        @Override
+        public void pullMessage()
+        {
+
+        }
+
         public AMQSessionModel getSessionModel()
         {
             return _target.getSessionModel();
diff --git a/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java b/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
index 3485b2f..5cec366 100644
--- a/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
+++ b/broker-core/src/test/java/org/apache/qpid/server/consumer/MockConsumer.java
@@ -155,9 +155,9 @@
     }
 
     @Override
-    public void sendNextMessage()
+    public boolean sendNextMessage()
     {
-
+        return false;
     }
 
     public void flushBatched()
@@ -279,6 +279,17 @@
         _stateChangeLock.unlock();
     }
 
+    @Override
+    public boolean isPullOnly()
+    {
+        return false;
+    }
+
+    @Override
+    public boolean isMultiQueue()
+    {
+        return false;
+    }
 
     private static class MockSessionModel implements AMQSessionModel<MockSessionModel>
     {
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
index c263e9c..b93be3d 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ConsumerTarget_0_10.java
@@ -22,9 +22,7 @@
 
 import java.io.IOException;
 import java.util.Collection;
-import java.util.List;
 import java.util.Map;
-import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -89,7 +87,6 @@
 
     private int _deferredMessageCredit;
     private long _deferredSizeCredit;
-    private final List<MessageInstanceConsumer> _consumers = new CopyOnWriteArrayList<>();
 
     private final StateChangeListener<MessageInstance, EntryState> _unacknowledgedMessageListener = new StateChangeListener<MessageInstance, EntryState>()
     {
@@ -117,9 +114,10 @@
                                MessageAcquireMode acquireMode,
                                MessageFlowMode flowMode,
                                FlowCreditManager_0_10 creditManager,
-                               Map<String, Object> arguments)
+                               Map<String, Object> arguments,
+                               boolean multiQueue)
     {
-        super(State.SUSPENDED);
+        super(State.SUSPENDED, isPullOnly(arguments), multiQueue, session.getAMQPConnection());
         _session = session;
         _postIdSettingAction = new AddMessageDispositionListenerAction(session);
         _acceptMode = acceptMode;
@@ -138,6 +136,12 @@
         }
     }
 
+    private static boolean isPullOnly(Map<String, Object> arguments)
+    {
+        return arguments.containsKey(PULL_ONLY_CONSUMER)
+               && Boolean.valueOf(String.valueOf(arguments.get(PULL_ONLY_CONSUMER)));
+    }
+
     @Override
     public boolean isFlowSuspended()
     {
@@ -145,17 +149,6 @@
         // TODO check for Session suspension
     }
 
-    @Override
-    protected void afterCloseInternal()
-    {
-
-        for (MessageInstanceConsumer consumer : _consumers)
-        {
-            consumer.close();
-        }
-    }
-
-    @Override
     protected void doCloseInternal()
     {
         _creditManager.removeListener(this);
@@ -621,7 +614,7 @@
     public void flush()
     {
         flushCreditState(true);
-        for(MessageInstanceConsumer consumer : _consumers)
+        for(MessageInstanceConsumer consumer : getConsumers())
         {
             consumer.flush();
         }
@@ -652,22 +645,6 @@
         return _targetAddress;
     }
 
-    @Override
-    public void consumerAdded(final MessageInstanceConsumer consumer)
-    {
-        _consumers.add(consumer);
-    }
-
-    @Override
-    public void consumerRemoved(final MessageInstanceConsumer consumer)
-    {
-        _consumers.remove(consumer);
-        if(_consumers.isEmpty())
-        {
-            close();
-        }
-    }
-
     public long getUnacknowledgedBytes()
     {
         return _unacknowledgedBytes.longValue();
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
index 1835d58..cb1fb9b 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSession.java
@@ -1257,7 +1257,10 @@
         Collection<ConsumerTarget_0_10> consumerTargets = getSubscriptions();
         for(ConsumerTarget_0_10 consumerTarget: consumerTargets)
         {
-            consumerTarget.notifyCurrentState();
+            if(!consumerTarget.isPullOnly())
+            {
+                consumerTarget.notifyCurrentState();
+            }
         }
     }
 
diff --git a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
index 48f7833..a999d4e 100644
--- a/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
+++ b/broker-plugins/amqp-0-10-protocol/src/main/java/org/apache/qpid/server/protocol/v0_10/ServerSessionDelegate.java
@@ -217,13 +217,8 @@
 
                 final Collection<MessageSource> sources = new HashSet<>();
                 final MessageSource queue = addressSpace.getAttainedMessageSource(queueName);
-                if(queue != null)
-                {
-                    sources.add(queue);
-                }
-                else if(getContextValue(session, Boolean.class, "qpid.enableMultiQueueConsumers")
-                        && method.getArguments() != null
-                        && method.getArguments().get("x-multiqueue") instanceof Collection)
+
+                if(method.getArguments() != null && method.getArguments().get("x-multiqueue") instanceof Collection)
                 {
                     for(Object object : (Collection<Object>)method.getArguments().get("x-multiqueue"))
                     {
@@ -245,6 +240,10 @@
                     }
                     queueName = method.getArguments().get("x-multiqueue").toString();
                 }
+                else if(queue != null)
+                {
+                    sources.add(queue);
+                }
 
                 if(sources.isEmpty())
                 {
@@ -306,13 +305,14 @@
 
                     }
 
-
+                    boolean multiQueue = sources.size()>1;
                     ConsumerTarget_0_10 target = new ConsumerTarget_0_10((ServerSession)session, destination,
-                                                                                 method.getAcceptMode(),
-                                                                                 method.getAcquireMode(),
-                                                                                 MessageFlowMode.WINDOW,
-                                                                                 creditManager,
-                                                                                 method.getArguments()
+                                                                         method.getAcceptMode(),
+                                                                         method.getAcquireMode(),
+                                                                         MessageFlowMode.WINDOW,
+                                                                         creditManager,
+                                                                         method.getArguments(),
+                                                                         multiQueue
                     );
 
                     Integer priority = null;
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
index 474d677..46d5939 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
@@ -106,7 +106,6 @@
 import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.StoredMessage;
 import org.apache.qpid.server.store.TransactionLogResource;
-import org.apache.qpid.server.transport.AMQPConnection;
 import org.apache.qpid.server.txn.AsyncAutoCommitTransaction;
 import org.apache.qpid.server.txn.LocalTransaction;
 import org.apache.qpid.server.txn.LocalTransaction.ActivityTimeAccessor;
@@ -732,20 +731,20 @@
 
         ConsumerTarget_0_8 target;
         EnumSet<ConsumerOption> options = EnumSet.noneOf(ConsumerOption.class);
-
+        final boolean multiQueue = sources.size()>1;
         if(arguments != null && Boolean.TRUE.equals(arguments.get(AMQPFilterTypes.NO_CONSUME.getValue())))
         {
-            target = ConsumerTarget_0_8.createBrowserTarget(this, tag, arguments, _noAckCreditManager);
+            target = ConsumerTarget_0_8.createBrowserTarget(this, tag, arguments, _noAckCreditManager, multiQueue);
         }
         else if(acks)
         {
-            target = ConsumerTarget_0_8.createAckTarget(this, tag, arguments, _creditManager);
+            target = ConsumerTarget_0_8.createAckTarget(this, tag, arguments, _creditManager, multiQueue);
             options.add(ConsumerOption.ACQUIRES);
             options.add(ConsumerOption.SEES_REQUEUES);
         }
         else
         {
-            target = ConsumerTarget_0_8.createNoAckTarget(this, tag, arguments, _noAckCreditManager);
+            target = ConsumerTarget_0_8.createNoAckTarget(this, tag, arguments, _noAckCreditManager, multiQueue);
             options.add(ConsumerOption.ACQUIRES);
             options.add(ConsumerOption.SEES_REQUEUES);
         }
@@ -1419,7 +1418,7 @@
     }
 
     @Override
-    public AMQPConnection<?> getAMQPConnection()
+    public AMQPConnection_0_8 getAMQPConnection()
     {
         return _connection;
     }
@@ -2113,13 +2112,8 @@
 
         MessageSource queue1 = queueName == null ? getDefaultQueue() : vHost.getAttainedMessageSource(queueName);
         final Collection<MessageSource> sources = new HashSet<>();
-        if (queue1 != null)
-        {
-            sources.add(queue1);
-        }
-        else if (_connection.getContextProvider().getContextValue(Boolean.class, "qpid.enableMultiQueueConsumers")
-                 && arguments != null
-                 && arguments.get("x-multiqueue") instanceof Collection)
+
+        if (arguments != null && arguments.get("x-multiqueue") instanceof Collection)
         {
             for (Object object : (Collection<Object>) arguments.get("x-multiqueue"))
             {
@@ -2141,6 +2135,11 @@
             }
             queueName = arguments.get("x-multiqueue").toString();
         }
+        else if (queue1 != null)
+        {
+            sources.add(queue1);
+        }
+
 
         if (sources.isEmpty())
         {
@@ -3793,7 +3792,10 @@
     {
         for(ConsumerTarget_0_8 consumerTarget : getConsumerTargets())
         {
-            consumerTarget.notifyCurrentState();
+            if(!consumerTarget.isPullOnly())
+            {
+                consumerTarget.notifyCurrentState();
+            }
         }
     }
 
diff --git a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
index 0fb70f6..c549fe1 100644
--- a/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
+++ b/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/ConsumerTarget_0_8.java
@@ -20,8 +20,6 @@
  */
 package org.apache.qpid.server.protocol.v0_8;
 
-import java.util.List;
-import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -56,16 +54,16 @@
 
     private final AtomicLong _unacknowledgedCount = new AtomicLong(0);
     private final AtomicLong _unacknowledgedBytes = new AtomicLong(0);
-    private final List<MessageInstanceConsumer> _consumers = new CopyOnWriteArrayList<>();
     private final AtomicBoolean _needToClose = new AtomicBoolean();
     private final String _targetAddress;
 
 
     public static ConsumerTarget_0_8 createBrowserTarget(AMQChannel channel,
                                                          AMQShortString consumerTag, FieldTable filters,
-                                                         FlowCreditManager creditManager)
+                                                         FlowCreditManager creditManager, final boolean multiQueue)
     {
-        return new BrowserConsumer(channel, consumerTag, filters, creditManager, channel.getClientDeliveryMethod(), channel.getRecordDeliveryMethod());
+        return new BrowserConsumer(channel, consumerTag, filters, creditManager, channel.getClientDeliveryMethod(), channel.getRecordDeliveryMethod(),
+                                   multiQueue);
     }
 
     public static ConsumerTarget_0_8 createGetNoAckTarget(final AMQChannel channel,
@@ -78,22 +76,18 @@
         return new GetNoAckConsumer(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod);
     }
 
-    public List<MessageInstanceConsumer> getConsumers()
-    {
-        return _consumers;
-    }
-
-
     static final class BrowserConsumer extends ConsumerTarget_0_8
     {
         public BrowserConsumer(AMQChannel channel,
-                               AMQShortString consumerTag, FieldTable filters,
+                               AMQShortString consumerTag,
+                               FieldTable filters,
                                FlowCreditManager creditManager,
                                ClientDeliveryMethod deliveryMethod,
-                               RecordDeliveryMethod recordMethod)
+                               RecordDeliveryMethod recordMethod,
+                               boolean multiQueue)
         {
             super(channel, consumerTag,
-                  filters, creditManager, deliveryMethod, recordMethod);
+                  filters, creditManager, deliveryMethod, recordMethod, multiQueue);
         }
 
         /**
@@ -124,19 +118,12 @@
     }
 
     public static ConsumerTarget_0_8 createNoAckTarget(AMQChannel channel,
-                                                           AMQShortString consumerTag, FieldTable filters,
-                                                           FlowCreditManager creditManager)
-    {
-        return new NoAckConsumer(channel, consumerTag, filters, creditManager, channel.getClientDeliveryMethod(), channel.getRecordDeliveryMethod());
-    }
-
-    public static ConsumerTarget_0_8 createNoAckTarget(AMQChannel channel,
                                                        AMQShortString consumerTag, FieldTable filters,
                                                        FlowCreditManager creditManager,
-                                                       ClientDeliveryMethod deliveryMethod,
-                                                       RecordDeliveryMethod recordMethod) throws QpidException
+                                                       boolean multiQueue)
     {
-        return new NoAckConsumer(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod);
+        return new NoAckConsumer(channel, consumerTag, filters, creditManager, channel.getClientDeliveryMethod(), channel.getRecordDeliveryMethod(),
+                                 multiQueue);
     }
 
     public static class NoAckConsumer extends ConsumerTarget_0_8
@@ -144,12 +131,14 @@
         private final AutoCommitTransaction _txn;
 
         public NoAckConsumer(AMQChannel channel,
-                             AMQShortString consumerTag, FieldTable filters,
+                             AMQShortString consumerTag,
+                             FieldTable filters,
                              FlowCreditManager creditManager,
                              ClientDeliveryMethod deliveryMethod,
-                             RecordDeliveryMethod recordMethod)
+                             RecordDeliveryMethod recordMethod,
+                             boolean multiQueue)
         {
-            super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod);
+            super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod, multiQueue);
 
             _txn = new AutoCommitTransaction(channel.getAddressSpace().getMessageStore());
         }
@@ -218,17 +207,24 @@
                                 ClientDeliveryMethod deliveryMethod,
                                 RecordDeliveryMethod recordMethod)
         {
-            super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod);
+            super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod, false);
         }
 
     }
 
 
     public static ConsumerTarget_0_8 createAckTarget(AMQChannel channel,
-                                                         AMQShortString consumerTag, FieldTable filters,
-                                                         FlowCreditManager creditManager)
+                                                     AMQShortString consumerTag,
+                                                     FieldTable filters,
+                                                     FlowCreditManager creditManager,
+                                                     boolean multiQueue)
     {
-        return new AckConsumer(channel,consumerTag,filters,creditManager, channel.getClientDeliveryMethod(), channel.getRecordDeliveryMethod());
+        return new AckConsumer(channel,
+                               consumerTag,
+                               filters, creditManager,
+                               channel.getClientDeliveryMethod(),
+                               channel.getRecordDeliveryMethod(),
+                               multiQueue);
     }
 
 
@@ -238,7 +234,7 @@
                                                          ClientDeliveryMethod deliveryMethod,
                                                          RecordDeliveryMethod recordMethod)
     {
-        return new AckConsumer(channel,consumerTag,filters,creditManager, deliveryMethod, recordMethod);
+        return new AckConsumer(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod, false);
     }
 
     static final class AckConsumer extends ConsumerTarget_0_8
@@ -247,9 +243,10 @@
                            AMQShortString consumerTag, FieldTable filters,
                            FlowCreditManager creditManager,
                            ClientDeliveryMethod deliveryMethod,
-                           RecordDeliveryMethod recordMethod)
+                           RecordDeliveryMethod recordMethod,
+                           boolean multiQueue)
         {
-            super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod);
+            super(channel, consumerTag, filters, creditManager, deliveryMethod, recordMethod, multiQueue);
         }
 
         /**
@@ -303,9 +300,10 @@
                               FieldTable arguments,
                               FlowCreditManager creditManager,
                               ClientDeliveryMethod deliveryMethod,
-                              RecordDeliveryMethod recordMethod)
+                              RecordDeliveryMethod recordMethod,
+                              boolean multiQueue)
     {
-        super(State.ACTIVE);
+        super(State.ACTIVE, isPullOnly(arguments), multiQueue, channel.getAMQPConnection());
 
         _channel = channel;
         _consumerTag = consumerTag;
@@ -344,20 +342,10 @@
         }
     }
 
-    @Override
-    public void consumerRemoved(final MessageInstanceConsumer consumer)
+    private static boolean isPullOnly(FieldTable arguments)
     {
-        _consumers.remove(consumer);
-        if(_consumers.isEmpty())
-        {
-            close();
-        }
-    }
-
-    @Override
-    public void consumerAdded(final MessageInstanceConsumer consumer)
-    {
-        _consumers.add(consumer);
+        return arguments.containsKey(PULL_ONLY_CONSUMER)
+               && Boolean.valueOf(String.valueOf(arguments.get(PULL_ONLY_CONSUMER)));
     }
 
     @Override
@@ -405,12 +393,6 @@
     }
 
     @Override
-    protected void afterCloseInternal()
-    {
-
-    }
-
-    @Override
     protected void doCloseInternal()
     {
         _creditManager.removeListener(this);
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
index 40ed270..2e9a24c 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/ConsumerTarget_1_0.java
@@ -21,6 +21,7 @@
 package org.apache.qpid.server.protocol.v1_0;
 
 import java.nio.ByteBuffer;
+import java.util.Arrays;
 import java.util.Collection;
 
 import org.slf4j.Logger;
@@ -34,6 +35,7 @@
 import org.apache.qpid.server.protocol.v1_0.type.Binary;
 import org.apache.qpid.server.protocol.v1_0.type.DeliveryState;
 import org.apache.qpid.server.protocol.v1_0.type.Outcome;
+import org.apache.qpid.server.protocol.v1_0.type.Symbol;
 import org.apache.qpid.server.protocol.v1_0.type.Target;
 import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger;
 import org.apache.qpid.server.protocol.v1_0.type.codec.AMQPDescribedTypeRegistry;
@@ -41,6 +43,7 @@
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Header;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Modified;
 import org.apache.qpid.server.protocol.v1_0.type.messaging.Released;
+import org.apache.qpid.server.protocol.v1_0.type.messaging.Source;
 import org.apache.qpid.server.protocol.v1_0.type.transaction.TransactionalState;
 import org.apache.qpid.server.protocol.v1_0.type.transport.SenderSettleMode;
 import org.apache.qpid.server.protocol.v1_0.type.transport.Transfer;
@@ -67,22 +70,22 @@
     private Binary _transactionId;
     private final AMQPDescribedTypeRegistry _typeRegistry;
     private final SectionEncoder _sectionEncoder;
-    private MessageInstanceConsumer _consumer;
     private boolean _queueEmpty;
 
     public ConsumerTarget_1_0(final SendingLink_1_0 link,
                               boolean acquires)
     {
-        super(State.SUSPENDED);
+        super(State.SUSPENDED, isPullOnly(link), false, link.getSession().getAMQPConnection());
         _link = link;
         _typeRegistry = link.getEndpoint().getSession().getConnection().getDescribedTypeRegistry();
         _sectionEncoder = new SectionEncoderImpl(_typeRegistry);
         _acquires = acquires;
     }
 
-    public MessageInstanceConsumer getConsumer()
+    private static boolean isPullOnly(SendingLink_1_0 link)
     {
-        return _consumer;
+        Source source = (Source) link.getEndpoint().getSource();
+        return Arrays.asList(source.getCapabilities()).contains(Symbol.getSymbol("QPID:PULL-ONLY"));
     }
 
     private SendingLinkEndpoint getEndpoint()
@@ -98,12 +101,6 @@
     }
 
     @Override
-    protected void afterCloseInternal()
-    {
-
-    }
-
-    @Override
     protected void doCloseInternal()
     {
 
@@ -212,7 +209,7 @@
                 else
                 {
                     UnsettledAction action = _acquires
-                            ? new DispositionAction(tag, entry)
+                            ? new DispositionAction(tag, entry, consumer)
                             : new DoNothingAction(tag, entry);
 
                     _link.addUnsettled(tag, action, entry);
@@ -239,7 +236,7 @@
 
                             public void onRollback()
                             {
-                                entry.release(getConsumer());
+                                entry.release(consumer);
                                 _link.getEndpoint().updateDisposition(tag, (DeliveryState) null, true);
                             }
                         });
@@ -251,7 +248,7 @@
             }
             else
             {
-                entry.release(getConsumer());
+                entry.release(consumer);
             }
 
         }
@@ -332,7 +329,10 @@
 
     public void flush()
     {
-        _consumer.flush();
+        for(MessageInstanceConsumer consumer : getConsumers())
+        {
+            consumer.flush();
+        }
     }
 
     private class DispositionAction implements UnsettledAction
@@ -340,11 +340,18 @@
 
         private final MessageInstance _queueEntry;
         private final Binary _deliveryTag;
+        private final MessageInstanceConsumer _consumer;
 
-        public DispositionAction(Binary tag, MessageInstance queueEntry)
+        public DispositionAction(Binary tag, MessageInstance queueEntry, final MessageInstanceConsumer consumer)
         {
             _deliveryTag = tag;
             _queueEntry = queueEntry;
+            _consumer = consumer;
+        }
+
+        public MessageInstanceConsumer getConsumer()
+        {
+            return _consumer;
         }
 
         public boolean process(DeliveryState state, final Boolean settled)
@@ -515,19 +522,6 @@
         return target instanceof org.apache.qpid.server.protocol.v1_0.type.messaging.Target ? ((org.apache.qpid.server.protocol.v1_0.type.messaging.Target) target).getAddress() : _link.getEndpoint().getName();
     }
 
-    @Override
-    public void consumerAdded(final MessageInstanceConsumer consumer)
-    {
-        _consumer = consumer;
-    }
-
-    @Override
-    public void consumerRemoved(final MessageInstanceConsumer consumer)
-    {
-        close();
-    }
-
-    @Override
     public long getUnacknowledgedBytes()
     {
         // TODO
diff --git a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
index cec03e6..40f22ec 100644
--- a/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
+++ b/broker-plugins/amqp-1-0-protocol/src/main/java/org/apache/qpid/server/protocol/v1_0/Session_1_0.java
@@ -1582,7 +1582,11 @@
     {
         for(SendingLink_1_0 link : _sendingLinks)
         {
-            link.getConsumerTarget().notifyCurrentState();
+            ConsumerTarget_1_0 consumerTarget = link.getConsumerTarget();
+            if(!consumerTarget.isPullOnly())
+            {
+                consumerTarget.notifyCurrentState();
+            }
         }
     }
 
diff --git a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
index c8b6b50..e46273d 100644
--- a/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
+++ b/broker-plugins/management-amqp/src/main/java/org/apache/qpid/server/management/amqp/ManagementNodeConsumer.java
@@ -70,6 +70,18 @@
         return _identifier;
     }
 
+    public boolean hasAvailableMessages()
+    {
+        return !_queue.isEmpty();
+    }
+
+    @Override
+    public void pullMessage()
+    {
+
+    }
+
+
     public AMQSessionModel getSessionModel()
     {
         return _target.getSessionModel();