pendingReplies.cpp

Go to the documentation of this file.

#include <algorithm>
#include <daq/pendingReplies.hpp>
 
 
namespace daq {
 
ReplyToken::ReplyToken(uint64_t token, std::weak_ptr<PendingReplies> ptr) noexcept
    : m_token(token), m_pending_replies(std::move(ptr)) {
}
 
ReplyToken::~ReplyToken() {
    Release();
}
 
void
ReplyToken::Release() {
    if (auto pending = m_pending_replies.lock(); pending) {
        pending->Release(m_token);
    }
}
 
std::shared_ptr<PendingReplies> PendingReplies::Create() {
    return std::shared_ptr<PendingReplies>(new PendingReplies());
}
 
bool PendingReplies::HasPendingReplies() const {
    return !m_pending.empty();
}
 
std::vector<PendingReplies::PendingReply> PendingReplies::GetPendingReplies() const {
    std::vector<PendingReply> r;
    r.reserve(m_pending.size());
    std::transform(m_pending.begin(), m_pending.end(), std::back_inserter(r),
                   [](auto const& tup) -> PendingReply { return std::get<1>(tup); });
    return r;
}
 
ReplyToken PendingReplies::Acquire(std::string source_id, std::string request) {
    auto token = m_next_token_id++;
    m_pending.emplace_back(std::tuple<uint64_t, PendingReply>(
        {token, PendingReply{std::move(source_id), std::move(request)}}));
    return ReplyToken(token, shared_from_this());
}
 
std::shared_ptr<ReplyToken> PendingReplies::AcquireShared(std::string source_id,
                                                          std::string request) {
    auto token = m_next_token_id++;
    m_pending.emplace_back(std::tuple<uint64_t, PendingReply>(
        {token, PendingReply{std::move(source_id), std::move(request)}}));
    return std::shared_ptr<ReplyToken>(new ReplyToken(token, shared_from_this()));
}
 
void PendingReplies::Release(uint64_t token) {
    m_pending.erase(
        std::remove_if(m_pending.begin(), m_pending.end(),
                       [=](auto const& tup) -> bool {
                       return std::get<0>(tup) == token; }),
        m_pending.end());
}
 
}  // namespace daq