cpp/src/qpid/client/amqp0_10/AcceptTracker.cpp

/*
 *
 * 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 "AcceptTracker.h"

namespace qpid {
namespace client {
namespace amqp0_10 {

void AcceptTracker::State::accept()
{
    unconfirmed.add(unaccepted);
    unaccepted.clear();
}

SequenceSet AcceptTracker::State::accept(qpid::framing::SequenceNumber id, bool cumulative)
{
    SequenceSet accepting;
    if (cumulative) {
        for (SequenceSet::iterator i = unaccepted.begin(); i != unaccepted.end() && *i <= id; ++i) {
            accepting.add(*i);
        }
        unconfirmed.add(accepting);
        unaccepted.remove(accepting);
    } else {
        if (unaccepted.contains(id)) {
            unaccepted.remove(id);
            unconfirmed.add(id);
            accepting.add(id);
        }
    }
    return accepting;
}

void AcceptTracker::State::release()
{
    unaccepted.clear();
}

uint32_t AcceptTracker::State::acceptsPending()
{
    return unconfirmed.size();
}

void AcceptTracker::State::completed(qpid::framing::SequenceSet& set)
{
    unconfirmed.remove(set);
}

void AcceptTracker::delivered(const std::string& destination, const qpid::framing::SequenceNumber& id)
{
    aggregateState.unaccepted.add(id);
    destinationState[destination].unaccepted.add(id);
}

namespace
{
const size_t FLUSH_FREQUENCY = 1024;
}

void AcceptTracker::addToPending(qpid::client::AsyncSession& session, const Record& record)
{
    pending.push_back(record);
    if (pending.size() % FLUSH_FREQUENCY == 0) session.flush();
}


void AcceptTracker::accept(qpid::client::AsyncSession& session)
{
    for (StateMap::iterator i = destinationState.begin(); i != destinationState.end(); ++i) {
        i->second.accept();
    }
    Record record;
    record.status = session.messageAccept(aggregateState.unaccepted);
    record.accepted = aggregateState.unaccepted;
    addToPending(session, record);
    aggregateState.accept();
}

void AcceptTracker::accept(qpid::framing::SequenceNumber id, qpid::client::AsyncSession& session, bool cumulative)
{
    for (StateMap::iterator i = destinationState.begin(); i != destinationState.end(); ++i) {
        i->second.accept(id, cumulative);
    }
    Record record;
    record.accepted = aggregateState.accept(id, cumulative);
    record.status = session.messageAccept(record.accepted);
    addToPending(session, record);
}

void AcceptTracker::release(qpid::client::AsyncSession& session)
{
    session.messageRelease(aggregateState.unaccepted);
    for (StateMap::iterator i = destinationState.begin(); i != destinationState.end(); ++i) {
        i->second.release();
    }
    aggregateState.release();
}

uint32_t AcceptTracker::acceptsPending()
{
    checkPending();
    return aggregateState.acceptsPending();
}

uint32_t AcceptTracker::acceptsPending(const std::string& destination)
{
    checkPending();
    return destinationState[destination].acceptsPending();
}

void AcceptTracker::reset()
{
    destinationState.clear();
    aggregateState.unaccepted.clear();
    aggregateState.unconfirmed.clear();
    pending.clear();
}

void AcceptTracker::checkPending()
{
    while (!pending.empty() && pending.front().status.isComplete()) {        
        completed(pending.front().accepted);
        pending.pop_front();
    }
}

void AcceptTracker::completed(qpid::framing::SequenceSet& set)
{
    for (StateMap::iterator i = destinationState.begin(); i != destinationState.end(); ++i) {
        i->second.completed(set);
    }
    aggregateState.completed(set);    
}

}}} // namespace qpid::client::amqp0_10