qpid/cpp/src/qpid/broker/MessageGroupManager.cpp - qpid - Git at Google

 /*
  *
  * 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/framing/FieldTable.h"
 #include "qpid/types/Variant.h"
 #include "qpid/log/Statement.h"
 #include "qpid/broker/Queue.h"
 #include "qpid/broker/MessageGroupManager.h"

 using namespace qpid::broker;

 namespace {
     const std::string GROUP_QUERY_KEY("qpid.message_group_queue");
     const std::string GROUP_HEADER_KEY("group_header_key");
     const std::string GROUP_STATE_KEY("group_state");
     const std::string GROUP_ID_KEY("group_id");
     const std::string GROUP_MSG_COUNT("msg_count");
     const std::string GROUP_TIMESTAMP("timestamp");
     const std::string GROUP_CONSUMER("consumer");
 }


 const std::string MessageGroupManager::qpidMessageGroupKey("qpid.group_header_key");
 const std::string MessageGroupManager::qpidSharedGroup("qpid.shared_msg_group");
 const std::string MessageGroupManager::qpidMessageGroupTimestamp("qpid.group_timestamp");


 /** class ConsumerState **/

 /** the consumer owns the given group */
 void MessageGroupManager::ConsumerState::addGroup(const GroupState& group)
 {
     ownedGroups += 1;
     pendingMsgs += (group.totalMsgs() - group.acquiredMsgs());
 }

 /** the consumer releases the given group */
 void MessageGroupManager::ConsumerState::removeGroup(const GroupState& group)
 {
     assert(ownedGroups != 0);
     ownedGroups -= 1;
     uint32_t del = (group.totalMsgs() - group.acquiredMsgs());
     assert(del <= pendingMsgs);
     pendingMsgs -= del;
 }

 /** notify the consumer that a new message has arrived at one if its owned groups */
 void MessageGroupManager::ConsumerState::msgAvailable(const GroupState&,
                                                       const QueuedMessage& )
 {
     assert(ownedGroups != 0);
     pendingMsgs += 1;
 }

 /** notify the consumer that an available message has been acquired */
 void MessageGroupManager::ConsumerState::msgAcquired(const GroupState&,
                                                      const QueuedMessage& )
 {
     assert(pendingMsgs != 0);
     pendingMsgs -= 1;
 }


 void MessageGroupManager::consumerAdded( const Consumer& c)
 {
     const std::string& name = c.getName();
     ConsumerState& state = consumers[name];
     state.setName(name);
     state.uncancel();  // just in case old consumer resubcribed
     QPID_LOG( trace, "group queue " << qName << ": consumer " << name << " added.");
 }

 void MessageGroupManager::consumerRemoved( const Consumer& c)
 {
     const std::string& name = c.getName();
     ConsumerMap::iterator cs = consumers.find( name );
     assert(cs != consumers.end());
     ConsumerState& state = cs->second;
     state.cancel();
     if (state.groupCount() == 0) {
         assert(state.remainingMsgs() == 0);
         consumers.erase(cs);
         QPID_LOG( trace, "group queue " << qName << ": consumer " << name << " removed.");
     }
 }


 /** GroupFifo */
 void MessageGroupManager::GroupFifo::addGroup(const GroupState& group)
 {
     assert(group.totalMsgs() != 0);
     const framing::SequenceNumber& next = group.nextMsg();
     assert(fifo.find(next) == fifo.end());
     fifo[next] = &group;
 }

 void MessageGroupManager::GroupFifo::removeGroup(const GroupState& group)
 {
     Fifo::iterator pos = fifo.find( group.nextMsg() );
     assert( pos != fifo.end() && pos->second == &group );
     fifo.erase( pos );
 }

 const MessageGroupManager::GroupState& MessageGroupManager::GroupFifo::nextGroup() const
 {
     return *(fifo.begin()->second);
 }


 /** GroupState */
 void MessageGroupManager::GroupState::setOwner( ConsumerState& consumer )
 {
     assert(owner == 0);
     owner = &consumer;
     owner->addGroup( *this );
 }

 void MessageGroupManager::GroupState::resetOwner()
 {
     assert(owner);
     owner->removeGroup( *this );
     owner = 0;
 }

 const qpid::framing::SequenceNumber& MessageGroupManager::GroupState::nextMsg() const
 {
     assert(members.size() != 0);
     return members.front();
 }


 void MessageGroupManager::GroupState::enqueueMsg(const QueuedMessage& msg)
 {
     members.push_back(msg.position);
     if (owner) {
         owner->msgAvailable(*this, msg);
     }
 }


 void MessageGroupManager::GroupState::acquireMsg(const QueuedMessage& msg)
 {
     assert(members.size());   // there are msgs present
     acquired += 1;
     if (owner) {
         owner->msgAcquired(*this, msg);
     }
 }

 void MessageGroupManager::GroupState::requeueMsg(const QueuedMessage& msg)
 {
     assert(acquired != 0);
     acquired -= 1;
     if (owner) {
         owner->msgAvailable(*this, msg);
     }
 }


 void MessageGroupManager::GroupState::dequeueMsg(const QueuedMessage& msg)
 {
     assert( members.size() != 0 );
     assert( acquired != 0 );
     acquired -= 1;

     // likely to be at or near begin() if dequeued in order
     if (members.front() == msg.position) {
         members.pop_front();
     } else {
         unsigned long diff = msg.position.getValue() - members.front().getValue();
         long maxEnd = diff < members.size() ? (diff + 1) : members.size();
         GroupState::PositionFifo::iterator i =
           std::lower_bound(members.begin(), members.begin()+maxEnd, msg.position);
         assert(i != members.end() && *i == msg.position);
         members.erase(i);
     }
 }

 void MessageGroupManager::GroupState::getPositions(framing::Array& positions) const
 {
     for (PositionFifo::const_iterator p = members.begin();
          p != members.end(); ++p)
         positions.push_back(framing::Array::ValuePtr(new framing::IntegerValue( *p )));
 }

 void MessageGroupManager::GroupState::setPositions(const framing::Array& positions)
 {
     members.clear();
     for (framing::Array::const_iterator p = positions.begin(); p != positions.end(); ++p)
         members.push_back((*p)->getIntegerValue<uint32_t, 4>());
 }


 MessageGroupManager::GroupState& MessageGroupManager::findGroup(const QueuedMessage& qm)
 {
     uint32_t thisMsg = qm.position.getValue();
     if (cachedGroup && lastMsg == thisMsg) {
         hits++;
         return *cachedGroup;
     }

     std::string group = defaultGroupId;
     const qpid::framing::FieldTable* headers = qm.payload->getApplicationHeaders();
     if (headers) {
         qpid::framing::FieldTable::ValuePtr id = headers->get( groupIdHeader );
         if (id && id->convertsTo<std::string>()) {
             std::string tmp = id->get<std::string>();
             if (!tmp.empty())   // empty group is reserved
                 group = tmp;
         }
     }

     if (cachedGroup && group == lastGroup) {
         hits++;
         lastMsg = thisMsg;
         return *cachedGroup;
     }

     misses++;

     cachedGroup = &messageGroups[group];
     if (cachedGroup->getName().empty())
         cachedGroup->setName(group);    // new group, assign name
     lastMsg = thisMsg;
     lastGroup = group;
     return *cachedGroup;
 }


 void MessageGroupManager::deleteGroup(GroupState& group)
 {
     if (cachedGroup == &group)
         cachedGroup = 0;
     std::string name = group.getName();
     messageGroups.erase(name);
 }

 void MessageGroupManager::enqueued( const QueuedMessage& qm )
 {
     // @todo KAG optimization - store reference to group state in QueuedMessage
     // issue: const-ness??
     GroupState& state = findGroup(qm);
     state.enqueueMsg(qm);
     uint32_t total = state.totalMsgs();
     QPID_LOG( trace, "group queue " << qName <<
               ": added message to group id=" << state.getName() << " total=" << total );
     if (total == 1) {
         // newly created group, no owner
         freeGroups.addGroup(state);
     }
 }


 void MessageGroupManager::acquired( const QueuedMessage& qm )
 {
     GroupState& state = findGroup(qm);
     state.acquireMsg(qm);
     QPID_LOG( trace, "group queue " << qName <<
               ": acquired message in group id=" << state.getName() << " acquired=" << state.acquiredMsgs());
 }


 void MessageGroupManager::requeued( const QueuedMessage& qm )
 {
     // @todo KAG  avoid lookup: retrieve direct reference to group state from QueuedMessage
     // issue: const-ness??
     GroupState& state = findGroup(qm);
     state.requeueMsg(qm);
     if (state.acquiredMsgs() == 0 && state.getOwner()) {
         disownGroup(state);
         freeGroups.addGroup(state);
     }
     QPID_LOG( trace, "group queue " << qName <<
               ": requeued message to group id=" << state.getName() << " acquired=" << state.acquiredMsgs());
 }


 void MessageGroupManager::dequeued( const QueuedMessage& qm )
 {
     GroupState& group = findGroup(qm);
     bool freeNeeded = false;
     if (group.isFree()) {       // dequeue is occuring via mgmt, not subscriber!
         const framing::SequenceNumber next = group.nextMsg();
         if (next == qm.position) {
             /* we are about to remove the head message of this group.  This message is
              * used to index the freeGroups fifo, so we must temporarily remove it from
              * the fifo until we are done updating the head message.
              */
             freeGroups.removeGroup(group);
             freeNeeded = true;
         }
     }
     group.dequeueMsg(qm);

     uint32_t total = group.totalMsgs();
     QPID_LOG( trace, "group queue " << qName <<
               ": dequeued message from group id=" << group.getName() << " total=" << total );

     // if no more outstanding acquired messages, free the group from the consumer
     if (group.acquiredMsgs() == 0 && group.getOwner()) {
         // group is now available again
         disownGroup(group);
         freeNeeded = true;
     }

     QPID_LOG( trace, "group queue " << qName <<
               ": dequeued message from group id=" << group.getName() << " total=" << total );

     if (total == 0) {
         QPID_LOG( trace, "group queue " << qName << ": deleting group id=" << group.getName());
         deleteGroup(group);
     } else if (freeNeeded) {
         freeGroups.addGroup(group);
     }
 }

 /** remove the owner of the group */
 void MessageGroupManager::disownGroup(GroupState& group)
 {
     ConsumerState& owner = *group.getOwner();
     QPID_LOG( trace, "group queue " << qName <<
               ": consumer name=" << owner.getName() << " released group id=" << group.getName());
     group.resetOwner();
     if (owner.cancelled() && owner.groupCount() == 0) {
         // this owner has unsubscribed, we can release it now.
         std::string name = owner.getName();
         consumers.erase(name);
         QPID_LOG( error, "group queue " << qName << ": consumer " << name << " removed.");
     }
 }

 namespace {
     unsigned long found = 0;
     unsigned long failed = 0;
     unsigned long missCount = 0;
     unsigned long earlyRet = 0;
 }

 MessageGroupManager::~MessageGroupManager()
 {
     QPID_LOG( debug, "group queue " << qName << " cache results: hits=" << hits << " misses=" << misses );
 }
 bool MessageGroupManager::nextConsumableMessage( Consumer::shared_ptr& c, QueuedMessage& next )
 {
     if (messages.empty())
         return false;

     ConsumerState& cState = consumers.find(c->getName())->second;

     next.position = c->position;
     if (freeGroups.groupCount() != 0) {
         const framing::SequenceNumber& nextFree = freeGroups.nextGroup().nextMsg();
         if (nextFree <= next.position) {     // a free message is older than current
             next.position = nextFree;
             --next.position;
         }
     } else if (cState.remainingMsgs() == 0) {   // no more msgs from owned groups
         earlyRet += 1;
         return false;
     }

     int count = 1;
     while (messages.next( next.position, next )) {
         GroupState& group = findGroup(next);
         if (group.getOwner() == &cState) {
             found += 1;
             return true;
         } else if (group.isFree()) {
             if (group.nextMsg() == next.position) {    // only take from head!
                 found += 1;
                 return true;
             }
             QPID_LOG(debug, "Skipping " << next.position << " since group " << group.getName()
                      << "'s head message still pending. pos=" << group.nextMsg());
         }
         count += 1;
     }
     failed += 1;
     missCount += 1;
     return false;
 }


 bool MessageGroupManager::allocate(const std::string& consumer, const QueuedMessage& qm)
 {
     // @todo KAG avoid lookup: retrieve direct reference to group state from QueuedMessage
     GroupState& state = findGroup(qm);

     if (state.isFree()) {
         freeGroups.removeGroup(state);
         ConsumerMap::iterator cs = consumers.find( consumer );
         assert(cs != consumers.end());
         ConsumerState& owner = cs->second;
         state.setOwner( owner );
         QPID_LOG( trace, "group queue " << qName <<
                   ": consumer name=" << consumer << " has acquired group id=" << state.getName());
         return true;
     }
     return state.getOwner()->getName() == consumer;
 }

 bool MessageGroupManager::nextBrowsableMessage( Consumer::shared_ptr& c, QueuedMessage& next )
 {
     // browse: allow access to any available msg, regardless of group ownership (?ok?)
     if (!messages.empty() && messages.next(c->position, next))
         return true;
     return false;
 }

 void MessageGroupManager::query(qpid::types::Variant::Map& status) const
 {
     /** Add a description of the current state of the message groups for this queue.
         FORMAT:
         { "qpid.message_group_queue":
             { "group_header_key" : "<KEY>",
               "group_state" :
                    [ { "group_id"  : "<name>",
                        "msg_count" : <int>,
                        "timestamp" : <absTime>,
                        "consumer"  : <consumer name> },
                      {...} // one for each known group
                    ]
             }
         }
     **/

     assert(status.find(GROUP_QUERY_KEY) == status.end());
     qpid::types::Variant::Map state;
     qpid::types::Variant::List groups;

     state[GROUP_HEADER_KEY] = groupIdHeader;
     for (GroupMap::const_iterator g = messageGroups.begin();
          g != messageGroups.end(); ++g) {
         qpid::types::Variant::Map info;
         info[GROUP_ID_KEY] = g->first;
         info[GROUP_MSG_COUNT] = g->second.totalMsgs();
         info[GROUP_TIMESTAMP] = 0;   /** @todo KAG - NEED HEAD MSG TIMESTAMP */
         if (g->second.getOwner()) {
             info[GROUP_CONSUMER] = g->second.getOwner()->getName();
         } else {
             info[GROUP_CONSUMER] = std::string("");
         }
         groups.push_back(info);
     }
     state[GROUP_STATE_KEY] = groups;
     status[GROUP_QUERY_KEY] = state;
 }


 boost::shared_ptr<MessageGroupManager> MessageGroupManager::create( const std::string& qName,
                                                                     Messages& messages,
                                                                     const qpid::framing::FieldTable& settings )
 {
     boost::shared_ptr<MessageGroupManager> empty;

     if (settings.isSet(qpidMessageGroupKey)) {

         // @todo: remove once "sticky" consumers are supported - see QPID-3347
         if (!settings.isSet(qpidSharedGroup)) {
             QPID_LOG( error, "Only shared groups are supported in this version of the broker. Use '--shared-groups' in qpid-config." );
             return empty;
         }

         std::string headerKey = settings.getAsString(qpidMessageGroupKey);
         if (headerKey.empty()) {
             QPID_LOG( error, "A Message Group header key must be configured, queue=" << qName);
             return empty;
         }
         unsigned int timestamp = settings.getAsInt(qpidMessageGroupTimestamp);

         boost::shared_ptr<MessageGroupManager> manager( new MessageGroupManager( headerKey, qName, messages, timestamp ) );

         QPID_LOG( debug, "Configured Queue '" << qName <<
                   "' for message grouping using header key '" << headerKey << "'" <<
                   " (timestamp=" << timestamp << ")");
         return manager;
     }
     return empty;
 }

 std::string MessageGroupManager::defaultGroupId;
 void MessageGroupManager::setDefaults(const std::string& groupId)   // static
 {
     defaultGroupId = groupId;
 }

 /** Cluster replication:

    state map format:

    { "group-state": [ {"name": <group-name>,
                        "owner": <consumer-name>-or-empty,
                        "acquired-ct": <acquired count>,
                        "positions": [Seqnumbers, ... ]},
                       {...}
                     ]
    }
 */

 namespace {
     const std::string GROUP_NAME("name");
     const std::string GROUP_OWNER("owner");
     const std::string GROUP_ACQUIRED_CT("acquired-ct");
     const std::string GROUP_POSITIONS("positions");
     const std::string GROUP_STATE("group-state");
     const std::string OWNER_STATE("owner-state");
     const std::string CANCELLED("cancelled");
     const std::string YES("yes");
     const std::string NO("no");
     const std::string OWNER_NAME("name");
 }


 /** Runs on UPDATER to snapshot current state */
 void MessageGroupManager::getState(qpid::framing::FieldTable& state ) const
 {
     using namespace qpid::framing;
     state.clear();
     framing::Array groupState(TYPE_CODE_MAP);
     for (GroupMap::const_iterator g = messageGroups.begin();
          g != messageGroups.end(); ++g) {

         framing::FieldTable group;
         group.setString(GROUP_NAME, g->first);
         if (g->second.getOwner()) {
             group.setString(GROUP_OWNER, g->second.getOwner()->getName());
         } else {
             group.setString(GROUP_OWNER, std::string(""));
         }
         group.setInt(GROUP_ACQUIRED_CT, g->second.acquiredMsgs());
         framing::Array positions(TYPE_CODE_UINT32);
         g->second.getPositions(positions);
         group.setArray(GROUP_POSITIONS, positions);
         groupState.push_back(framing::Array::ValuePtr(new FieldTableValue(group)));
     }
     state.setArray(GROUP_STATE, groupState);

     framing::Array ownerState(TYPE_CODE_MAP);
     for (ConsumerMap::const_iterator c = consumers.begin();
          c != consumers.end(); ++c) {
         framing::FieldTable owner;
         owner.setString(OWNER_NAME, c->first);
         owner.setString(CANCELLED, c->second.cancelled() ? YES : NO);
         ownerState.push_back(framing::Array::ValuePtr(new FieldTableValue(owner)));
     }
     state.setArray(OWNER_STATE, ownerState);

     QPID_LOG(debug, "Queue \"" << qName << "\": replicating message group state, key=" << groupIdHeader);
 }


 /** called on UPDATEE to set state from snapshot */
 void MessageGroupManager::setState(const qpid::framing::FieldTable& state)
 {
     using namespace qpid::framing;
     consumers.clear();
     messageGroups.clear();
     freeGroups.clear();
     cachedGroup = 0;

     // set up the known owners
     framing::Array ownerState(TYPE_CODE_MAP);
     bool ok = state.getArray(OWNER_STATE, ownerState);
     if (!ok) {
         QPID_LOG(error, "Unable to find message group owner state information for queue \"" <<
                  qName << "\": cluster inconsistency error!");
         return;
     }

     for (framing::Array::const_iterator c = ownerState.begin(); c != ownerState.end(); ++c) {
         framing::FieldTable ownerMap;
         ok = framing::getEncodedValue<FieldTable>(*c, ownerMap);
         if (!ok) {
             QPID_LOG(error, "Invalid message group owner information for queue \"" <<
                      qName << "\": table encoding error!");
             return;
         }
         if (!ownerMap.isSet(OWNER_NAME) || !ownerMap.isSet(CANCELLED)) {
             QPID_LOG(error, "Invalid message group owner information for queue \"" <<
                      qName << "\": fields missing error!");
             return;
         }

         const std::string name = ownerMap.getAsString(OWNER_NAME);
         ConsumerState& owner = consumers[name];
         owner.setName(name);
         if (ownerMap.getAsString(CANCELLED) == YES) {
             owner.cancel();
         }
     }

     // set up the known groups
     framing::Array groupState(TYPE_CODE_MAP);
     ok = state.getArray(GROUP_STATE, groupState);
     if (!ok) {
         QPID_LOG(error, "Unable to find message group state information for queue \"" <<
                  qName << "\": cluster inconsistency error!");
         return;
     }

     for (framing::Array::const_iterator g = groupState.begin(); g != groupState.end(); ++g) {
         framing::FieldTable groupMap;
         ok = framing::getEncodedValue<FieldTable>(*g, groupMap);
         if (!ok) {
             QPID_LOG(error, "Invalid message group state information for queue \"" <<
                      qName << "\": table encoding error!");
             return;
         }
         if (!groupMap.isSet(GROUP_NAME) || !groupMap.isSet(GROUP_OWNER) || !groupMap.isSet(GROUP_ACQUIRED_CT)) {
             QPID_LOG(error, "Invalid message group state information for queue \"" <<
                      qName << "\": fields missing error!");
             return;
         }

         // replicate the group state
         std::string name = groupMap.getAsString(GROUP_NAME);
         MessageGroupManager::GroupState& group = messageGroups[name];
         assert(group.getName().empty());
         group.setName(name);
         group.setAcquired(groupMap.getAsInt(GROUP_ACQUIRED_CT));
         framing::Array positions(TYPE_CODE_UINT32);
         ok = groupMap.getArray(GROUP_POSITIONS, positions);
         if (!ok) {
             QPID_LOG(error, "Invalid message group state information for queue \"" <<
                      qName << "\": position encoding error!");
             return;
         }
         group.setPositions(positions);

         const std::string ownerName = groupMap.getAsString(GROUP_OWNER);
         if (!ownerName.empty()) {
             ConsumerState& owner = consumers[ownerName];
             group.setOwner(owner);
         } else {
             freeGroups.addGroup(group);
         }
     }

     QPID_LOG(debug, "Queue \"" << qName << "\": message group state replicated, key =" << groupIdHeader)
 }
	/*
	*
	* 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/framing/FieldTable.h"
	#include "qpid/types/Variant.h"
	#include "qpid/log/Statement.h"
	#include "qpid/broker/Queue.h"
	#include "qpid/broker/MessageGroupManager.h"

	using namespace qpid::broker;

	namespace {
	const std::string GROUP_QUERY_KEY("qpid.message_group_queue");
	const std::string GROUP_HEADER_KEY("group_header_key");
	const std::string GROUP_STATE_KEY("group_state");
	const std::string GROUP_ID_KEY("group_id");
	const std::string GROUP_MSG_COUNT("msg_count");
	const std::string GROUP_TIMESTAMP("timestamp");
	const std::string GROUP_CONSUMER("consumer");
	}


	const std::string MessageGroupManager::qpidMessageGroupKey("qpid.group_header_key");
	const std::string MessageGroupManager::qpidSharedGroup("qpid.shared_msg_group");
	const std::string MessageGroupManager::qpidMessageGroupTimestamp("qpid.group_timestamp");


	/ class ConsumerState /

	/** the consumer owns the given group */
	void MessageGroupManager::ConsumerState::addGroup(const GroupState& group)
	{
	ownedGroups += 1;
	pendingMsgs += (group.totalMsgs() - group.acquiredMsgs());
	}

	/** the consumer releases the given group */
	void MessageGroupManager::ConsumerState::removeGroup(const GroupState& group)
	{
	assert(ownedGroups != 0);
	ownedGroups -= 1;
	uint32_t del = (group.totalMsgs() - group.acquiredMsgs());
	assert(del <= pendingMsgs);
	pendingMsgs -= del;
	}

	/** notify the consumer that a new message has arrived at one if its owned groups */
	void MessageGroupManager::ConsumerState::msgAvailable(const GroupState&,
	const QueuedMessage& )
	{
	assert(ownedGroups != 0);
	pendingMsgs += 1;
	}

	/** notify the consumer that an available message has been acquired */
	void MessageGroupManager::ConsumerState::msgAcquired(const GroupState&,
	const QueuedMessage& )
	{
	assert(pendingMsgs != 0);
	pendingMsgs -= 1;
	}


	void MessageGroupManager::consumerAdded( const Consumer& c)
	{
	const std::string& name = c.getName();
	ConsumerState& state = consumers[name];
	state.setName(name);
	state.uncancel(); // just in case old consumer resubcribed
	QPID_LOG( trace, "group queue " << qName << ": consumer " << name << " added.");
	}

	void MessageGroupManager::consumerRemoved( const Consumer& c)
	{
	const std::string& name = c.getName();
	ConsumerMap::iterator cs = consumers.find( name );
	assert(cs != consumers.end());
	ConsumerState& state = cs->second;
	state.cancel();
	if (state.groupCount() == 0) {
	assert(state.remainingMsgs() == 0);
	consumers.erase(cs);
	QPID_LOG( trace, "group queue " << qName << ": consumer " << name << " removed.");
	}
	}


	/** GroupFifo */
	void MessageGroupManager::GroupFifo::addGroup(const GroupState& group)
	{
	assert(group.totalMsgs() != 0);
	const framing::SequenceNumber& next = group.nextMsg();
	assert(fifo.find(next) == fifo.end());
	fifo[next] = &group;
	}

	void MessageGroupManager::GroupFifo::removeGroup(const GroupState& group)
	{
	Fifo::iterator pos = fifo.find( group.nextMsg() );
	assert( pos != fifo.end() && pos->second == &group );
	fifo.erase( pos );
	}

	const MessageGroupManager::GroupState& MessageGroupManager::GroupFifo::nextGroup() const
	{
	return *(fifo.begin()->second);
	}


	/** GroupState */
	void MessageGroupManager::GroupState::setOwner( ConsumerState& consumer )
	{
	assert(owner == 0);
	owner = &consumer;
	owner->addGroup( *this );
	}

	void MessageGroupManager::GroupState::resetOwner()
	{
	assert(owner);
	owner->removeGroup( *this );
	owner = 0;
	}

	const qpid::framing::SequenceNumber& MessageGroupManager::GroupState::nextMsg() const
	{
	assert(members.size() != 0);
	return members.front();
	}


	void MessageGroupManager::GroupState::enqueueMsg(const QueuedMessage& msg)
	{
	members.push_back(msg.position);
	if (owner) {
	owner->msgAvailable(*this, msg);
	}
	}


	void MessageGroupManager::GroupState::acquireMsg(const QueuedMessage& msg)
	{
	assert(members.size()); // there are msgs present
	acquired += 1;
	if (owner) {
	owner->msgAcquired(*this, msg);
	}
	}

	void MessageGroupManager::GroupState::requeueMsg(const QueuedMessage& msg)
	{
	assert(acquired != 0);
	acquired -= 1;
	if (owner) {
	owner->msgAvailable(*this, msg);
	}
	}


	void MessageGroupManager::GroupState::dequeueMsg(const QueuedMessage& msg)
	{
	assert( members.size() != 0 );
	assert( acquired != 0 );
	acquired -= 1;

	// likely to be at or near begin() if dequeued in order
	if (members.front() == msg.position) {
	members.pop_front();
	} else {
	unsigned long diff = msg.position.getValue() - members.front().getValue();
	long maxEnd = diff < members.size() ? (diff + 1) : members.size();
	GroupState::PositionFifo::iterator i =
	std::lower_bound(members.begin(), members.begin()+maxEnd, msg.position);
	assert(i != members.end() && *i == msg.position);
	members.erase(i);
	}
	}

	void MessageGroupManager::GroupState::getPositions(framing::Array& positions) const
	{
	for (PositionFifo::const_iterator p = members.begin();
	p != members.end(); ++p)
	positions.push_back(framing::Array::ValuePtr(new framing::IntegerValue( *p )));
	}

	void MessageGroupManager::GroupState::setPositions(const framing::Array& positions)
	{
	members.clear();
	for (framing::Array::const_iterator p = positions.begin(); p != positions.end(); ++p)
	members.push_back((*p)->getIntegerValue<uint32_t, 4>());
	}



	MessageGroupManager::GroupState& MessageGroupManager::findGroup(const QueuedMessage& qm)
	{
	uint32_t thisMsg = qm.position.getValue();
	if (cachedGroup && lastMsg == thisMsg) {
	hits++;
	return *cachedGroup;
	}

	std::string group = defaultGroupId;
	const qpid::framing::FieldTable* headers = qm.payload->getApplicationHeaders();
	if (headers) {
	qpid::framing::FieldTable::ValuePtr id = headers->get( groupIdHeader );
	if (id && id->convertsTo<std::string>()) {
	std::string tmp = id->get<std::string>();
	if (!tmp.empty()) // empty group is reserved
	group = tmp;
	}
	}

	if (cachedGroup && group == lastGroup) {
	hits++;
	lastMsg = thisMsg;
	return *cachedGroup;
	}

	misses++;

	cachedGroup = &messageGroups[group];
	if (cachedGroup->getName().empty())
	cachedGroup->setName(group); // new group, assign name
	lastMsg = thisMsg;
	lastGroup = group;
	return *cachedGroup;
	}


	void MessageGroupManager::deleteGroup(GroupState& group)
	{
	if (cachedGroup == &group)
	cachedGroup = 0;
	std::string name = group.getName();
	messageGroups.erase(name);
	}

	void MessageGroupManager::enqueued( const QueuedMessage& qm )
	{
	// @todo KAG optimization - store reference to group state in QueuedMessage
	// issue: const-ness??
	GroupState& state = findGroup(qm);
	state.enqueueMsg(qm);
	uint32_t total = state.totalMsgs();
	QPID_LOG( trace, "group queue " << qName <<
	": added message to group id=" << state.getName() << " total=" << total );
	if (total == 1) {
	// newly created group, no owner
	freeGroups.addGroup(state);
	}
	}


	void MessageGroupManager::acquired( const QueuedMessage& qm )
	{
	GroupState& state = findGroup(qm);
	state.acquireMsg(qm);
	QPID_LOG( trace, "group queue " << qName <<
	": acquired message in group id=" << state.getName() << " acquired=" << state.acquiredMsgs());
	}


	void MessageGroupManager::requeued( const QueuedMessage& qm )
	{
	// @todo KAG avoid lookup: retrieve direct reference to group state from QueuedMessage
	// issue: const-ness??
	GroupState& state = findGroup(qm);
	state.requeueMsg(qm);
	if (state.acquiredMsgs() == 0 && state.getOwner()) {
	disownGroup(state);
	freeGroups.addGroup(state);
	}
	QPID_LOG( trace, "group queue " << qName <<
	": requeued message to group id=" << state.getName() << " acquired=" << state.acquiredMsgs());
	}


	void MessageGroupManager::dequeued( const QueuedMessage& qm )
	{
	GroupState& group = findGroup(qm);
	bool freeNeeded = false;
	if (group.isFree()) { // dequeue is occuring via mgmt, not subscriber!
	const framing::SequenceNumber next = group.nextMsg();
	if (next == qm.position) {
	/* we are about to remove the head message of this group. This message is
	* used to index the freeGroups fifo, so we must temporarily remove it from
	* the fifo until we are done updating the head message.
	*/
	freeGroups.removeGroup(group);
	freeNeeded = true;
	}
	}
	group.dequeueMsg(qm);

	uint32_t total = group.totalMsgs();
	QPID_LOG( trace, "group queue " << qName <<
	": dequeued message from group id=" << group.getName() << " total=" << total );

	// if no more outstanding acquired messages, free the group from the consumer
	if (group.acquiredMsgs() == 0 && group.getOwner()) {
	// group is now available again
	disownGroup(group);
	freeNeeded = true;
	}

	QPID_LOG( trace, "group queue " << qName <<
	": dequeued message from group id=" << group.getName() << " total=" << total );

	if (total == 0) {
	QPID_LOG( trace, "group queue " << qName << ": deleting group id=" << group.getName());
	deleteGroup(group);
	} else if (freeNeeded) {
	freeGroups.addGroup(group);
	}
	}

	/** remove the owner of the group */
	void MessageGroupManager::disownGroup(GroupState& group)
	{
	ConsumerState& owner = *group.getOwner();
	QPID_LOG( trace, "group queue " << qName <<
	": consumer name=" << owner.getName() << " released group id=" << group.getName());
	group.resetOwner();
	if (owner.cancelled() && owner.groupCount() == 0) {
	// this owner has unsubscribed, we can release it now.
	std::string name = owner.getName();
	consumers.erase(name);
	QPID_LOG( error, "group queue " << qName << ": consumer " << name << " removed.");
	}
	}

	namespace {
	unsigned long found = 0;
	unsigned long failed = 0;
	unsigned long missCount = 0;
	unsigned long earlyRet = 0;
	}

	MessageGroupManager::~MessageGroupManager()
	{
	QPID_LOG( debug, "group queue " << qName << " cache results: hits=" << hits << " misses=" << misses );
	}
	bool MessageGroupManager::nextConsumableMessage( Consumer::shared_ptr& c, QueuedMessage& next )
	{
	if (messages.empty())
	return false;

	ConsumerState& cState = consumers.find(c->getName())->second;

	next.position = c->position;
	if (freeGroups.groupCount() != 0) {
	const framing::SequenceNumber& nextFree = freeGroups.nextGroup().nextMsg();
	if (nextFree <= next.position) { // a free message is older than current
	next.position = nextFree;
	--next.position;
	}
	} else if (cState.remainingMsgs() == 0) { // no more msgs from owned groups
	earlyRet += 1;
	return false;
	}

	int count = 1;
	while (messages.next( next.position, next )) {
	GroupState& group = findGroup(next);
	if (group.getOwner() == &cState) {
	found += 1;
	return true;
	} else if (group.isFree()) {
	if (group.nextMsg() == next.position) { // only take from head!
	found += 1;
	return true;
	}
	QPID_LOG(debug, "Skipping " << next.position << " since group " << group.getName()
	<< "'s head message still pending. pos=" << group.nextMsg());
	}
	count += 1;
	}
	failed += 1;
	missCount += 1;
	return false;
	}


	bool MessageGroupManager::allocate(const std::string& consumer, const QueuedMessage& qm)
	{
	// @todo KAG avoid lookup: retrieve direct reference to group state from QueuedMessage
	GroupState& state = findGroup(qm);

	if (state.isFree()) {
	freeGroups.removeGroup(state);
	ConsumerMap::iterator cs = consumers.find( consumer );
	assert(cs != consumers.end());
	ConsumerState& owner = cs->second;
	state.setOwner( owner );
	QPID_LOG( trace, "group queue " << qName <<
	": consumer name=" << consumer << " has acquired group id=" << state.getName());
	return true;
	}
	return state.getOwner()->getName() == consumer;
	}

	bool MessageGroupManager::nextBrowsableMessage( Consumer::shared_ptr& c, QueuedMessage& next )
	{
	// browse: allow access to any available msg, regardless of group ownership (?ok?)
	if (!messages.empty() && messages.next(c->position, next))
	return true;
	return false;
	}

	void MessageGroupManager::query(qpid::types::Variant::Map& status) const
	{
	/** Add a description of the current state of the message groups for this queue.
	FORMAT:
	{ "qpid.message_group_queue":
	{ "group_header_key" : "<KEY>",
	"group_state" :
	[ { "group_id" : "<name>",
	"msg_count" : <int>,
	"timestamp" : <absTime>,
	"consumer" : <consumer name> },
	{...} // one for each known group
	]
	}
	}
	**/

	assert(status.find(GROUP_QUERY_KEY) == status.end());
	qpid::types::Variant::Map state;
	qpid::types::Variant::List groups;

	state[GROUP_HEADER_KEY] = groupIdHeader;
	for (GroupMap::const_iterator g = messageGroups.begin();
	g != messageGroups.end(); ++g) {
	qpid::types::Variant::Map info;
	info[GROUP_ID_KEY] = g->first;
	info[GROUP_MSG_COUNT] = g->second.totalMsgs();
	info[GROUP_TIMESTAMP] = 0; /** @todo KAG - NEED HEAD MSG TIMESTAMP */
	if (g->second.getOwner()) {
	info[GROUP_CONSUMER] = g->second.getOwner()->getName();
	} else {
	info[GROUP_CONSUMER] = std::string("");
	}
	groups.push_back(info);
	}
	state[GROUP_STATE_KEY] = groups;
	status[GROUP_QUERY_KEY] = state;
	}


	boost::shared_ptr<MessageGroupManager> MessageGroupManager::create( const std::string& qName,
	Messages& messages,
	const qpid::framing::FieldTable& settings )
	{
	boost::shared_ptr<MessageGroupManager> empty;

	if (settings.isSet(qpidMessageGroupKey)) {

	// @todo: remove once "sticky" consumers are supported - see QPID-3347
	if (!settings.isSet(qpidSharedGroup)) {
	QPID_LOG( error, "Only shared groups are supported in this version of the broker. Use '--shared-groups' in qpid-config." );
	return empty;
	}

	std::string headerKey = settings.getAsString(qpidMessageGroupKey);
	if (headerKey.empty()) {
	QPID_LOG( error, "A Message Group header key must be configured, queue=" << qName);
	return empty;
	}
	unsigned int timestamp = settings.getAsInt(qpidMessageGroupTimestamp);

	boost::shared_ptr<MessageGroupManager> manager( new MessageGroupManager( headerKey, qName, messages, timestamp ) );

	QPID_LOG( debug, "Configured Queue '" << qName <<
	"' for message grouping using header key '" << headerKey << "'" <<
	" (timestamp=" << timestamp << ")");
	return manager;
	}
	return empty;
	}

	std::string MessageGroupManager::defaultGroupId;
	void MessageGroupManager::setDefaults(const std::string& groupId) // static
	{
	defaultGroupId = groupId;
	}

	/** Cluster replication:

	state map format:

	{ "group-state": [ {"name": <group-name>,
	"owner": <consumer-name>-or-empty,
	"acquired-ct": <acquired count>,
	"positions": [Seqnumbers, ... ]},
	{...}
	]
	}
	*/

	namespace {
	const std::string GROUP_NAME("name");
	const std::string GROUP_OWNER("owner");
	const std::string GROUP_ACQUIRED_CT("acquired-ct");
	const std::string GROUP_POSITIONS("positions");
	const std::string GROUP_STATE("group-state");
	const std::string OWNER_STATE("owner-state");
	const std::string CANCELLED("cancelled");
	const std::string YES("yes");
	const std::string NO("no");
	const std::string OWNER_NAME("name");
	}


	/** Runs on UPDATER to snapshot current state */
	void MessageGroupManager::getState(qpid::framing::FieldTable& state ) const
	{
	using namespace qpid::framing;
	state.clear();
	framing::Array groupState(TYPE_CODE_MAP);
	for (GroupMap::const_iterator g = messageGroups.begin();
	g != messageGroups.end(); ++g) {

	framing::FieldTable group;
	group.setString(GROUP_NAME, g->first);
	if (g->second.getOwner()) {
	group.setString(GROUP_OWNER, g->second.getOwner()->getName());
	} else {
	group.setString(GROUP_OWNER, std::string(""));
	}
	group.setInt(GROUP_ACQUIRED_CT, g->second.acquiredMsgs());
	framing::Array positions(TYPE_CODE_UINT32);
	g->second.getPositions(positions);
	group.setArray(GROUP_POSITIONS, positions);
	groupState.push_back(framing::Array::ValuePtr(new FieldTableValue(group)));
	}
	state.setArray(GROUP_STATE, groupState);

	framing::Array ownerState(TYPE_CODE_MAP);
	for (ConsumerMap::const_iterator c = consumers.begin();
	c != consumers.end(); ++c) {
	framing::FieldTable owner;
	owner.setString(OWNER_NAME, c->first);
	owner.setString(CANCELLED, c->second.cancelled() ? YES : NO);
	ownerState.push_back(framing::Array::ValuePtr(new FieldTableValue(owner)));
	}
	state.setArray(OWNER_STATE, ownerState);

	QPID_LOG(debug, "Queue \"" << qName << "\": replicating message group state, key=" << groupIdHeader);
	}


	/** called on UPDATEE to set state from snapshot */
	void MessageGroupManager::setState(const qpid::framing::FieldTable& state)
	{
	using namespace qpid::framing;
	consumers.clear();
	messageGroups.clear();
	freeGroups.clear();
	cachedGroup = 0;

	// set up the known owners
	framing::Array ownerState(TYPE_CODE_MAP);
	bool ok = state.getArray(OWNER_STATE, ownerState);
	if (!ok) {
	QPID_LOG(error, "Unable to find message group owner state information for queue \"" <<
	qName << "\": cluster inconsistency error!");
	return;
	}

	for (framing::Array::const_iterator c = ownerState.begin(); c != ownerState.end(); ++c) {
	framing::FieldTable ownerMap;
	ok = framing::getEncodedValue<FieldTable>(*c, ownerMap);
	if (!ok) {
	QPID_LOG(error, "Invalid message group owner information for queue \"" <<
	qName << "\": table encoding error!");
	return;
	}
	if (!ownerMap.isSet(OWNER_NAME) \|\| !ownerMap.isSet(CANCELLED)) {
	QPID_LOG(error, "Invalid message group owner information for queue \"" <<
	qName << "\": fields missing error!");
	return;
	}

	const std::string name = ownerMap.getAsString(OWNER_NAME);
	ConsumerState& owner = consumers[name];
	owner.setName(name);
	if (ownerMap.getAsString(CANCELLED) == YES) {
	owner.cancel();
	}
	}

	// set up the known groups
	framing::Array groupState(TYPE_CODE_MAP);
	ok = state.getArray(GROUP_STATE, groupState);
	if (!ok) {
	QPID_LOG(error, "Unable to find message group state information for queue \"" <<
	qName << "\": cluster inconsistency error!");
	return;
	}

	for (framing::Array::const_iterator g = groupState.begin(); g != groupState.end(); ++g) {
	framing::FieldTable groupMap;
	ok = framing::getEncodedValue<FieldTable>(*g, groupMap);
	if (!ok) {
	QPID_LOG(error, "Invalid message group state information for queue \"" <<
	qName << "\": table encoding error!");
	return;
	}
	if (!groupMap.isSet(GROUP_NAME) \|\| !groupMap.isSet(GROUP_OWNER) \|\| !groupMap.isSet(GROUP_ACQUIRED_CT)) {
	QPID_LOG(error, "Invalid message group state information for queue \"" <<
	qName << "\": fields missing error!");
	return;
	}

	// replicate the group state
	std::string name = groupMap.getAsString(GROUP_NAME);
	MessageGroupManager::GroupState& group = messageGroups[name];
	assert(group.getName().empty());
	group.setName(name);
	group.setAcquired(groupMap.getAsInt(GROUP_ACQUIRED_CT));
	framing::Array positions(TYPE_CODE_UINT32);
	ok = groupMap.getArray(GROUP_POSITIONS, positions);
	if (!ok) {
	QPID_LOG(error, "Invalid message group state information for queue \"" <<
	qName << "\": position encoding error!");
	return;
	}
	group.setPositions(positions);

	const std::string ownerName = groupMap.getAsString(GROUP_OWNER);
	if (!ownerName.empty()) {
	ConsumerState& owner = consumers[ownerName];
	group.setOwner(owner);
	} else {
	freeGroups.addGroup(group);
	}
	}

	QPID_LOG(debug, "Queue \"" << qName << "\": message group state replicated, key =" << groupIdHeader)
	}