dispatcher.cpp

Go to the documentation of this file.

/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/*
 * Copyright (c) 2013-2023 Regents of the University of California.
 *
 * This file is part of ndn-cxx library (NDN C++ library with eXperimental eXtensions).
 *
 * ndn-cxx library is free software: you can redistribute it and/or modify it under the
 * terms of the GNU Lesser General Public License as published by the Free Software
 * Foundation, either version 3 of the License, or (at your option) any later version.
 *
 * ndn-cxx library is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
 * PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more details.
 *
 * You should have received copies of the GNU General Public License and GNU Lesser
 * General Public License along with ndn-cxx, e.g., in COPYING.md file.  If not, see
 * <http://www.gnu.org/licenses/>.
 *
 * See AUTHORS.md for complete list of ndn-cxx authors and contributors.
 */
 
#include "ndn-cxx/mgmt/dispatcher.hpp"
#include "ndn-cxx/lp/tags.hpp"
#include "ndn-cxx/util/logger.hpp"
 
#include <algorithm>
 
namespace ndn::mgmt {
 
NDN_LOG_INIT(ndn.mgmt.Dispatcher);
 
Authorization
makeAcceptAllAuthorization()
{
  return [] (const Name& prefix,
             const Interest& interest,
             const ControlParameters* params,
             const AcceptContinuation& accept,
             const RejectContinuation& reject) {
    accept("");
  };
}
 
Dispatcher::Dispatcher(Face& face, KeyChain& keyChain,
                       const security::SigningInfo& signingInfo,
                       size_t imsCapacity)
  : m_face(face)
  , m_keyChain(keyChain)
  , m_signingInfo(signingInfo)
  , m_storage(m_face.getIoContext(), imsCapacity)
{
}
 
Dispatcher::~Dispatcher() = default;
 
void
Dispatcher::addTopPrefix(const Name& prefix, bool wantRegister,
                         const security::SigningInfo& signingInfo)
{
  bool hasOverlap = std::any_of(m_topLevelPrefixes.begin(), m_topLevelPrefixes.end(),
                                [&prefix] (const auto& x) {
                                  return x.first.isPrefixOf(prefix) || prefix.isPrefixOf(x.first);
                                });
  if (hasOverlap) {
    NDN_THROW(std::out_of_range("top-level prefix overlaps"));
  }
 
  TopPrefixEntry& topPrefixEntry = m_topLevelPrefixes[prefix];
 
  if (wantRegister) {
    topPrefixEntry.registeredPrefix = m_face.registerPrefix(prefix,
      nullptr,
      [] (const Name&, const std::string& reason) {
        NDN_THROW(std::runtime_error("prefix registration failed: " + reason));
      },
      signingInfo);
  }
 
  for (const auto& entry : m_handlers) {
    Name fullPrefix = Name(prefix).append(entry.first);
    auto filterHdl = m_face.setInterestFilter(fullPrefix,
      [=, cb = entry.second] (const auto&, const auto& interest) {
        cb(prefix, interest);
      });
    topPrefixEntry.interestFilters.emplace_back(std::move(filterHdl));
  }
}
 
void
Dispatcher::removeTopPrefix(const Name& prefix)
{
  m_topLevelPrefixes.erase(prefix);
}
 
bool
Dispatcher::isOverlappedWithOthers(const PartialName& relPrefix) const
{
  bool hasOverlapWithHandlers =
    std::any_of(m_handlers.begin(), m_handlers.end(),
                [&] (const auto& entry) {
                  return entry.first.isPrefixOf(relPrefix) || relPrefix.isPrefixOf(entry.first);
                });
  bool hasOverlapWithStreams =
    std::any_of(m_streams.begin(), m_streams.end(),
                [&] (const auto& entry) {
                  return entry.first.isPrefixOf(relPrefix) || relPrefix.isPrefixOf(entry.first);
                });
 
  return hasOverlapWithHandlers || hasOverlapWithStreams;
}
 
void
Dispatcher::afterAuthorizationRejected(RejectReply act, const Interest& interest)
{
  if (act == RejectReply::STATUS403) {
    sendControlResponse(ControlResponse(403, "authorization rejected"), interest);
  }
}
 
void
Dispatcher::queryStorage(const Name& prefix, const Interest& interest,
                         const InterestHandler& missContinuation)
{
  auto data = m_storage.find(interest);
  if (data == nullptr) {
    // invoke missContinuation to process this Interest if the query fails.
    if (missContinuation)
      missContinuation(prefix, interest);
  }
  else {
    // send the fetched data through face if query succeeds.
    sendOnFace(*data);
  }
}
 
void
Dispatcher::sendData(const Name& dataName, const Block& content, const MetaInfo& metaInfo,
                     SendDestination option)
{
  auto data = make_shared<Data>(dataName);
  data->setContent(content).setMetaInfo(metaInfo).setFreshnessPeriod(1_s);
 
  m_keyChain.sign(*data, m_signingInfo);
 
  if (option == SendDestination::IMS || option == SendDestination::FACE_AND_IMS) {
    lp::CachePolicy policy;
    policy.setPolicy(lp::CachePolicyType::NO_CACHE);
    data->setTag(make_shared<lp::CachePolicyTag>(policy));
    m_storage.insert(*data, 1_s);
  }
 
  if (option == SendDestination::FACE || option == SendDestination::FACE_AND_IMS) {
    sendOnFace(*data);
  }
}
 
void
Dispatcher::sendOnFace(const Data& data)
{
  try {
    m_face.put(data);
  }
  catch (const Face::Error& e) {
    NDN_LOG_ERROR("sendOnFace(" << data.getName() << "): " << e.what());
  }
}
 
void
Dispatcher::processControlCommandInterest(const Name& prefix,
                                          const Name& relPrefix,
                                          const Interest& interest,
                                          const ControlParametersParser& parser,
                                          const Authorization& authorization,
                                          const AuthorizationAcceptedCallback& accepted,
                                          const AuthorizationRejectedCallback& rejected)
{
  // /<prefix>/<relPrefix>/<parameters>
  size_t parametersLoc = prefix.size() + relPrefix.size();
  const name::Component& pc = interest.getName().get(parametersLoc);
 
  shared_ptr<ControlParameters> parameters;
  try {
    parameters = parser(pc);
  }
  catch (const tlv::Error&) {
    return;
  }
 
  AcceptContinuation accept = [=] (const auto& req) { accepted(req, prefix, interest, parameters); };
  RejectContinuation reject = [=] (RejectReply reply) { rejected(reply, interest); };
  authorization(prefix, interest, parameters.get(), accept, reject);
}
 
void
Dispatcher::processAuthorizedControlCommandInterest(const std::string& requester,
                                                    const Name& prefix,
                                                    const Interest& interest,
                                                    const shared_ptr<ControlParameters>& parameters,
                                                    const ValidateParameters& validateParams,
                                                    const ControlCommandHandler& handler)
{
  if (validateParams(*parameters)) {
    handler(prefix, interest, *parameters,
            [=] (const auto& resp) { sendControlResponse(resp, interest); });
  }
  else {
    sendControlResponse(ControlResponse(400, "failed in validating parameters"), interest);
  }
}
 
void
Dispatcher::sendControlResponse(const ControlResponse& resp, const Interest& interest, bool isNack)
{
  MetaInfo metaInfo;
  if (isNack) {
    metaInfo.setType(tlv::ContentType_Nack);
  }
 
  // control response is always sent out through the face
  sendData(interest.getName(), resp.wireEncode(), metaInfo, SendDestination::FACE);
}
 
void
Dispatcher::addStatusDataset(const PartialName& relPrefix,
                             Authorization auth,
                             StatusDatasetHandler handler)
{
  if (!m_topLevelPrefixes.empty()) {
    NDN_THROW(std::domain_error("one or more top-level prefix has been added"));
  }
 
  if (isOverlappedWithOthers(relPrefix)) {
    NDN_THROW(std::out_of_range("status dataset name overlaps"));
  }
 
  AuthorizationAcceptedCallback accept =
    [this, handler = std::move(handler)] (auto&&, const auto& prefix, const auto& interest, auto&&) {
      processAuthorizedStatusDatasetInterest(prefix, interest, handler);
    };
  AuthorizationRejectedCallback reject =
    [this] (auto&&... args) {
      afterAuthorizationRejected(std::forward<decltype(args)>(args)...);
    };
  // follow the general path if storage is a miss
  InterestHandler missContinuation =
    [this, auth = std::move(auth), accept = std::move(accept), reject = std::move(reject)] (auto&&... args) {
      processStatusDatasetInterest(std::forward<decltype(args)>(args)..., auth, accept, reject);
    };
 
  m_handlers[relPrefix] = [this, miss = std::move(missContinuation)] (auto&&... args) {
    queryStorage(std::forward<decltype(args)>(args)..., miss);
  };
}
 
void
Dispatcher::processStatusDatasetInterest(const Name& prefix,
                                         const Interest& interest,
                                         const Authorization& authorization,
                                         const AuthorizationAcceptedCallback& accepted,
                                         const AuthorizationRejectedCallback& rejected)
{
  const Name& interestName = interest.getName();
  bool endsWithVersionOrSegment = interestName.size() >= 1 &&
                                  (interestName[-1].isVersion() || interestName[-1].isSegment());
  if (endsWithVersionOrSegment) {
    return;
  }
 
  AcceptContinuation accept = [=] (const auto& req) { accepted(req, prefix, interest, nullptr); };
  RejectContinuation reject = [=] (RejectReply reply) { rejected(reply, interest); };
  authorization(prefix, interest, nullptr, accept, reject);
}
 
void
Dispatcher::processAuthorizedStatusDatasetInterest(const Name& prefix,
                                                   const Interest& interest,
                                                   const StatusDatasetHandler& handler)
{
  StatusDatasetContext context(interest,
    [this] (auto&&... args) {
      sendStatusDatasetSegment(std::forward<decltype(args)>(args)...);
    },
    [this, interest] (auto&&... args) {
      sendControlResponse(std::forward<decltype(args)>(args)..., interest, true);
    });
  handler(prefix, interest, context);
}
 
void
Dispatcher::sendStatusDatasetSegment(const Name& dataName, const Block& content, bool isFinalBlock)
{
  // the first segment will be sent to both places (the face and the in-memory storage)
  // other segments will be inserted to the in-memory storage only
  auto destination = SendDestination::IMS;
  if (dataName[-1].toSegment() == 0) {
    destination = SendDestination::FACE_AND_IMS;
  }
 
  MetaInfo metaInfo;
  if (isFinalBlock) {
    metaInfo.setFinalBlock(dataName[-1]);
  }
 
  sendData(dataName, content, metaInfo, destination);
}
 
PostNotification
Dispatcher::addNotificationStream(const PartialName& relPrefix)
{
  if (!m_topLevelPrefixes.empty()) {
    NDN_THROW(std::domain_error("one or more top-level prefix has been added"));
  }
 
  if (isOverlappedWithOthers(relPrefix)) {
    NDN_THROW(std::out_of_range("notification stream name overlaps"));
  }
 
  // register a handler for the subscriber of this notification stream
  // keep silent if Interest does not match a stored notification
  m_handlers[relPrefix] = [this] (auto&&... args) {
    queryStorage(std::forward<decltype(args)>(args)..., nullptr);
  };
  m_streams[relPrefix] = 0;
 
  return [=] (const Block& b) { postNotification(b, relPrefix); };
}
 
void
Dispatcher::postNotification(const Block& notification, const PartialName& relPrefix)
{
  if (m_topLevelPrefixes.size() != 1) {
    NDN_LOG_WARN("postNotification: no top-level prefix or too many top-level prefixes");
    return;
  }
 
  Name streamName(m_topLevelPrefixes.begin()->first);
  streamName.append(relPrefix);
  streamName.appendSequenceNumber(m_streams[streamName]++);
 
  // notification is sent out via the face after inserting into the in-memory storage,
  // because a request may be pending in the PIT
  sendData(streamName, notification, {}, SendDestination::FACE_AND_IMS);
}
 
} // namespace ndn::mgmt