content server
Change-Id: I08eb9da3d7fcdb337b02c73695dd648bcfe82c81
diff --git a/src/content-server.cpp b/src/content-server.cpp
index b795afe..888baea 100644
--- a/src/content-server.cpp
+++ b/src/content-server.cpp
@@ -19,149 +19,135 @@
*/
#include "content-server.hpp"
-#include "logging.hpp"
-#include "periodic-task.hpp"
-#include "simple-interval-generator.hpp"
-#include "task.hpp"
+#include "core/logging.hpp"
+
+#include <ndn-cxx/util/string-helper.hpp>
+#include <ndn-cxx/security/signing-helpers.hpp>
+
#include <boost/lexical_cast.hpp>
-#include <boost/make_shared.hpp>
-#include <utility>
+
+namespace ndn {
+namespace chronoshare {
_LOG_INIT(ContentServer);
-using namespace Ndnx;
-using namespace std;
-using namespace boost;
-
static const int DB_CACHE_LIFETIME = 60;
-ContentServer::ContentServer(CcnxWrapperPtr ccnx, ActionLogPtr actionLog,
- const boost::filesystem::path& rootDir, const Ccnx::Name& userName,
+ContentServer::ContentServer(Face& face, ActionLogPtr actionLog,
+ const boost::filesystem::path& rootDir, const Name& userName,
const std::string& sharedFolderName, const std::string& appName,
- int freshness)
- : m_ndnx(ndnx)
+ KeyChain& keyChain, int freshness)
+ : m_face(face)
, m_actionLog(actionLog)
, m_dbFolder(rootDir / ".chronoshare")
, m_freshness(freshness)
- , m_scheduler(new Scheduler())
+ , m_scheduler(face.getIoService())
+ , m_flushStateDbCacheEvent(m_scheduler)
, m_userName(userName)
, m_sharedFolderName(sharedFolderName)
, m_appName(appName)
+ , m_keyChain(keyChain)
{
- m_scheduler->start();
- TaskPtr flushStaleDbCacheTask =
- boost::make_shared<PeriodicTask>(boost::bind(&ContentServer::flushStaleDbCache, this),
- "flush-state-db-cache", m_scheduler,
- boost::make_shared<SimpleIntervalGenerator>(DB_CACHE_LIFETIME));
- m_scheduler->addTask(flushStaleDbCacheTask);
+ m_flushStateDbCacheEvent = m_scheduler.scheduleEvent(time::seconds(DB_CACHE_LIFETIME),
+ bind(&ContentServer::flushStaleDbCache, this));
}
ContentServer::~ContentServer()
{
- m_scheduler->shutdown();
-
ScopedLock lock(m_mutex);
- for (PrefixIt forwardingHint = m_prefixes.begin(); forwardingHint != m_prefixes.end();
- ++forwardingHint) {
- m_ccnx->clearInterestFilter(*forwardingHint);
+ for (FilterIdIt it = m_interestFilterIds.begin(); it != m_interestFilterIds.end(); ++it) {
+ m_face.unsetInterestFilter(it->second);
}
- m_prefixes.clear();
+ m_interestFilterIds.clear();
}
void
ContentServer::registerPrefix(const Name& forwardingHint)
{
// Format for files: /<forwarding-hint>/<device_name>/<appname>/file/<hash>/<segment>
- // Format for actions: /<forwarding-hint>/<device_name>/<appname>/action/<shared-folder>/<action-seq>
+ // Format for actions:
+ // /<forwarding-hint>/<device_name>/<appname>/action/<shared-folder>/<action-seq>
_LOG_DEBUG(">> content server: register " << forwardingHint);
- m_ccnx->setInterestFilter(forwardingHint,
- bind(&ContentServer::filterAndServe, this, forwardingHint, _1));
-
ScopedLock lock(m_mutex);
- m_prefixes.insert(forwardingHint);
+ m_interestFilterIds[forwardingHint] =
+ m_face.setInterestFilter(InterestFilter(forwardingHint),
+ bind(&ContentServer::filterAndServe, this, _1, _2),
+ RegisterPrefixSuccessCallback(), RegisterPrefixFailureCallback());
}
void
ContentServer::deregisterPrefix(const Name& forwardingHint)
{
_LOG_DEBUG("<< content server: deregister " << forwardingHint);
- m_ccnx->clearInterestFilter(forwardingHint);
+ m_face.unsetInterestFilter(m_interestFilterIds[forwardingHint]);
ScopedLock lock(m_mutex);
- m_prefixes.erase(forwardingHint);
+ m_interestFilterIds.erase(forwardingHint);
}
-
void
ContentServer::filterAndServeImpl(const Name& forwardingHint, const Name& name, const Name& interest)
{
// interest for files: /<forwarding-hint>/<device_name>/<appname>/file/<hash>/<segment>
- // interest for actions: /<forwarding-hint>/<device_name>/<appname>/action/<shared-folder>/<action-seq>
+ // interest for actions:
+ // /<forwarding-hint>/<device_name>/<appname>/action/<shared-folder>/<action-seq>
// name for files: /<device_name>/<appname>/file/<hash>/<segment>
// name for actions: /<device_name>/<appname>/action/<shared-folder>/<action-seq>
- try {
- if (name.size() >= 4 && name.getCompFromBackAsString(3) == m_appName) {
- string type = name.getCompFromBackAsString(2);
- if (type == "file") {
- serve_File(forwardingHint, name, interest);
- }
- else if (type == "action") {
- string folder = name.getCompFromBackAsString(1);
- if (folder == m_sharedFolderName) {
- serve_Action(forwardingHint, name, interest);
- }
+ if (name.size() >= 4 && name.get(-4).toUri() == m_appName) {
+ std::string type = name.get(-3).toUri();
+ if (type == "file") {
+ serve_File(forwardingHint, name, interest);
+ }
+ else if (type == "action") {
+ std::string folder = name.get(-2).toUri();
+ if (folder == m_sharedFolderName) {
+ serve_Action(forwardingHint, name, interest);
}
}
}
- catch (Ccnx::NameException& ne) {
- // ignore any unexpected interests and errors
- _LOG_ERROR(boost::get_error_info<Ccnx::error_info_str>(ne));
- }
}
void
-ContentServer::filterAndServe(Name forwardingHint, const Name& interest)
+ContentServer::filterAndServe(const InterestFilter& interestFilter, const Interest& interestTrue)
{
- try {
- if (forwardingHint.size() > 0 && m_userName.size() >= forwardingHint.size() &&
- m_userName.getPartialName(0, forwardingHint.size()) == forwardingHint) {
- filterAndServeImpl(Name("/"), interest, interest); // try without forwarding hints
- }
+ Name forwardingHint = Name(interestFilter);
+ Name interest = interestTrue.getName();
+ _LOG_DEBUG("I'm serving ForwardingHint: " << forwardingHint << " Interest: " << interest);
+ if (forwardingHint.size() > 0 && m_userName.size() >= forwardingHint.size() &&
+ m_userName.getSubName(0, forwardingHint.size()) == forwardingHint) {
+ _LOG_DEBUG("Triggered without Forwardinghint!");
+ filterAndServeImpl(Name("/"), interest, interest); // try without forwarding hints
+ }
- filterAndServeImpl(forwardingHint, interest.getPartialName(forwardingHint.size()),
- interest); // always try with hint... :( have to
- }
- catch (Ccnx::NameException& ne) {
- // ignore any unexpected interests and errors
- _LOG_ERROR(boost::get_error_info<Ccnx::error_info_str>(ne));
- }
+ _LOG_DEBUG("Triggered with Forwardinghint~!");
+ filterAndServeImpl(forwardingHint, interest.getSubName(forwardingHint.size()),
+ interest); // always try with hint... :( have to
}
void
ContentServer::serve_Action(const Name& forwardingHint, const Name& name, const Name& interest)
{
- _LOG_DEBUG(">> content server serving ACTION, hint: " << forwardingHint
- << ", interest: " << interest);
- m_scheduler->scheduleOneTimeTask(m_scheduler, 0, bind(&ContentServer::serve_Action_Execute, this,
- forwardingHint, name, interest),
- boost::lexical_cast<string>(name));
+ _LOG_DEBUG(
+ ">> content server serving ACTION, hint: " << forwardingHint << ", interest: " << interest);
+ m_scheduler.scheduleEvent(time::seconds(0),
+ bind(&ContentServer::serve_Action_Execute, this, forwardingHint, name,
+ interest));
// need to unlock ccnx mutex... or at least don't lock it
}
void
ContentServer::serve_File(const Name& forwardingHint, const Name& name, const Name& interest)
{
- _LOG_DEBUG(">> content server serving FILE, hint: " << forwardingHint
- << ", interest: " << interest);
+ _LOG_DEBUG(">> content server serving FILE, hint: " << forwardingHint << ", interest: " << interest);
- m_scheduler->scheduleOneTimeTask(m_scheduler, 0, bind(&ContentServer::serve_File_Execute, this,
- forwardingHint, name, interest),
- boost::lexical_cast<string>(name));
+ m_scheduler.scheduleEvent(time::seconds(0),
+ bind(&ContentServer::serve_File_Execute, this, forwardingHint, name,
+ interest));
// need to unlock ccnx mutex... or at least don't lock it
}
@@ -172,17 +158,16 @@
// interest: /<forwarding-hint>/<device_name>/<appname>/file/<hash>/<segment>
// name: /<device_name>/<appname>/file/<hash>/<segment>
- int64_t segment = name.getCompFromBackAsInt(0);
- Name deviceName = name.getPartialName(0, name.size() - 4);
- Hash hash(head(name.getCompFromBack(1)), name.getCompFromBack(1).size());
+ int64_t segment = name.get(-1).toNumber();
+ Name deviceName = name.getSubName(0, name.size() - 4);
+ Buffer hash(name.get(-2).value(), name.get(-2).value_size());
- _LOG_DEBUG(" server FILE for device: " << deviceName << ", file_hash: " << hash.shortHash()
- << " segment: "
+ _LOG_DEBUG(" server FILE for device: " << deviceName << ", file_hash: " << toHex(hash) << " segment: "
<< segment);
- string hashStr = lexical_cast<string>(hash);
+ std::string hashStr = toHex(hash);
- ObjectDbPtr db;
+ shared_ptr<ObjectDb> db;
ScopedLock(m_dbCacheMutex);
{
@@ -191,39 +176,41 @@
db = it->second;
}
else {
- if (ObjectDb::DoesExist(m_dbFolder, deviceName,
+ if (ObjectDb::doesExist(m_dbFolder, deviceName,
hashStr)) // this is kind of overkill, as it counts available segments
{
- db = boost::make_shared<ObjectDb>(m_dbFolder, hashStr);
+ db = make_shared<ObjectDb>(m_dbFolder, hashStr);
m_dbCache.insert(make_pair(hash, db));
}
else {
- _LOG_ERROR("ObjectDd doesn't exist for device: " << deviceName << ", file_hash: "
- << hash.shortHash());
+ _LOG_ERROR(
+ "ObjectDd doesn't exist for device: " << deviceName << ", file_hash: " << toHex(hash));
}
}
}
if (db) {
- BytesPtr co = db->fetchSegment(deviceName, segment);
- if (co) {
+ shared_ptr<Data> data = db->fetchSegment(deviceName, segment);
+ if (data) {
if (forwardingHint.size() == 0) {
- _LOG_DEBUG(ParsedContentObject(*co).name());
- m_ccnx->putToCcnd(*co);
+ m_face.put(*data);
}
else {
- if (m_freshness > 0) {
- m_ccnx->publishData(interest, *co, m_freshness);
- }
- else {
- m_ccnx->publishData(interest, *co);
- }
+ shared_ptr<Data> outerData = make_shared<Data>();
+
+ outerData->setContent(data->wireEncode());
+ outerData->setFreshnessPeriod(time::seconds(m_freshness));
+ outerData->setName(interest);
+
+ m_keyChain.sign(*outerData, signingWithSha256());;
+ m_face.put(*outerData);
}
+ _LOG_DEBUG("Send File Data Done!");
}
else {
_LOG_ERROR("ObjectDd exists, but no segment " << segment << " for device: " << deviceName
<< ", file_hash: "
- << hash.shortHash());
+ << toHex(hash));
}
}
}
@@ -235,24 +222,24 @@
// interest: /<forwarding-hint>/<device_name>/<appname>/action/<shared-folder>/<action-seq>
// name for actions: /<device_name>/<appname>/action/<shared-folder>/<action-seq>
- int64_t seqno = name.getCompFromBackAsInt(0);
- Name deviceName = name.getPartialName(0, name.size() - 4);
+ int64_t seqno = name.get(-1).toNumber();
+ Name deviceName = name.getSubName(0, name.size() - 4);
_LOG_DEBUG(" server ACTION for device: " << deviceName << " and seqno: " << seqno);
- PcoPtr pco = m_actionLog->LookupActionPco(deviceName, seqno);
- if (pco) {
+ shared_ptr<Data> data = m_actionLog->LookupActionData(deviceName, seqno);
+ if (data) {
if (forwardingHint.size() == 0) {
- m_ccnx->putToCcnd(pco->buf());
+ m_keyChain.sign(*data);
+ m_face.put(*data);
}
else {
- const Bytes& content = pco->buf();
+ data->setName(interest);
if (m_freshness > 0) {
- m_ccnx->publishData(interest, content, m_freshness);
+ data->setFreshnessPeriod(time::seconds(m_freshness));
}
- else {
- m_ccnx->publishData(interest, content);
- }
+ m_keyChain.sign(*data);
+ m_face.put(*data);
}
}
else {
@@ -266,12 +253,18 @@
ScopedLock(m_dbCacheMutex);
DbCache::iterator it = m_dbCache.begin();
while (it != m_dbCache.end()) {
- ObjectDbPtr db = it->second;
- if (db->secondsSinceLastUse() >= DB_CACHE_LIFETIME) {
+ shared_ptr<ObjectDb> db = it->second;
+ if (time::steady_clock::now() >= time::seconds(DB_CACHE_LIFETIME) + db->getLastUsed()) {
m_dbCache.erase(it++);
}
else {
++it;
}
}
+
+ m_flushStateDbCacheEvent = m_scheduler.scheduleEvent(time::seconds(DB_CACHE_LIFETIME),
+ bind(&ContentServer::flushStaleDbCache, this));
}
+
+} // namespace chronoshare
+} // namespace ndn
diff --git a/src/content-server.hpp b/src/content-server.hpp
index c8676a1..ccba737 100644
--- a/src/content-server.hpp
+++ b/src/content-server.hpp
@@ -1,6 +1,6 @@
/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */
/**
- * Copyright (c) 2013-2016, Regents of the University of California.
+ * Copyright (c) 2013-2017, Regents of the University of California.
*
* This file is part of ChronoShare, a decentralized file sharing application over NDN.
*
@@ -21,21 +21,31 @@
#ifndef CONTENT_SERVER_H
#define CONTENT_SERVER_H
+#include "core/chronoshare-common.hpp"
+
#include "action-log.hpp"
-#include "ccnx-wrapper.hpp"
#include "object-db.hpp"
-#include "scheduler.hpp"
-#include <boost/thread/locks.hpp>
-#include <boost/thread/shared_mutex.hpp>
+
+#include <ndn-cxx/face.hpp>
+#include <ndn-cxx/security/key-chain.hpp>
+#include <ndn-cxx/util/scheduler-scoped-event-id.hpp>
+#include <ndn-cxx/util/scheduler.hpp>
+
#include <map>
#include <set>
+#include <boost/thread/locks.hpp>
+#include <boost/thread/shared_mutex.hpp>
+
+namespace ndn {
+namespace chronoshare {
+
class ContentServer
{
public:
- ContentServer(Ccnx::CcnxWrapperPtr ccnx, ActionLogPtr actionLog,
- const boost::filesystem::path& rootDir, const Ccnx::Name& userName,
- const std::string& sharedFolderName, const std::string& appName, int freshness = -1);
+ ContentServer(Face& face, ActionLogPtr actionLog, const boost::filesystem::path& rootDir,
+ const Name& userName, const std::string& sharedFolderName,
+ const std::string& appName, KeyChain& keyChain, int freshness = -1);
~ContentServer();
// the assumption is, when the interest comes in, interest is informs of
@@ -43,54 +53,58 @@
// currently /topology-independent-name must begin with /action or /file
// so that ContentServer knows where to look for the content object
void
- registerPrefix(const Ccnx::Name& prefix);
+ registerPrefix(const Name& prefix);
void
- deregisterPrefix(const Ccnx::Name& prefix);
+ deregisterPrefix(const Name& prefix);
private:
void
- filterAndServe(Ccnx::Name forwardingHint, const Ccnx::Name& interest);
+ filterAndServe(const InterestFilter& forwardingHint, const Interest& interest);
void
- filterAndServeImpl(const Ccnx::Name& forwardingHint, const Ccnx::Name& name,
- const Ccnx::Name& interest);
+ filterAndServeImpl(const Name& forwardingHint, const Name& name, const Name& interest);
void
- serve_Action(const Ccnx::Name& forwardingHint, const Ccnx::Name& name, const Ccnx::Name& interest);
+ serve_Action(const Name& forwardingHint, const Name& name, const Name& interest);
void
- serve_File(const Ccnx::Name& forwardingHint, const Ccnx::Name& name, const Ccnx::Name& interest);
+ serve_File(const Name& forwardingHint, const Name& name, const Name& interest);
void
- serve_Action_Execute(const Ccnx::Name& forwardingHint, const Ccnx::Name& name,
- const Ccnx::Name& interest);
+ serve_Action_Execute(const Name& forwardingHint, const Name& name, const Name& interest);
void
- serve_File_Execute(const Ccnx::Name& forwardingHint, const Ccnx::Name& name,
- const Ccnx::Name& interest);
+ serve_File_Execute(const Name& forwardingHint, const Name& name, const Name& interest);
void
flushStaleDbCache();
private:
- Ndnx::NdnxWrapperPtr m_ndnx;
+ Face& m_face;
ActionLogPtr m_actionLog;
typedef boost::shared_mutex Mutex;
-
typedef boost::unique_lock<Mutex> ScopedLock;
- typedef std::set<Ndnx::Name>::iterator PrefixIt;
- std::set<Ndnx::Name> m_prefixes;
+
+ typedef std::map<Name, const RegisteredPrefixId*>::iterator FilterIdIt;
+ std::map<Name, const RegisteredPrefixId*> m_interestFilterIds;
+
Mutex m_mutex;
boost::filesystem::path m_dbFolder;
int m_freshness;
- SchedulerPtr m_scheduler;
- typedef std::map<Hash, ObjectDbPtr> DbCache;
+ Scheduler m_scheduler;
+ util::scheduler::ScopedEventId m_flushStateDbCacheEvent;
+ typedef std::map<Buffer, shared_ptr<ObjectDb>> DbCache;
DbCache m_dbCache;
Mutex m_dbCacheMutex;
- Ccnx::Name m_userName;
+ Name m_userName;
std::string m_sharedFolderName;
std::string m_appName;
+ KeyChain& m_keyChain;
};
+
+} // namespace chronoshare
+} // namespace ndn
+
#endif // CONTENT_SERVER_H
diff --git a/src/dispatcher.cpp b/src/dispatcher.cpp
index be7d3c4..3bc3bff 100644
--- a/src/dispatcher.cpp
+++ b/src/dispatcher.cpp
@@ -44,14 +44,16 @@
: m_ccnx(ccnx)
, m_core(NULL)
, m_rootDir(rootDir)
- , m_executor(
- 1) // creates problems with file assembly. need to ensure somehow that FinishExectute is called after all Segment_Execute finished
+ , m_executor(1)
+ // creates problems with file assembly. need to ensure somehow that
+ // FinishExectute is called after all Segment_Execute finished
, m_objectManager(ccnx, rootDir, CHRONOSHARE_APP)
, m_localUserName(localUserName)
, m_sharedFolder(sharedFolder)
, m_server(NULL)
, m_enablePrefixDiscovery(enablePrefixDiscovery)
{
+ KeyChain keyChain;
m_syncLog = make_shared<SyncLog>(m_rootDir, localUserName);
m_actionLog =
make_shared<ActionLog>(m_ccnx, m_rootDir, m_syncLog, sharedFolder, CHRONOSHARE_APP,
@@ -64,7 +66,7 @@
// m_server needs a different ccnx face
m_server = new ContentServer(make_shared<CcnxWrapper>(), m_actionLog, rootDir, m_localUserName,
- m_sharedFolder, CHRONOSHARE_APP, CONTENT_FRESHNESS);
+ m_sharedFolder, CHRONOSHARE_APP, keyChain, CONTENT_FRESHNESS);
m_server->registerPrefix(Name("/"));
m_server->registerPrefix(Name(BROADCAST_DOMAIN));
@@ -256,7 +258,6 @@
{
filesystem::path absolutePath = m_rootDir / relativeFilePath;
if (filesystem::exists(absolutePath)) {
- //BOOST_THROW_EXCEPTION (Error::Dispatcher() << error_info_str("Delete notification but file exists: " + absolutePath.string() ));
_LOG_ERROR("DELETE command, but file still exists: " << absolutePath.string());
return;
}
@@ -283,7 +284,8 @@
* - from ActionLog/AddOrUpdate: when action applied (file state changes, file added or modified) -> do nothing?
*
* - from FetchManager/Files: when file segment is retrieved -> save it in ObjectDb
- * when file fetch is completed -> if file belongs to FileState, then assemble it to filesystem. Don't do anything otherwise
+ * when file fetch is completed -> if file belongs to FileState, then assemble it to filesystem.
+ * Don't do anything otherwise
*/
void
@@ -355,7 +357,8 @@
FetchManager::PRIORITY_NORMAL);
}
}
- // if necessary (when version number is the highest) delete will be applied through the trigger in m_actionLog->AddRemoteAction call
+ // if necessary (when version number is the highest) delete will be
+ // applied through the trigger in m_actionLog->AddRemoteAction call
}
void
diff --git a/wscript b/wscript
index 8ac2af3..552b26f 100644
--- a/wscript
+++ b/wscript
@@ -109,6 +109,7 @@
'src/action-log.cpp',
'src/object-*.cpp',
'src/fetch*.cpp',
+ 'src/content-server.cpp'
]),
use='core-objects adhoc NDN_CXX BOOST TINYXML SQLITE3',
includes="src",