blob: df0fcd14a03347fdc17be635e011d97271921ca2 [file] [log] [blame]
/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
/*
* Copyright (c) 2012-2017 University of California, Los Angeles
*
* This file is part of ChronoSync, synchronization library for distributed realtime
* applications for NDN.
*
* ChronoSync is free software: you can redistribute it and/or modify it under the terms
* of the GNU General Public License as published by the Free Software Foundation, either
* version 3 of the License, or (at your option) any later version.
*
* ChronoSync 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 General Public License for more details.
*
* You should have received a copy of the GNU General Public License along with
* ChronoSync, e.g., in COPYING.md file. If not, see <http://www.gnu.org/licenses/>.
*
* @author Zhenkai Zhu <http://irl.cs.ucla.edu/~zhenkai/>
* @author Chaoyi Bian <bcy@pku.edu.cn>
* @author Alexander Afanasyev <http://lasr.cs.ucla.edu/afanasyev/index.html>
* @author Yingdi Yu <yingdi@cs.ucla.edu>
* @author Sonu Mishra <https://www.linkedin.com/in/mishrasonu>
*/
#include "logic.hpp"
#include "logger.hpp"
#include <ndn-cxx/util/string-helper.hpp>
INIT_LOGGER(Logic);
#ifdef _DEBUG
#define _LOG_DEBUG_ID(v) _LOG_DEBUG("Instance" << m_instanceId << ": " << v)
#else
#define _LOG_DEBUG_ID(v) _LOG_DEBUG(v)
#endif
namespace chronosync {
using ndn::ConstBufferPtr;
using ndn::EventId;
const uint8_t EMPTY_DIGEST_VALUE[] = {
0xe3, 0xb0, 0xc4, 0x42, 0x98, 0xfc, 0x1c, 0x14,
0x9a, 0xfb, 0xf4, 0xc8, 0x99, 0x6f, 0xb9, 0x24,
0x27, 0xae, 0x41, 0xe4, 0x64, 0x9b, 0x93, 0x4c,
0xa4, 0x95, 0x99, 0x1b, 0x78, 0x52, 0xb8, 0x55
};
#ifdef _DEBUG
int Logic::m_instanceCounter = 0;
#endif
const ndn::Name Logic::DEFAULT_NAME;
const ndn::Name Logic::EMPTY_NAME;
const std::shared_ptr<ndn::Validator> Logic::DEFAULT_VALIDATOR;
const time::steady_clock::Duration Logic::DEFAULT_RESET_TIMER = time::seconds(0);
const time::steady_clock::Duration Logic::DEFAULT_CANCEL_RESET_TIMER = time::milliseconds(500);
const time::milliseconds Logic::DEFAULT_RESET_INTEREST_LIFETIME(1000);
const time::milliseconds Logic::DEFAULT_SYNC_INTEREST_LIFETIME(1000);
const time::milliseconds Logic::DEFAULT_SYNC_REPLY_FRESHNESS(1000);
const time::milliseconds Logic::DEFAULT_RECOVERY_INTEREST_LIFETIME(1000);
const ndn::ConstBufferPtr Logic::EMPTY_DIGEST(new ndn::Buffer(EMPTY_DIGEST_VALUE, 32));
const ndn::name::Component Logic::RESET_COMPONENT("reset");
const ndn::name::Component Logic::RECOVERY_COMPONENT("recovery");
Logic::Logic(ndn::Face& face,
const Name& syncPrefix,
const Name& defaultUserPrefix,
const UpdateCallback& onUpdate,
const Name& defaultSigningId,
std::shared_ptr<ndn::Validator> validator,
const time::steady_clock::Duration& resetTimer,
const time::steady_clock::Duration& cancelResetTimer,
const time::milliseconds& resetInterestLifetime,
const time::milliseconds& syncInterestLifetime,
const time::milliseconds& syncReplyFreshness,
const time::milliseconds& recoveryInterestLifetime)
: m_face(face)
, m_syncPrefix(syncPrefix)
, m_defaultUserPrefix(defaultUserPrefix)
, m_interestTable(m_face.getIoService())
, m_outstandingInterestId(0)
, m_isInReset(false)
, m_needPeriodReset(resetTimer > time::steady_clock::Duration::zero())
, m_onUpdate(onUpdate)
, m_scheduler(m_face.getIoService())
, m_randomGenerator(static_cast<unsigned int>(std::time(0)))
, m_rangeUniformRandom(m_randomGenerator, boost::uniform_int<>(100,500))
, m_reexpressionJitter(m_randomGenerator, boost::uniform_int<>(100,500))
, m_resetTimer(resetTimer)
, m_cancelResetTimer(cancelResetTimer)
, m_resetInterestLifetime(resetInterestLifetime)
, m_syncInterestLifetime(syncInterestLifetime)
, m_syncReplyFreshness(syncReplyFreshness)
, m_recoveryInterestLifetime(recoveryInterestLifetime)
, m_validator(validator)
{
#ifdef _DEBUG
m_instanceId = m_instanceCounter++;
#endif
_LOG_DEBUG_ID(">> Logic::Logic");
addUserNode(m_defaultUserPrefix, defaultSigningId);
m_syncReset = m_syncPrefix;
m_syncReset.append("reset");
_LOG_DEBUG_ID("Listen to: " << m_syncPrefix);
m_syncRegisteredPrefixId =
m_face.setInterestFilter(m_syncPrefix,
bind(&Logic::onSyncInterest, this, _1, _2),
bind(&Logic::onSyncRegisterFailed, this, _1, _2));
sendSyncInterest();
_LOG_DEBUG_ID("<< Logic::Logic");
}
Logic::~Logic()
{
m_scheduler.cancelAllEvents();
m_interestTable.clear();
m_face.shutdown();
}
void
Logic::reset(bool isOnInterest)
{
m_isInReset = true;
m_state.reset();
m_log.clear();
if (!isOnInterest)
sendResetInterest();
// reset outstanding interest name, so that data for previous interest will be dropped.
if (m_outstandingInterestId != 0) {
m_face.removePendingInterest(m_outstandingInterestId);
m_outstandingInterestId = 0;
}
sendSyncInterest();
if (static_cast<bool>(m_delayedInterestProcessingId))
m_scheduler.cancelEvent(m_delayedInterestProcessingId);
m_delayedInterestProcessingId =
m_scheduler.scheduleEvent(m_cancelResetTimer,
bind(&Logic::cancelReset, this));
}
void
Logic::setDefaultUserPrefix(const Name& defaultUserPrefix)
{
if (defaultUserPrefix != EMPTY_NAME) {
if (m_nodeList.find(defaultUserPrefix) != m_nodeList.end()) {
m_defaultUserPrefix = defaultUserPrefix;
}
}
}
void
Logic::addUserNode(const Name& userPrefix, const Name& signingId)
{
if (userPrefix == EMPTY_NAME)
return;
if (m_defaultUserPrefix == EMPTY_NAME) {
m_defaultUserPrefix = userPrefix;
}
if (m_nodeList.find(userPrefix) == m_nodeList.end()) {
m_nodeList[userPrefix].userPrefix = userPrefix;
m_nodeList[userPrefix].signingId = signingId;
Name sessionName = userPrefix;
sessionName.appendNumber(ndn::time::toUnixTimestamp(ndn::time::system_clock::now()).count());
m_nodeList[userPrefix].sessionName = sessionName;
m_nodeList[userPrefix].seqNo = 0;
reset(false);
}
}
void
Logic::removeUserNode(const Name& userPrefix)
{
auto userNode = m_nodeList.find(userPrefix);
if (userNode != m_nodeList.end()) {
m_nodeList.erase(userNode);
if (m_defaultUserPrefix == userPrefix) {
if (!m_nodeList.empty()) {
m_defaultUserPrefix = m_nodeList.begin()->second.userPrefix;
}
else {
m_defaultUserPrefix = EMPTY_NAME;
}
}
reset(false);
}
}
const Name&
Logic::getSessionName(Name prefix)
{
if (prefix == EMPTY_NAME)
prefix = m_defaultUserPrefix;
auto node = m_nodeList.find(prefix);
if (node != m_nodeList.end())
return node->second.sessionName;
else
BOOST_THROW_EXCEPTION(Error("Refer to non-existent node:" + prefix.toUri()));
}
const SeqNo&
Logic::getSeqNo(Name prefix)
{
if (prefix == EMPTY_NAME)
prefix = m_defaultUserPrefix;
auto node = m_nodeList.find(prefix);
if (node != m_nodeList.end())
return node->second.seqNo;
else
BOOST_THROW_EXCEPTION(Logic::Error("Refer to non-existent node:" + prefix.toUri()));
}
void
Logic::updateSeqNo(const SeqNo& seqNo, const Name& updatePrefix)
{
Name prefix;
if (updatePrefix == EMPTY_NAME) {
if (m_defaultUserPrefix == EMPTY_NAME)
return;
prefix = m_defaultUserPrefix;
}
else
prefix = updatePrefix;
auto it = m_nodeList.find(prefix);
if (it != m_nodeList.end()) {
NodeInfo& node = it->second;
_LOG_DEBUG_ID(">> Logic::updateSeqNo");
_LOG_DEBUG_ID("seqNo: " << seqNo << " m_seqNo: " << node.seqNo);
if (seqNo < node.seqNo || seqNo == 0)
return;
node.seqNo = seqNo;
_LOG_DEBUG_ID("updateSeqNo: m_seqNo " << node.seqNo);
if (!m_isInReset) {
_LOG_DEBUG_ID("updateSeqNo: not in Reset ");
ndn::ConstBufferPtr previousRoot = m_state.getRootDigest();
{
std::string hash = ndn::toHex(previousRoot->buf(), previousRoot->size(), false);
_LOG_DEBUG_ID("Hash: " << hash);
}
bool isInserted = false;
bool isUpdated = false;
SeqNo oldSeq;
boost::tie(isInserted, isUpdated, oldSeq) = m_state.update(node.sessionName,
node.seqNo);
_LOG_DEBUG_ID("Insert: " << std::boolalpha << isInserted);
_LOG_DEBUG_ID("Updated: " << std::boolalpha << isUpdated);
if (isInserted || isUpdated) {
DiffStatePtr commit = make_shared<DiffState>();
commit->update(node.sessionName, node.seqNo);
commit->setRootDigest(m_state.getRootDigest());
insertToDiffLog(commit, previousRoot);
satisfyPendingSyncInterests(prefix, commit);
formAndSendExcludeInterest(prefix, *commit, previousRoot);
}
}
}
}
ConstBufferPtr
Logic::getRootDigest() const
{
return m_state.getRootDigest();
}
void
Logic::printState(std::ostream& os) const
{
BOOST_FOREACH(ConstLeafPtr leaf, m_state.getLeaves())
{
os << *leaf << "\n";
}
}
std::set<Name>
Logic::getSessionNames() const
{
std::set<Name> sessionNames;
BOOST_FOREACH(ConstLeafPtr leaf, m_state.getLeaves())
{
sessionNames.insert(leaf->getSessionName());
}
return sessionNames;
}
void
Logic::onSyncInterest(const Name& prefix, const Interest& interest)
{
_LOG_DEBUG_ID(">> Logic::onSyncInterest");
Name name = interest.getName();
_LOG_DEBUG_ID("InterestName: " << name);
if (name.size() >= 1 && RESET_COMPONENT == name.get(-1)) {
processResetInterest(interest);
}
else if (name.size() >= 2 && RECOVERY_COMPONENT == name.get(-2)) {
processRecoveryInterest(interest);
}
else {
processSyncInterest(interest.shared_from_this());
}
_LOG_DEBUG_ID("<< Logic::onSyncInterest");
}
void
Logic::onSyncRegisterFailed(const Name& prefix, const std::string& msg)
{
//Sync prefix registration failed
_LOG_DEBUG_ID(">> Logic::onSyncRegisterFailed");
}
void
Logic::onSyncData(const Interest& interest, const Data& data)
{
_LOG_DEBUG_ID(">> Logic::onSyncData");
// if (static_cast<bool>(m_validator))
// m_validator->validate(data,
// bind(&Logic::onSyncDataValidated, this, _1),
// bind(&Logic::onSyncDataValidationFailed, this, _1));
// else
// onSyncDataValidated(data.shared_from_this());
if (interest.getExclude().empty()) {
_LOG_DEBUG_ID("First data");
onSyncDataValidated(data.shared_from_this());
}
else {
_LOG_DEBUG_ID("Data obtained using exclude filter");
onSyncDataValidated(data.shared_from_this(), false);
}
sendExcludeInterest(interest, data);
_LOG_DEBUG_ID("<< Logic::onSyncData");
}
void
Logic::onResetData(const Interest& interest, const Data& data)
{
// This should not happened, drop the received data.
}
void
Logic::onSyncTimeout(const Interest& interest)
{
// It is OK. Others will handle the time out situation.
_LOG_DEBUG_ID(">> Logic::onSyncTimeout");
_LOG_DEBUG_ID("Interest: " << interest.getName());
_LOG_DEBUG_ID("<< Logic::onSyncTimeout");
}
void
Logic::onSyncDataValidationFailed(const shared_ptr<const Data>& data)
{
// SyncReply cannot be validated.
}
void
Logic::onSyncDataValidated(const shared_ptr<const Data>& data, bool firstData)
{
Name name = data->getName();
ConstBufferPtr digest = make_shared<ndn::Buffer>(name.get(-1).value(), name.get(-1).value_size());
processSyncData(name, digest, data->getContent().blockFromValue(), firstData);
}
void
Logic::processSyncInterest(const shared_ptr<const Interest>& interest,
bool isTimedProcessing/*=false*/)
{
_LOG_DEBUG_ID(">> Logic::processSyncInterest");
const Name& name = interest->getName();
ConstBufferPtr digest =
make_shared<ndn::Buffer>(name.get(-1).value(), name.get(-1).value_size());
ConstBufferPtr rootDigest = m_state.getRootDigest();
// If the digest of the incoming interest is the same as root digest
// Put the interest into InterestTable
if (*rootDigest == *digest) {
_LOG_DEBUG_ID("Oh, we are in the same state");
m_interestTable.insert(interest, digest, false);
if (!m_isInReset)
return;
if (!isTimedProcessing) {
_LOG_DEBUG_ID("Non timed processing in reset");
// Still in reset, our own seq has not been put into state yet
// Do not hurry, some others may be also resetting and may send their reply
if (static_cast<bool>(m_delayedInterestProcessingId))
m_scheduler.cancelEvent(m_delayedInterestProcessingId);
time::milliseconds after(m_rangeUniformRandom());
_LOG_DEBUG_ID("After: " << after);
m_delayedInterestProcessingId =
m_scheduler.scheduleEvent(after,
bind(&Logic::processSyncInterest, this, interest, true));
}
else {
_LOG_DEBUG_ID("Timed processing in reset");
// Now we can get out of reset state by putting our own stuff into m_state.
cancelReset();
}
return;
}
// If the digest of incoming interest is an "empty" digest
if (*digest == *EMPTY_DIGEST) {
_LOG_DEBUG_ID("Poor guy, he knows nothing");
sendSyncData(m_defaultUserPrefix, name, m_state);
return;
}
DiffStateContainer::iterator stateIter = m_log.find(digest);
// If the digest of incoming interest can be found from the log
if (stateIter != m_log.end()) {
_LOG_DEBUG_ID("It is ok, you are so close");
sendSyncData(m_defaultUserPrefix, name, *(*stateIter)->diff());
return;
}
if (!isTimedProcessing) {
_LOG_DEBUG_ID("Let's wait, just wait for a while");
// Do not hurry, some incoming SyncReplies may help us to recognize the digest
bool doesExist = m_interestTable.has(digest);
m_interestTable.insert(interest, digest, true);
if (doesExist)
// Original comment (not sure): somebody else replied, so restart random-game timer
// YY: Get the same SyncInterest again, refresh the timer
m_scheduler.cancelEvent(m_delayedInterestProcessingId);
m_delayedInterestProcessingId =
m_scheduler.scheduleEvent(time::milliseconds(m_rangeUniformRandom()),
bind(&Logic::processSyncInterest, this, interest, true));
}
else {
// OK, nobody is helping us, just tell the truth.
_LOG_DEBUG_ID("OK, nobody is helping us, let us try to recover");
m_interestTable.erase(digest);
sendRecoveryInterest(digest);
}
_LOG_DEBUG_ID("<< Logic::processSyncInterest");
}
void
Logic::processResetInterest(const Interest& interest)
{
_LOG_DEBUG_ID(">> Logic::processResetInterest");
reset(true);
}
void
Logic::processSyncData(const Name& name,
ndn::ConstBufferPtr digest,
const Block& syncReplyBlock,
bool firstData)
{
_LOG_DEBUG_ID(">> Logic::processSyncData");
DiffStatePtr commit = make_shared<DiffState>();
ndn::ConstBufferPtr previousRoot = m_state.getRootDigest();
try {
m_interestTable.erase(digest); // Remove satisfied interest from PIT
State reply;
reply.wireDecode(syncReplyBlock);
std::vector<MissingDataInfo> v;
BOOST_FOREACH(ConstLeafPtr leaf, reply.getLeaves().get<ordered>())
{
BOOST_ASSERT(leaf != 0);
const Name& info = leaf->getSessionName();
SeqNo seq = leaf->getSeq();
bool isInserted = false;
bool isUpdated = false;
SeqNo oldSeq;
boost::tie(isInserted, isUpdated, oldSeq) = m_state.update(info, seq);
if (isInserted || isUpdated) {
commit->update(info, seq);
oldSeq++;
MissingDataInfo mdi = {info, oldSeq, seq};
v.push_back(mdi);
}
}
if (!v.empty()) {
m_onUpdate(v);
commit->setRootDigest(m_state.getRootDigest());
insertToDiffLog(commit, previousRoot);
}
else {
_LOG_DEBUG_ID("What? nothing new");
}
}
catch (const State::Error&) {
_LOG_DEBUG_ID("Something really fishy happened during state decoding");
// Something really fishy happened during state decoding;
commit.reset();
return;
}
if (static_cast<bool>(commit) && !commit->getLeaves().empty() && firstData) {
// state changed and it is safe to express a new interest
time::steady_clock::Duration after = time::milliseconds(m_reexpressionJitter());
_LOG_DEBUG_ID("Reschedule sync interest after: " << after);
EventId eventId = m_scheduler.scheduleEvent(after,
bind(&Logic::sendSyncInterest, this));
m_scheduler.cancelEvent(m_reexpressingInterestId);
m_reexpressingInterestId = eventId;
}
}
void
Logic::satisfyPendingSyncInterests(const Name& updatedPrefix, ConstDiffStatePtr commit)
{
_LOG_DEBUG_ID(">> Logic::satisfyPendingSyncInterests");
try {
_LOG_DEBUG_ID("InterestTable size: " << m_interestTable.size());
for (InterestTable::const_iterator it = m_interestTable.begin();
it != m_interestTable.end(); it++) {
ConstUnsatisfiedInterestPtr request = *it;
if (request->isUnknown)
sendSyncData(updatedPrefix, request->interest->getName(), m_state);
else
sendSyncData(updatedPrefix, request->interest->getName(), *commit);
}
m_interestTable.clear();
}
catch (const InterestTable::Error&) {
// ok. not really an error
}
_LOG_DEBUG_ID("<< Logic::satisfyPendingSyncInterests");
}
void
Logic::insertToDiffLog(DiffStatePtr commit, ndn::ConstBufferPtr previousRoot)
{
_LOG_DEBUG_ID(">> Logic::insertToDiffLog");
// Connect to the history
if (!m_log.empty())
(*m_log.find(previousRoot))->setNext(commit);
// Insert the commit
m_log.erase(commit->getRootDigest());
m_log.insert(commit);
_LOG_DEBUG_ID("<< Logic::insertToDiffLog");
}
void
Logic::sendResetInterest()
{
_LOG_DEBUG_ID(">> Logic::sendResetInterest");
if (m_needPeriodReset) {
_LOG_DEBUG_ID("Need Period Reset");
_LOG_DEBUG_ID("ResetTimer: " << m_resetTimer);
EventId eventId =
m_scheduler.scheduleEvent(m_resetTimer + ndn::time::milliseconds(m_reexpressionJitter()),
bind(&Logic::sendResetInterest, this));
m_scheduler.cancelEvent(m_resetInterestId);
m_resetInterestId = eventId;
}
Interest interest(m_syncReset);
interest.setMustBeFresh(true);
interest.setInterestLifetime(m_resetInterestLifetime);
m_face.expressInterest(interest,
bind(&Logic::onResetData, this, _1, _2),
bind(&Logic::onSyncTimeout, this, _1), // Nack
bind(&Logic::onSyncTimeout, this, _1));
_LOG_DEBUG_ID("<< Logic::sendResetInterest");
}
void
Logic::sendSyncInterest()
{
_LOG_DEBUG_ID(">> Logic::sendSyncInterest");
Name interestName;
interestName.append(m_syncPrefix)
.append(ndn::name::Component(*m_state.getRootDigest()));
m_outstandingInterestName = interestName;
#ifdef _DEBUG
printDigest(m_state.getRootDigest());
#endif
EventId eventId =
m_scheduler.scheduleEvent(m_syncInterestLifetime / 2 +
ndn::time::milliseconds(m_reexpressionJitter()),
bind(&Logic::sendSyncInterest, this));
m_scheduler.cancelEvent(m_reexpressingInterestId);
m_reexpressingInterestId = eventId;
Interest interest(interestName);
interest.setMustBeFresh(true);
interest.setInterestLifetime(m_syncInterestLifetime);
m_outstandingInterestId = m_face.expressInterest(interest,
bind(&Logic::onSyncData, this, _1, _2),
bind(&Logic::onSyncTimeout, this, _1), // Nack
bind(&Logic::onSyncTimeout, this, _1));
_LOG_DEBUG_ID("Send interest: " << interest.getName());
_LOG_DEBUG_ID("<< Logic::sendSyncInterest");
}
void
Logic::sendSyncData(const Name& nodePrefix, const Name& name, const State& state)
{
_LOG_DEBUG_ID(">> Logic::sendSyncData");
shared_ptr<Data> syncReply = make_shared<Data>(name);
syncReply->setContent(state.wireEncode());
syncReply->setFreshnessPeriod(m_syncReplyFreshness);
if (m_nodeList.find(nodePrefix) == m_nodeList.end())
return;
if (m_nodeList[nodePrefix].signingId.empty())
m_keyChain.sign(*syncReply);
else
m_keyChain.sign(*syncReply, security::signingByIdentity(m_nodeList[nodePrefix].signingId));
m_face.put(*syncReply);
// checking if our own interest got satisfied
if (m_outstandingInterestName == name) {
// remove outstanding interest
if (m_outstandingInterestId != 0) {
m_face.removePendingInterest(m_outstandingInterestId);
m_outstandingInterestId = 0;
}
// re-schedule sending Sync interest
time::milliseconds after(m_reexpressionJitter());
_LOG_DEBUG_ID("Satisfy our own interest");
_LOG_DEBUG_ID("Reschedule sync interest after " << after);
EventId eventId = m_scheduler.scheduleEvent(after, bind(&Logic::sendSyncInterest, this));
m_scheduler.cancelEvent(m_reexpressingInterestId);
m_reexpressingInterestId = eventId;
}
_LOG_DEBUG_ID("<< Logic::sendSyncData");
}
void
Logic::cancelReset()
{
_LOG_DEBUG_ID(">> Logic::cancelReset");
if (!m_isInReset)
return;
m_isInReset = false;
for (const auto& node : m_nodeList) {
updateSeqNo(node.second.seqNo, node.first);
}
_LOG_DEBUG_ID("<< Logic::cancelReset");
}
void
Logic::printDigest(ndn::ConstBufferPtr digest)
{
std::string hash = ndn::toHex(digest->buf(), digest->size(), false);
_LOG_DEBUG_ID("Hash: " << hash);
}
void
Logic::sendRecoveryInterest(ndn::ConstBufferPtr digest)
{
_LOG_DEBUG_ID(">> Logic::sendRecoveryInterest");
Name interestName;
interestName.append(m_syncPrefix)
.append(RECOVERY_COMPONENT)
.append(ndn::name::Component(*digest));
Interest interest(interestName);
interest.setMustBeFresh(true);
interest.setInterestLifetime(m_recoveryInterestLifetime);
m_face.expressInterest(interest,
bind(&Logic::onRecoveryData, this, _1, _2),
bind(&Logic::onRecoveryTimeout, this, _1), // Nack
bind(&Logic::onRecoveryTimeout, this, _1));
_LOG_DEBUG_ID("interest: " << interest.getName());
_LOG_DEBUG_ID("<< Logic::sendRecoveryInterest");
}
void
Logic::processRecoveryInterest(const Interest& interest)
{
_LOG_DEBUG_ID(">> Logic::processRecoveryInterest");
const Name& name = interest.getName();
ConstBufferPtr digest = make_shared<ndn::Buffer>(name.get(-1).value(), name.get(-1).value_size());
ConstBufferPtr rootDigest = m_state.getRootDigest();
DiffStateContainer::iterator stateIter = m_log.find(digest);
if (stateIter != m_log.end() || *digest == *EMPTY_DIGEST || *rootDigest == *digest) {
_LOG_DEBUG_ID("I can help you recover");
sendSyncData(m_defaultUserPrefix, name, m_state);
return;
}
_LOG_DEBUG_ID("<< Logic::processRecoveryInterest");
}
void
Logic::onRecoveryData(const Interest& interest, const Data& data)
{
_LOG_DEBUG_ID(">> Logic::onRecoveryData");
onSyncDataValidated(data.shared_from_this());
_LOG_DEBUG_ID("<< Logic::onRecoveryData");
}
void
Logic::onRecoveryTimeout(const Interest& interest)
{
_LOG_DEBUG_ID(">> Logic::onRecoveryTimeout");
_LOG_DEBUG_ID("Interest: " << interest.getName());
_LOG_DEBUG_ID("<< Logic::onRecoveryTimeout");
}
void
Logic::sendExcludeInterest(const Interest& interest, const Data& data)
{
_LOG_DEBUG_ID(">> Logic::sendExcludeInterest");
Name interestName = interest.getName();
Interest excludeInterest(interestName);
Exclude exclude = interest.getExclude();
exclude.excludeOne(data.getFullName().get(-1));
excludeInterest.setExclude(exclude);
excludeInterest.setInterestLifetime(m_syncInterestLifetime);
if (excludeInterest.wireEncode().size() > ndn::MAX_NDN_PACKET_SIZE) {
return;
}
m_face.expressInterest(excludeInterest,
bind(&Logic::onSyncData, this, _1, _2),
bind(&Logic::onSyncTimeout, this, _1), // Nack
bind(&Logic::onSyncTimeout, this, _1));
_LOG_DEBUG_ID("Send interest: " << excludeInterest.getName());
_LOG_DEBUG_ID("<< Logic::sendExcludeInterest");
}
void
Logic::formAndSendExcludeInterest(const Name& nodePrefix, const State& commit, ndn::ConstBufferPtr previousRoot)
{
_LOG_DEBUG_ID(">> Logic::formAndSendExcludeInterest");
Name interestName;
interestName.append(m_syncPrefix)
.append(ndn::name::Component(*previousRoot));
Interest interest(interestName);
shared_ptr<Data> data = make_shared<Data>(interestName);
data->setContent(commit.wireEncode());
data->setFreshnessPeriod(m_syncReplyFreshness);
if (m_nodeList.find(nodePrefix) == m_nodeList.end())
return;
if (m_nodeList[nodePrefix].signingId.empty())
m_keyChain.sign(*data);
else
m_keyChain.sign(*data, security::signingByIdentity(m_nodeList[nodePrefix].signingId));
sendExcludeInterest(interest, *data);
_LOG_DEBUG_ID("<< Logic::formAndSendExcludeInterest");
}
} // namespace chronosync