Introduce broker::Cluster interface.

See cpp/src/qpid/cluster/new-cluster-design.txt and new-cluster-plan.txt.

qpid/cpp/src/tests/BrokerClusterCalls.cpp is a unit test that verifies
the broker makes the expected calls on broker::Cluster in various situations.

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@1023966 13f79535-47bb-0310-9956-ffa450edef68

10 files changed, 303 insertions, 51 deletions

diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp
index 33364e48df..a288da00c7 100644
--- a/cpp/src/qpid/broker/Broker.cpp
+++ b/cpp/src/qpid/broker/Broker.cpp
@@ -24,6 +24,7 @@
 #include "qpid/broker/FanOutExchange.h"
 #include "qpid/broker/HeadersExchange.h"
 #include "qpid/broker/MessageStoreModule.h"
+#include "qpid/broker/NullCluster.h"
 #include "qpid/broker/NullMessageStore.h"
 #include "qpid/broker/RecoveryManagerImpl.h"
 #include "qpid/broker/SaslAuthenticator.h"
@@ -146,6 +147,7 @@ Broker::Broker(const Broker::Options& conf) :
                                                           conf.qmf2Support)
                                     : 0),
     store(new NullMessageStore),
+    cluster(new NullCluster),
     acl(0),
     dataDir(conf.noDataDir ? std::string() : conf.dataDir),
     queues(this),
diff --git a/cpp/src/qpid/broker/Broker.h b/cpp/src/qpid/broker/Broker.h
index 6636b5d912..dcb20c4fe3 100644
--- a/cpp/src/qpid/broker/Broker.h
+++ b/cpp/src/qpid/broker/Broker.h
@@ -70,6 +70,7 @@ namespace broker {
 
 class ExpiryPolicy;
 class Message;
+class Cluster;
 
 static const  uint16_t DEFAULT_PORT=5672;
 
@@ -153,6 +154,7 @@ public:
     std::auto_ptr<management::ManagementAgent> managementAgent;
     ProtocolFactoryMap protocolFactories;
     std::auto_ptr<MessageStore> store;
+    std::auto_ptr<Cluster> cluster;
     AclModule* acl;
     DataDir dataDir;
 
@@ -273,6 +275,9 @@ public:
     void setClusterUpdatee(bool set) { clusterUpdatee = set; }
     bool isClusterUpdatee() const { return clusterUpdatee; }
 
+    QPID_BROKER_EXTERN void setCluster(std::auto_ptr<Cluster> c) { cluster = c; }
+    QPID_BROKER_EXTERN Cluster& getCluster() { return *cluster; }
+
     management::ManagementAgent* getManagementAgent() { return managementAgent.get(); }
     
     ConnectionCounter& getConnectionCounter() {return connectionCounter;}
diff --git a/cpp/src/qpid/broker/Cluster.h b/cpp/src/qpid/broker/Cluster.h
new file mode 100644
index 0000000000..91b52e8af1
--- /dev/null
+++ b/cpp/src/qpid/broker/Cluster.h
@@ -0,0 +1,103 @@
+#ifndef QPID_BROKER_CLUSTER_H
+#define QPID_BROKER_CLUSTER_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <boost/intrusive_ptr.hpp>
+
+namespace qpid {
+
+namespace framing {
+class FieldTable;
+}
+
+namespace broker {
+
+class Message;
+struct QueuedMessage;
+class Queue;
+class Exchange;
+
+/**
+ * NOTE: this is part of an experimental cluster implementation that is not
+ * yet fully functional. The original cluster implementation remains in place.
+ * See ../cluster/new-cluster-design.txt
+ *
+ * Interface for cluster implementations. Functions on this interface are
+ * called at relevant points in the Broker's processing.
+ */
+class Cluster
+{
+  public:
+    virtual ~Cluster() {}
+
+    // Messages
+
+    /** In Exchange::route, before the message is enqueued. */
+    virtual void routing(const boost::intrusive_ptr<Message>&) = 0;
+    /** A message is delivered to a queue. */
+    virtual void enqueue(QueuedMessage&) = 0;
+    /** In Exchange::route, after all enqueues for the message. */
+    virtual void routed(const boost::intrusive_ptr<Message>&) = 0;
+
+    /** A message is acquired by a local consumer, it is unavailable to replicas. */
+    virtual void acquire(const QueuedMessage&) = 0;
+    /** A locally-acquired message is accepted, it is removed from all replicas. */
+    virtual void accept(const QueuedMessage&) = 0;
+
+    /** A locally-acquired message is rejected, and may be re-routed. */
+    virtual void reject(const QueuedMessage&) = 0;
+    /** Re-routing (if any) is complete for a rejected message. */
+    virtual void rejected(const QueuedMessage&) = 0;
+
+    /** A locally-acquired message is released by the consumer and re-queued. */
+    virtual void release(const QueuedMessage&) = 0;
+    /** A message is dropped from the queue, e.g. expired or replaced on an LVQ.
+     * This function does only local book-keeping, it does not multicast.
+     * It is reasonable to call with a queue lock held.
+     */
+    virtual void dequeue(const QueuedMessage&) = 0;
+
+    // Consumers
+
+    /** A new consumer subscribes to a queue. */
+    virtual void consume(const Queue&, size_t consumerCount) = 0;
+    /** A consumer cancels its subscription to a queue */
+    virtual void cancel(const Queue&, size_t consumerCount) = 0;
+
+    // Wiring
+
+    /** A queue is created */
+    virtual void create(const Queue&) = 0;
+    /** A queue is destroyed */
+    virtual void destroy(const Queue&) = 0;
+    /** An exchange is created */
+    virtual void create(const Exchange&) = 0;
+    /** An exchange is destroyed */
+    virtual void destroy(const Exchange&) = 0;
+    /** A binding is created */
+    virtual void bind(const Queue&, const Exchange&, const std::string& key, const framing::FieldTable& args) = 0;
+};
+
+}} // namespace qpid::broker
+
+#endif  /*!QPID_BROKER_CLUSTER_H*/
diff --git a/cpp/src/qpid/broker/DeliveryRecord.cpp b/cpp/src/qpid/broker/DeliveryRecord.cpp
index 9443eb6ea5..315b1af2a8 100644
--- a/cpp/src/qpid/broker/DeliveryRecord.cpp
+++ b/cpp/src/qpid/broker/DeliveryRecord.cpp
@@ -112,7 +112,7 @@ void DeliveryRecord::complete()  {
 
 bool DeliveryRecord::accept(TransactionContext* ctxt) {
     if (acquired && !ended) {
-        queue->dequeue(ctxt, msg);
+        queue->accept(ctxt, msg);
         setEnded();
         QPID_LOG(debug, "Accepted " << id);
     }
@@ -130,19 +130,8 @@ void DeliveryRecord::committed() const{
 }
 
 void DeliveryRecord::reject() 
-{    
-    Exchange::shared_ptr alternate = queue->getAlternateExchange();
-    if (alternate) {
-        DeliverableMessage delivery(msg.payload);
-        alternate->route(delivery, msg.payload->getRoutingKey(), msg.payload->getApplicationHeaders());
-        QPID_LOG(info, "Routed rejected message from " << queue->getName() << " to " 
-                 << alternate->getName());
-    } else {
-        //just drop it
-        QPID_LOG(info, "Dropping rejected message from " << queue->getName());
-    }
-
-    dequeue();
+{
+    queue->reject(msg);
 }
 
 uint32_t DeliveryRecord::getCredit() const
@@ -156,7 +145,7 @@ void DeliveryRecord::acquire(DeliveryIds& results) {
         results.push_back(id);
         if (!acceptExpected) {
             if (ended) { QPID_LOG(error, "Can't dequeue ended message"); }
-            else { queue->dequeue(0, msg); setEnded(); }
+            else { queue->accept(0, msg); setEnded(); }
         }
     } else {
         QPID_LOG(info, "Message already acquired " << id.getValue());
diff --git a/cpp/src/qpid/broker/Exchange.cpp b/cpp/src/qpid/broker/Exchange.cpp
index 98980e0360..aaf0805543 100644
--- a/cpp/src/qpid/broker/Exchange.cpp
+++ b/cpp/src/qpid/broker/Exchange.cpp
@@ -22,6 +22,7 @@
 #include "qpid/broker/Exchange.h"
 #include "qpid/broker/ExchangeRegistry.h"
 #include "qpid/broker/Broker.h"
+#include "qpid/broker/Cluster.h"
 #include "qpid/management/ManagementAgent.h"
 #include "qpid/broker/Queue.h"
 #include "qpid/log/Statement.h"
@@ -78,10 +79,23 @@ Exchange::PreRoute::~PreRoute(){
     }
 }
 
+// Bracket a scope with calls to Cluster::routing and Cluster::routed
+struct ScopedClusterRouting {
+    Broker* broker;
+    boost::intrusive_ptr<Message> message;
+    ScopedClusterRouting(Broker* b, boost::intrusive_ptr<Message> m)
+        : broker(b), message(m) {
+        if (broker) broker->getCluster().routing(message);
+    }
+    ~ScopedClusterRouting() {
+        if (broker) broker->getCluster().routed(message);
+    }
+};
+
 void Exchange::doRoute(Deliverable& msg, ConstBindingList b)
 {
+    ScopedClusterRouting scr(broker, &msg.getMessage());
     int count = 0;
-
     if (b.get()) {
         // Block the content release if the message is transient AND there is more than one binding
         if (!msg.getMessage().isPersistent() && b->size() > 1) {
diff --git a/cpp/src/qpid/broker/NullCluster.h b/cpp/src/qpid/broker/NullCluster.h
new file mode 100644
index 0000000000..4f3485eb40
--- /dev/null
+++ b/cpp/src/qpid/broker/NullCluster.h
@@ -0,0 +1,66 @@
+#ifndef QPID_BROKER_NULLCLUSTER_H
+#define QPID_BROKER_NULLCLUSTER_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include <qpid/broker/Cluster.h>
+
+namespace qpid {
+namespace broker {
+
+/**
+ * No-op implementation of Cluster interface, installed by broker when
+ * no cluster plug-in is present or clustering is disabled.
+ */
+class NullCluster : public Cluster
+{
+  public:
+
+    // Messages
+
+    virtual void routing(const boost::intrusive_ptr<Message>&) {}
+    virtual void enqueue(QueuedMessage&) {}
+    virtual void routed(const boost::intrusive_ptr<Message>&) {}
+    virtual void acquire(const QueuedMessage&) {}
+    virtual void accept(const QueuedMessage&) {}
+    virtual void reject(const QueuedMessage&) {}
+    virtual void rejected(const QueuedMessage&) {}
+    virtual void release(const QueuedMessage&) {}
+    virtual void dequeue(const QueuedMessage&) {}
+
+    // Consumers
+
+    virtual void consume(const Queue&, size_t) {}
+    virtual void cancel(const Queue&, size_t) {}
+
+    // Wiring
+
+    virtual void create(const Queue&) {}
+    virtual void destroy(const Queue&) {}
+    virtual void create(const Exchange&) {}
+    virtual void destroy(const Exchange&) {}
+    virtual void bind(const Queue&, const Exchange&, const std::string&, const framing::FieldTable&) {}
+};
+
+}} // namespace qpid::broker
+
+#endif
diff --git a/cpp/src/qpid/broker/Queue.cpp b/cpp/src/qpid/broker/Queue.cpp
index e59857462c..b05172f984 100644
--- a/cpp/src/qpid/broker/Queue.cpp
+++ b/cpp/src/qpid/broker/Queue.cpp
@@ -20,6 +20,7 @@
  */
 
 #include "qpid/broker/Broker.h"
+#include "qpid/broker/Cluster.h"
 #include "qpid/broker/Queue.h"
 #include "qpid/broker/QueueEvents.h"
 #include "qpid/broker/Exchange.h"
@@ -224,6 +225,7 @@ void Queue::requeue(const QueuedMessage& msg){
             }
         }
     }
+    if (broker) broker->getCluster().release(msg);
     copy.notify();
 }
 
@@ -236,8 +238,22 @@ void Queue::clearLVQIndex(const QueuedMessage& msg){
     }
 }
 
+// Inform the cluster of an acquired message on exit from a function
+// that does the acquiring. The calling function should set qmsg
+// to the acquired message.
+struct ClusterAcquireOnExit {
+    Broker* broker;
+    QueuedMessage qmsg;
+    ClusterAcquireOnExit(Broker* b) : broker(b) {}
+    ~ClusterAcquireOnExit() {
+        if (broker && qmsg.queue) broker->getCluster().acquire(qmsg);
+    }
+};
+
 bool Queue::acquireMessageAt(const SequenceNumber& position, QueuedMessage& message) 
 {
+    ClusterAcquireOnExit willAcquire(broker);
+
     Mutex::ScopedLock locker(messageLock);
     assertClusterSafe();
     QPID_LOG(debug, "Attempting to acquire message at " << position);
@@ -248,16 +264,18 @@ bool Queue::acquireMessageAt(const SequenceNumber& position, QueuedMessage& mess
         if (lastValueQueue) {
             clearLVQIndex(*i);
         }
-        QPID_LOG(debug,
-                 "Acquired message at " << i->position << " from " << name);
+        QPID_LOG(debug, "Acquired message at " << i->position << " from " << name);
+        willAcquire.qmsg = *i;
         messages.erase(i);
         return true;
-    } 
+    }
     QPID_LOG(debug, "Could not acquire message at " << position << " from " << name << "; no message at that position");
     return false;
 }
 
 bool Queue::acquire(const QueuedMessage& msg) {
+    ClusterAcquireOnExit acquire(broker);
+
     Mutex::ScopedLock locker(messageLock);
     assertClusterSafe();
 
@@ -265,16 +283,17 @@ bool Queue::acquire(const QueuedMessage& msg) {
     Messages::iterator i = findAt(msg.position); 
     if ((i != messages.end() && i->position == msg.position) && // note that in some cases payload not be set
         (!lastValueQueue ||
-        (lastValueQueue && msg.payload.get() == checkLvqReplace(*i).payload.get()) ) // note this is safe for no payload set 0==0
-        )  {
+         (lastValueQueue && msg.payload.get() == checkLvqReplace(*i).payload.get()) ) // note this is safe for no payload set 0==0
+    )  {
 
         clearLVQIndex(msg);
         QPID_LOG(debug,
                  "Match found, acquire succeeded: " <<
                  i->position << " == " << msg.position);
+        acquire.qmsg = *i;
         messages.erase(i);
         return true;
-    } 
+    }
     
     QPID_LOG(debug, "Acquire failed for " << msg.position);
     return false;
@@ -314,6 +333,8 @@ bool Queue::getNextMessage(QueuedMessage& m, Consumer::shared_ptr c)
 Queue::ConsumeCode Queue::consumeNextMessage(QueuedMessage& m, Consumer::shared_ptr c)
 {
     while (true) {
+        ClusterAcquireOnExit willAcquire(broker); // Outside the lock
+
         Mutex::ScopedLock locker(messageLock);
         if (messages.empty()) { 
             QPID_LOG(debug, "No messages to dispatch on queue '" << name << "'");
@@ -330,6 +351,7 @@ Queue::ConsumeCode Queue::consumeNextMessage(QueuedMessage& m, Consumer::shared_
             if (c->filter(msg.payload)) {
                 if (c->accept(msg.payload)) {            
                     m = msg;
+                    willAcquire.qmsg = msg;
                     popMsg(msg);
                     return CONSUMED;
                 } else {
@@ -451,40 +473,51 @@ QueuedMessage Queue::find(SequenceNumber pos) const {
     return QueuedMessage();
 }
 
-void Queue::consume(Consumer::shared_ptr c, bool requestExclusive){
+void Queue::consume(Consumer::shared_ptr c, bool requestExclusive) {
     assertClusterSafe();
-    Mutex::ScopedLock locker(consumerLock);
-    if(exclusive) {
-        throw ResourceLockedException(
-            QPID_MSG("Queue " << getName() << " has an exclusive consumer. No more consumers allowed."));
-    } else if(requestExclusive) {
-        if(consumerCount) {
+    size_t consumers;
+    {
+        Mutex::ScopedLock locker(consumerLock);
+        if(exclusive) {
             throw ResourceLockedException(
-                QPID_MSG("Queue " << getName() << " already has consumers. Exclusive access denied."));
-        } else {
-            exclusive = c->getSession();
+                QPID_MSG("Queue " << getName() << " has an exclusive consumer. No more consumers allowed."));
+        } else if(requestExclusive) {
+            if(consumerCount) {
+                throw ResourceLockedException(
+                    QPID_MSG("Queue " << getName() << " already has consumers. Exclusive access denied."));
+            } else {
+                exclusive = c->getSession();
+            }
         }
+        consumers = ++consumerCount;
+        if (mgmtObject != 0)
+            mgmtObject->inc_consumerCount ();
     }
-    consumerCount++;
-    if (mgmtObject != 0)
-        mgmtObject->inc_consumerCount ();
+    if (broker) broker->getCluster().consume(*this, consumers);
 }
 
 void Queue::cancel(Consumer::shared_ptr c){
     removeListener(c);
-    Mutex::ScopedLock locker(consumerLock);
-    consumerCount--;
-    if(exclusive) exclusive = 0;
-    if (mgmtObject != 0)
-        mgmtObject->dec_consumerCount ();
+    size_t consumers;
+    {
+        Mutex::ScopedLock locker(consumerLock);
+        consumers = --consumerCount;
+        if(exclusive) exclusive = 0;
+        if (mgmtObject != 0)
+            mgmtObject->dec_consumerCount ();
+    }
+    if (broker) broker->getCluster().cancel(*this, consumers);
 }
 
 QueuedMessage Queue::get(){
+    ClusterAcquireOnExit acquire(broker); // Outside lock
+
     Mutex::ScopedLock locker(messageLock);
     QueuedMessage msg(this);
 
     if(!messages.empty()){
         msg = getFront();
+        acquire.qmsg = msg;
         popMsg(msg);
     }
     return msg;
@@ -609,10 +642,11 @@ void Queue::popMsg(QueuedMessage& qmsg)
 
 void Queue::push(boost::intrusive_ptr<Message>& msg, bool isRecovery){
     assertClusterSafe();
+    QueuedMessage qm;
     QueueListeners::NotificationSet copy;
     {
         Mutex::ScopedLock locker(messageLock);   
-        QueuedMessage qm(this, msg, ++sequence);
+        qm = QueuedMessage(this, msg, ++sequence);
         if (insertSeqNo) msg->getOrInsertHeaders().setInt64(seqNoKey, sequence);
          
         LVQ::iterator i;
@@ -629,12 +663,14 @@ void Queue::push(boost::intrusive_ptr<Message>& msg, bool isRecovery){
                 boost::intrusive_ptr<Message> old = i->second->getReplacementMessage(this);
                 if (!old) old = i->second;
                 i->second->setReplacementMessage(msg,this);
+                // FIXME aconway 2010-10-15: it is incorrect to use qm.position below
+                // should be using the position of the message being replaced.
                 if (isRecovery) {
                     //can't issue new requests for the store until
                     //recovery is complete
                     pendingDequeues.push_back(QueuedMessage(qm.queue, old, qm.position));
                 } else {
-                    Mutex::ScopedUnlock u(messageLock);   
+                    Mutex::ScopedUnlock u(messageLock);
                     dequeue(0, QueuedMessage(qm.queue, old, qm.position));
                 }
             }           
@@ -651,6 +687,7 @@ void Queue::push(boost::intrusive_ptr<Message>& msg, bool isRecovery){
         }
     }
     copy.notify();
+    if (broker) broker->getCluster().enqueue(qm);
 }
 
 QueuedMessage Queue::getFront()
@@ -792,12 +829,42 @@ void Queue::enqueueAborted(boost::intrusive_ptr<Message> msg)
     if (policy.get()) policy->enqueueAborted(msg);       
 }
 
+void Queue::accept(TransactionContext* ctxt, const QueuedMessage& msg) {
+    if (broker) broker->getCluster().accept(msg);
+    dequeue(ctxt, msg);
+}
+
+struct ScopedClusterReject {
+    Broker* broker;
+    const QueuedMessage& qmsg;
+    ScopedClusterReject(Broker* b, const QueuedMessage& m) : broker(b), qmsg(m) {
+        if (broker) broker->getCluster().reject(qmsg);
+    }
+    ~ScopedClusterReject() {
+        if (broker) broker->getCluster().rejected(qmsg);
+    }
+};
+
+void Queue::reject(const QueuedMessage &msg) {
+    ScopedClusterReject scr(broker, msg);
+    Exchange::shared_ptr alternate = getAlternateExchange();
+    if (alternate) {
+        DeliverableMessage delivery(msg.payload);
+        alternate->route(delivery, msg.payload->getRoutingKey(), msg.payload->getApplicationHeaders());
+        QPID_LOG(info, "Routed rejected message from " << getName() << " to " 
+                 << alternate->getName());
+    } else {
+        //just drop it
+        QPID_LOG(info, "Dropping rejected message from " << getName());
+    }
+    dequeue(0, msg);
+}
+
 // return true if store exists, 
 bool Queue::dequeue(TransactionContext* ctxt, const QueuedMessage& msg)
 {
     ScopedUse u(barrier);
     if (!u.acquired) return false;
-
     {
         Mutex::ScopedLock locker(messageLock);
         if (!isEnqueued(msg)) return false;
@@ -846,6 +913,9 @@ void Queue::popAndDequeue()
  */
 void Queue::dequeued(const QueuedMessage& msg)
 {
+    // Note: Cluster::dequeued does only local book-keeping, no multicast
+    // So OK to call here with lock held.
+    if (broker) broker->getCluster().dequeue(msg);
     if (policy.get()) policy->dequeued(msg);
     mgntDeqStats(msg.payload);
     if (eventMode == ENQUEUE_AND_DEQUEUE && eventMgr) {
@@ -861,6 +931,7 @@ void Queue::create(const FieldTable& _settings)
         store->create(*this, _settings);
     }
     configure(_settings);
+    if (broker) broker->getCluster().create(*this);
 }
 
 void Queue::configure(const FieldTable& _settings, bool recovering)
@@ -934,6 +1005,7 @@ void Queue::destroy()
         store->destroy(*this);
         store = 0;//ensure we make no more calls to the store for this queue
     }
+    if (broker) broker->getCluster().destroy(*this);
 }
 
 void Queue::notifyDeleted()
diff --git a/cpp/src/qpid/broker/Queue.h b/cpp/src/qpid/broker/Queue.h
index 96c79d1b92..572f3dc0e2 100644
--- a/cpp/src/qpid/broker/Queue.h
+++ b/cpp/src/qpid/broker/Queue.h
@@ -259,6 +259,13 @@ class Queue : public boost::enable_shared_from_this<Queue>,
 
     bool enqueue(TransactionContext* ctxt, boost::intrusive_ptr<Message>& msg, bool suppressPolicyCheck = false);
     void enqueueAborted(boost::intrusive_ptr<Message> msg);
+
+    /** Message acknowledged, dequeue it. */
+    QPID_BROKER_EXTERN void accept(TransactionContext* ctxt, const QueuedMessage &msg);
+
+    /** Message rejected, dequeue it and re-route to alternate exchange if necessary. */
+    QPID_BROKER_EXTERN void reject(const QueuedMessage &msg);
+    
     /**
      * dequeue from store (only done once messages is acknowledged)
      */
diff --git a/cpp/src/qpid/broker/QueuedMessage.h b/cpp/src/qpid/broker/QueuedMessage.h
index 35e48b11f3..8cf73bda52 100644
--- a/cpp/src/qpid/broker/QueuedMessage.h
+++ b/cpp/src/qpid/broker/QueuedMessage.h
@@ -34,10 +34,9 @@ struct QueuedMessage
     framing::SequenceNumber position;
     Queue* queue;
 
-    QueuedMessage() : queue(0) {}
+    QueuedMessage(Queue* q=0) : position(0), queue(q) {}
     QueuedMessage(Queue* q, boost::intrusive_ptr<Message> msg, framing::SequenceNumber sn) : 
         payload(msg), position(sn), queue(q) {}
-    QueuedMessage(Queue* q) : queue(q) {}
     
 };
     inline bool operator<(const QueuedMessage& a, const QueuedMessage& b) { return a.position < b.position; } 
diff --git a/cpp/src/qpid/broker/SemanticState.cpp b/cpp/src/qpid/broker/SemanticState.cpp
index c91cfba2f8..f393879c16 100644
--- a/cpp/src/qpid/broker/SemanticState.cpp
+++ b/cpp/src/qpid/broker/SemanticState.cpp
@@ -333,7 +333,7 @@ bool SemanticState::ConsumerImpl::deliver(QueuedMessage& msg)
         parent->record(record);
     } 
     if (acquire && !ackExpected) {
-        queue->dequeue(0, msg);
+        queue->accept(0, msg);
     }
     if (mgmtObject) { mgmtObject->inc_delivered(); }
     return true;
@@ -347,11 +347,6 @@ bool SemanticState::ConsumerImpl::filter(intrusive_ptr<Message>)
 bool SemanticState::ConsumerImpl::accept(intrusive_ptr<Message> msg)
 {
     assertClusterSafe();
-    // FIXME aconway 2009-06-08: if we have byte & message credit but
-    // checkCredit fails because the message is to big, we should
-    // remain on queue's listener list for possible smaller messages
-    // in future.
-    // 
     blocked = !(filter(msg) && checkCredit(msg));
     return !blocked;
 }