Merge "Changing naming convention"
diff --git a/ccnx/ccnx-wrapper.cpp b/ccnx/ccnx-wrapper.cpp
index d1c4672..5a6cfd3 100644
--- a/ccnx/ccnx-wrapper.cpp
+++ b/ccnx/ccnx-wrapper.cpp
@@ -49,9 +49,7 @@
, m_connected (false)
, m_executor (new Executor(1))
{
- connectCcnd();
- m_thread = thread (&CcnxWrapper::ccnLoop, this);
- m_executor->start();
+ start ();
}
void
@@ -82,16 +80,36 @@
CcnxWrapper::~CcnxWrapper()
{
+ shutdown ();
+}
+
+void
+CcnxWrapper::start () // called automatically in constructor
+{
+ connectCcnd();
+ m_thread = thread (&CcnxWrapper::ccnLoop, this);
+ m_executor->start();
+}
+
+void
+CcnxWrapper::shutdown () // called in destructor, but can called manually
+{
+ m_executor->shutdown();
+
{
UniqueRecLock lock(m_mutex);
m_running = false;
}
- m_executor->shutdown();
+ _LOG_DEBUG ("+++++++++SHUTDOWN+++++++");
+ if (m_connected)
+ {
+ m_thread.join ();
- m_thread.join ();
- ccn_disconnect (m_handle);
- //ccn_destroy (&m_handle);
+ ccn_disconnect (m_handle);
+ //ccn_destroy (&m_handle);
+ m_connected = false;
+ }
}
void
@@ -186,6 +204,15 @@
Bytes
CcnxWrapper::createContentObject(const Name &name, const void *buf, size_t len, int freshness)
{
+ {
+ UniqueRecLock lock(m_mutex);
+ if (!m_running || !m_connected)
+ {
+ _LOG_TRACE ("<< not running or connected");
+ return Bytes ();
+ }
+ }
+
CcnxCharbufPtr ptr = name.toCcnxCharbuf();
ccn_charbuf *pname = ptr->getBuf();
ccn_charbuf *content = ccn_charbuf_create();
diff --git a/ccnx/ccnx-wrapper.h b/ccnx/ccnx-wrapper.h
index 32d180c..b264d14 100644
--- a/ccnx/ccnx-wrapper.h
+++ b/ccnx/ccnx-wrapper.h
@@ -47,6 +47,15 @@
CcnxWrapper();
virtual ~CcnxWrapper();
+ void
+ start (); // called automatically in constructor
+
+ /**
+ * @brief Because of uncertainty with executor, in some case it is necessary to call shutdown explicitly (see test-server-and-fetch.cc)
+ */
+ void
+ shutdown (); // called in destructor, but can called manually
+
virtual int
setInterestFilter (const Name &prefix, const InterestCallback &interestCallback);
diff --git a/src/action-log.cc b/src/action-log.cc
index 7851c4c..85a691c 100644
--- a/src/action-log.cc
+++ b/src/action-log.cc
@@ -109,12 +109,13 @@
ActionLog::ActionLog (Ccnx::CcnxWrapperPtr ccnx, const boost::filesystem::path &path,
SyncLogPtr syncLog,
- const std::string &sharedFolder,
+ const std::string &sharedFolder, const std::string &appName,
OnFileAddedOrChangedCallback onFileAddedOrChanged, OnFileRemovedCallback onFileRemoved)
: DbHelper (path / ".chronoshare", "action-log.db")
, m_syncLog (syncLog)
, m_ccnx (ccnx)
, m_sharedFolderName (sharedFolder)
+ , m_appName (appName)
, m_onFileAddedOrChanged (onFileAddedOrChanged)
, m_onFileRemoved (onFileRemoved)
{
@@ -277,9 +278,12 @@
// assign name to the action, serialize action, and create content object
+
string item_msg;
item->SerializeToString (&item_msg);
- Name actionName = Name (m_syncLog->GetLocalName ())("action")(m_sharedFolderName)(seq_no);
+
+ // action name: /<appname>/<shared-folder>/action/<device_name>/<action-seq>
+ Name actionName = Name ("/")(m_appName)(m_sharedFolderName)("action")(m_syncLog->GetLocalName ())(seq_no);
_LOG_DEBUG ("ActionName: " << actionName);
Bytes actionData = m_ccnx->createContentObject (actionName, item_msg.c_str (), item_msg.size ());
@@ -379,7 +383,10 @@
string item_msg;
item->SerializeToString (&item_msg);
- Name actionName = Name (m_syncLog->GetLocalName ())("action")(m_sharedFolderName)(seq_no);
+
+ // action name: /<appname>/<shared-folder>/action/<device_name>/<action-seq>
+ Name actionName = Name ("/")(m_appName)(m_sharedFolderName)("action")(m_syncLog->GetLocalName ())(seq_no);
+ _LOG_DEBUG ("ActionName: " << actionName);
Bytes actionData = m_ccnx->createContentObject (actionName, item_msg.c_str (), item_msg.size ());
CcnxCharbufPtr namePtr = actionName.toCcnxCharbuf ();
@@ -556,23 +563,23 @@
ActionLog::AddRemoteAction (Ccnx::PcoPtr actionPco)
{
Name name = actionPco->name ();
- // <device_name>/"action"/<shared_folder_name_one_component>/<seqno>
+ // action name: /<appname>/<shared-folder>/action/<device_name>/<action-seq>
uint64_t seqno = name.getCompFromBackAsInt (0);
- string sharedFolder = name.getCompFromBackAsString (1);
+ string sharedFolder = name.getCompAsString (1);
if (sharedFolder != m_sharedFolderName)
{
BOOST_THROW_EXCEPTION (Error::ActionLog () << errmsg_info_str ("Action doesn't belong to this shared folder"));
}
- string action = name.getCompFromBackAsString (2);
+ string action = name.getCompAsString (2);
if (action != "action")
{
BOOST_THROW_EXCEPTION (Error::ActionLog () << errmsg_info_str ("not an action"));
}
- Name deviceName = name.getPartialName (0, name.size ()-3);
+ Name deviceName = name.getPartialName (3, name.size ()-4);
_LOG_DEBUG ("From [" << name << "] extracted deviceName: " << deviceName << ", sharedFolder: " << sharedFolder << ", seqno: " << seqno);
diff --git a/src/action-log.h b/src/action-log.h
index 9c6fdd6..b175001 100644
--- a/src/action-log.h
+++ b/src/action-log.h
@@ -48,7 +48,7 @@
public:
ActionLog (Ccnx::CcnxWrapperPtr ccnx, const boost::filesystem::path &path,
SyncLogPtr syncLog,
- const std::string &sharedFolder,
+ const std::string &sharedFolder, const std::string &appName,
OnFileAddedOrChangedCallback onFileAddedOrChanged, OnFileRemovedCallback onFileRemoved);
virtual ~ActionLog () { }
@@ -143,6 +143,7 @@
Ccnx::CcnxWrapperPtr m_ccnx;
std::string m_sharedFolderName;
+ std::string m_appName;
OnFileAddedOrChangedCallback m_onFileAddedOrChanged;
OnFileRemovedCallback m_onFileRemoved;
diff --git a/src/content-server.cc b/src/content-server.cc
index e960e14..66bf26b 100644
--- a/src/content-server.cc
+++ b/src/content-server.cc
@@ -32,6 +32,7 @@
ContentServer::ContentServer(CcnxWrapperPtr ccnx, ActionLogPtr actionLog,
const boost::filesystem::path &rootDir,
const Ccnx::Name &deviceName, const std::string &sharedFolderName,
+ const std::string &appName,
int freshness)
: m_ccnx(ccnx)
, m_actionLog(actionLog)
@@ -40,6 +41,7 @@
, m_executor (1)
, m_deviceName (deviceName)
, m_sharedFolderName (sharedFolderName)
+ , m_appName (appName)
{
m_executor.start ();
}
@@ -51,79 +53,104 @@
ScopedLock lock (m_mutex);
for (PrefixIt it = m_prefixes.begin(); it != m_prefixes.end(); ++it)
{
- m_ccnx->clearInterestFilter (*it);
+ Name filePrefix = Name (*it)(m_appName)("file");
+ Name actionPrefix = Name (*it)(m_appName)(m_sharedFolderName)("action");
+
+ m_ccnx->clearInterestFilter(filePrefix);
+ m_ccnx->clearInterestFilter(actionPrefix);
}
+
+ m_prefixes.clear ();
}
void
-ContentServer::registerPrefix(const Name &prefix)
+ContentServer::registerPrefix(const Name &forwardingHint)
{
- m_ccnx->setInterestFilter (prefix, bind(&ContentServer::serve, this, prefix, _1));
+ // Format for files: /<forwarding-hint>/<appname>/file/<hash>/<device_name>/<segment>
+ // Format for actions: /<forwarding-hint>/<appname>/<shared-folder>/action/<device_name>/<action-seq>
- _LOG_DEBUG (">> content server: register " << prefix);
+ Name filePrefix = Name (forwardingHint)(m_appName)("file");
+ Name actionPrefix = Name (forwardingHint)(m_appName)(m_sharedFolderName)("action");
+
+ m_ccnx->setInterestFilter (filePrefix, bind(&ContentServer::serve_File, this, forwardingHint, filePrefix, _1));
+ m_ccnx->setInterestFilter (actionPrefix, bind(&ContentServer::serve_Action, this, forwardingHint, actionPrefix, _1));
+
+ _LOG_DEBUG (">> content server: register FILE " << filePrefix);
+ _LOG_DEBUG (">> content server: register ACTION " << actionPrefix);
ScopedLock lock (m_mutex);
- m_prefixes.insert(prefix);
+ m_prefixes.insert(forwardingHint);
}
void
-ContentServer::deregisterPrefix (const Name &prefix)
+ContentServer::deregisterPrefix (const Name &forwardingHint)
{
+ Name filePrefix = Name (forwardingHint)(m_appName)("file");
+ Name actionPrefix = Name (forwardingHint)(m_appName)(m_sharedFolderName)("action");
- m_ccnx->clearInterestFilter(prefix);
+ m_ccnx->clearInterestFilter(filePrefix);
+ m_ccnx->clearInterestFilter(actionPrefix);
- _LOG_DEBUG ("<< content server: deregister " << prefix);
+ _LOG_DEBUG ("<< content server: deregister FILE " << filePrefix);
+ _LOG_DEBUG ("<< content server: deregister ACTION " << actionPrefix);
+
ScopedLock lock (m_mutex);
- m_prefixes.erase (prefix);
+ m_prefixes.erase (forwardingHint);
}
-void
-ContentServer::serve(Name forwardingHint, const Name &interest)
-{
- // /forwardingHint/device-name/action/shared-folder/action-seq
- // /forwardingHint/device-name/file/file-hash/segment
+// void
+// ContentServer::serve(Name forwardingHint, const Name &interest)
+// {
+// // /forwardingHint/app-name/device-name/action/shared-folder/action-seq
+// // /forwardingHint/app-name/device-name/file/file-hash/segment
- Name name = interest.getPartialName(forwardingHint.size());
- if (name.size() > 3)
- {
- string type = name.getCompAsString(name.size() - 3);
- if (type == "action")
- {
- serve_Action (forwardingHint, interest);
- }
- else if (type == "file")
- {
- serve_File (forwardingHint, interest);
- }
- }
-}
+// Name name = interest.getPartialName(forwardingHint.size());
+// if (name.size() > 3)
+// {
+// string type = name.getCompAsString(name.size() - 3);
+// if (type == "action")
+// {
+// serve_Action (forwardingHint, interest);
+// }
+// else if (type == "file")
+// {
+// serve_File (forwardingHint, interest);
+// }
+// }
+// }
void
-ContentServer::serve_Action (Name forwardingHint, const Name &interest)
+ContentServer::serve_Action (Name forwardingHint, Name locatorPrefix, Name interest)
{
- _LOG_DEBUG (">> content server serving, forwardHing " << forwardingHint << ", interest " << interest);
- m_executor.execute (bind (&ContentServer::serve_Action_Execute, this, forwardingHint, interest));
+ _LOG_DEBUG (">> content server serving ACTION, hint: " << forwardingHint << ", locatorPrefix: " << locatorPrefix << ", interest: " << interest);
+ m_executor.execute (bind (&ContentServer::serve_Action_Execute, this, forwardingHint, locatorPrefix, interest));
// need to unlock ccnx mutex... or at least don't lock it
}
void
-ContentServer::serve_File (Name forwardingHint, const Name &interest)
+ContentServer::serve_File (Name forwardingHint, Name locatorPrefix, Name interest)
{
- _LOG_DEBUG (">> content server serving, forwardHing " << forwardingHint << ", interest " << interest);
- m_executor.execute (bind (&ContentServer::serve_File_Execute, this, forwardingHint, interest));
+ _LOG_DEBUG (">> content server serving FILE, hint: " << forwardingHint << ", locatorPrefix: " << locatorPrefix << ", interest: " << interest);
+
+ m_executor.execute (bind (&ContentServer::serve_File_Execute, this, forwardingHint, locatorPrefix, interest));
// need to unlock ccnx mutex... or at least don't lock it
}
void
-ContentServer::serve_File_Execute (Name forwardingHint, Name interest)
+ContentServer::serve_File_Execute (Name forwardingHint, Name locatorPrefix, Name interest)
{
- // /device-name/file/<file-hash>/segment, or
+ // forwardingHint: /<forwarding-hint>
+ // locatorPrefix: /<forwarding-hint>/<appname>/file
+ // interest: /<forwarding-hint>/<appname>/file/<hash>/<device_name>/<segment>
- int64_t segment = interest.getCompFromBackAsInt (0);
- Name deviceName = interest.getPartialName (forwardingHint.size (), interest.size () - forwardingHint.size () - 3);
- Hash hash (head(interest.getCompFromBack (1)), interest.getCompFromBack (1).size());
+ Name pureInterest = interest.getPartialName (locatorPrefix.size ());
+ // pureInterest: /<hash>/<device_name>/<segment>
- _LOG_DEBUG (" server FILE for device: " << deviceName << ", file_hash: " << hash << " segment: " << segment);
+ int64_t segment = pureInterest.getCompFromBackAsInt (0);
+ Name deviceName = pureInterest.getPartialName (1, pureInterest.size () - 2);
+ Hash hash (head(pureInterest.getComp (0)), pureInterest.getComp (0).size());
+
+ _LOG_DEBUG (" server FILE for device: " << deviceName << ", file_hash: " << hash.shortHash () << " segment: " << segment);
string hashStr = lexical_cast<string> (hash);
if (ObjectDb::DoesExist (m_dbFolder, deviceName, hashStr)) // this is kind of overkill, as it counts available segments
@@ -136,6 +163,7 @@
{
if (forwardingHint.size () == 0)
{
+ _LOG_DEBUG (ParsedContentObject (*co).name ());
m_ccnx->putToCcnd (*co);
}
else
@@ -153,23 +181,28 @@
}
else
{
- _LOG_ERROR ("ObjectDd exists, but no segment " << segment << " for device: " << deviceName << ", file_hash: " << hash);
+ _LOG_ERROR ("ObjectDd exists, but no segment " << segment << " for device: " << deviceName << ", file_hash: " << hash.shortHash ());
}
}
else
{
- _LOG_ERROR ("ObjectDd doesn't exist for device: " << deviceName << ", file_hash: " << hash);
+ _LOG_ERROR ("ObjectDd doesn't exist for device: " << deviceName << ", file_hash: " << hash.shortHash ());
}
}
void
-ContentServer::serve_Action_Execute (Name forwardingHint, Name interest)
+ContentServer::serve_Action_Execute (Name forwardingHint, Name locatorPrefix, Name interest)
{
- // /device-name/action/shared-folder/seq
+ // forwardingHint: /<forwarding-hint>
+ // locatorPrefix: /<forwarding-hint>/<appname>/<shared-folder>/action
+ // interest: /<forwarding-hint>/<appname>/<shared-folder>/action/<device_name>/<action-seq>
- int64_t seqno = interest.getCompFromBackAsInt (0);
- Name deviceName = interest.getPartialName (forwardingHint.size (), interest.size () - forwardingHint.size () - 3);
+ Name pureInterest = interest.getPartialName (locatorPrefix.size ());
+ // pureInterest: /<device_name>/<action-seq>
+
+ int64_t seqno = pureInterest.getCompFromBackAsInt (0);
+ Name deviceName = pureInterest.getPartialName (0, pureInterest.size () - 1);
_LOG_DEBUG (" server ACTION for device: " << deviceName << " and seqno: " << seqno);
diff --git a/src/content-server.h b/src/content-server.h
index 47640d9..c4b70c6 100644
--- a/src/content-server.h
+++ b/src/content-server.h
@@ -34,7 +34,8 @@
{
public:
ContentServer(Ccnx::CcnxWrapperPtr ccnx, ActionLogPtr actionLog, const boost::filesystem::path &rootDir,
- const Ccnx::Name &deviceName, const std::string &sharedFolderName, int freshness = -1);
+ const Ccnx::Name &deviceName, const std::string &sharedFolderName, const std::string &appName,
+ int freshness = -1);
~ContentServer();
// the assumption is, when the interest comes in, interest is informs of
@@ -45,20 +46,20 @@
void deregisterPrefix(const Ccnx::Name &prefix);
private:
- void
- serve (Ccnx::Name forwardingHint, const Ccnx::Name &interest);
+ // void
+ // serve (Ccnx::Name forwardingHint, const Ccnx::Name &interest);
void
- serve_Action (Ccnx::Name forwardingHint, const Ccnx::Name &interest);
+ serve_Action (Ccnx::Name forwardingHint, Ccnx::Name locatorPrefix, Ccnx::Name interest);
void
- serve_File (Ccnx::Name forwardingHint, const Ccnx::Name &interest);
+ serve_File (Ccnx::Name forwardingHint, Ccnx::Name locatorPrefix, Ccnx::Name interest);
void
- serve_Action_Execute(Ccnx::Name forwardingHint, Ccnx::Name interest);
+ serve_Action_Execute(Ccnx::Name forwardingHint, Ccnx::Name locatorPrefix, Ccnx::Name interest);
void
- serve_File_Execute(Ccnx::Name forwardingHint, Ccnx::Name interest);
+ serve_File_Execute(Ccnx::Name forwardingHint, Ccnx::Name locatorPrefix, Ccnx::Name interest);
private:
Ccnx::CcnxWrapperPtr m_ccnx;
@@ -76,5 +77,6 @@
Ccnx::Name m_deviceName;
std::string m_sharedFolderName;
+ std::string m_appName;
};
#endif // CONTENT_SERVER_H
diff --git a/src/dispatcher.cc b/src/dispatcher.cc
index 9c611cb..22fbdbd 100644
--- a/src/dispatcher.cc
+++ b/src/dispatcher.cc
@@ -33,7 +33,9 @@
INIT_LOGGER ("Dispatcher");
-static const string BROADCAST_DOMAIN = "/ndn/broadcast/chronoshare";
+static const string CHRONOSHARE_APP = "chronoshare";
+static const string BROADCAST_DOMAIN = "/ndn/broadcast";
+
static const int CONTENT_FRESHNESS = 1800; // seconds
const static double DEFAULT_SYNC_INTEREST_INTERVAL = 10.0; // seconds;
@@ -47,36 +49,36 @@
, 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_objectManager(ccnx, rootDir)
+ , m_objectManager(ccnx, rootDir, CHRONOSHARE_APP)
, m_localUserName(localUserName)
, m_sharedFolder(sharedFolder)
, m_server(NULL)
, m_enablePrefixDiscovery(enablePrefixDiscovery)
{
m_syncLog = make_shared<SyncLog>(m_rootDir, localUserName);
- m_actionLog = make_shared<ActionLog>(m_ccnx, m_rootDir, m_syncLog, sharedFolder,
+ m_actionLog = make_shared<ActionLog>(m_ccnx, m_rootDir, m_syncLog, sharedFolder, CHRONOSHARE_APP,
// bind (&Dispatcher::Did_ActionLog_ActionApply_AddOrModify, this, _1, _2, _3, _4, _5, _6, _7),
ActionLog::OnFileAddedOrChangedCallback (), // don't really need this callback
bind (&Dispatcher::Did_ActionLog_ActionApply_Delete, this, _1));
- Name syncPrefix = Name(BROADCAST_DOMAIN)(sharedFolder);
+ Name syncPrefix = Name(BROADCAST_DOMAIN)(CHRONOSHARE_APP)(sharedFolder);
// m_server needs a different ccnx face
- m_server = new ContentServer(make_shared<CcnxWrapper>(), m_actionLog, rootDir, m_localUserName, m_sharedFolder, CONTENT_FRESHNESS);
- m_server->registerPrefix(Name ("/"));
+ m_server = new ContentServer(make_shared<CcnxWrapper>(), m_actionLog, rootDir, m_localUserName, m_sharedFolder, CHRONOSHARE_APP, CONTENT_FRESHNESS);
+ m_server->registerPrefix(Name("/"));
m_server->registerPrefix(Name(BROADCAST_DOMAIN));
- m_core = new SyncCore (m_syncLog, localUserName, Name ("/"), syncPrefix,
+ m_core = new SyncCore (m_syncLog, localUserName, Name("/"), syncPrefix,
bind(&Dispatcher::Did_SyncLog_StateChange, this, _1), ccnx, DEFAULT_SYNC_INTEREST_INTERVAL);
FetchTaskDbPtr actionTaskDb = make_shared<FetchTaskDb>(m_rootDir, "action");
m_actionFetcher = make_shared<FetchManager> (m_ccnx, bind (&SyncLog::LookupLocator, &*m_syncLog, _1), 3,
- bind (&Dispatcher::Did_FetchManager_ActionFetch, this, _1, _2, _3, _4), FetchManager::FinishCallback(), actionTaskDb);
+ bind (&Dispatcher::Did_FetchManager_ActionFetch, this, _1, _2, _3, _4), FetchManager::FinishCallback(), actionTaskDb);
FetchTaskDbPtr fileTaskDb = make_shared<FetchTaskDb>(m_rootDir, "file");
- m_fileFetcher = make_shared<FetchManager> (m_ccnx, bind (&SyncLog::LookupLocator, &*m_syncLog, _1), 3,
- bind (&Dispatcher::Did_FetchManager_FileSegmentFetch, this, _1, _2, _3, _4),
- bind (&Dispatcher::Did_FetchManager_FileFetchComplete, this, _1, _2),
- fileTaskDb);
+ m_fileFetcher = make_shared<FetchManager> (m_ccnx, bind (&SyncLog::LookupLocator, &*m_syncLog, _1), 3,
+ bind (&Dispatcher::Did_FetchManager_FileSegmentFetch, this, _1, _2, _3, _4),
+ bind (&Dispatcher::Did_FetchManager_FileFetchComplete, this, _1, _2),
+ fileTaskDb);
if (m_enablePrefixDiscovery)
@@ -91,7 +93,7 @@
Dispatcher::~Dispatcher()
{
- // _LOG_DEBUG ("Enter destructor of dispatcher");
+ _LOG_DEBUG ("Enter destructor of dispatcher");
m_executor.shutdown ();
// _LOG_DEBUG (">>");
@@ -266,7 +268,7 @@
Name userName (reinterpret_cast<const unsigned char *> (state.name ().c_str ()), state.name ().size ());
// fetch actions with oldSeq + 1 to newSeq (inclusive)
- Name actionNameBase = Name(userName)("action")(m_sharedFolder);
+ Name actionNameBase = Name ("/")(CHRONOSHARE_APP)(m_sharedFolder)("action")(userName);
m_actionFetcher->Enqueue (userName, actionNameBase,
std::max<uint64_t> (oldSeq + 1, 1), newSeq, FetchManager::PRIORITY_HIGH);
@@ -288,7 +290,7 @@
{
Hash hash (action->file_hash ().c_str(), action->file_hash ().size ());
- Name fileNameBase = Name (deviceName)("file")(hash.GetHash (), hash.GetHashBytes ());
+ Name fileNameBase = Name ("/")(CHRONOSHARE_APP)("file")(hash.GetHash (), hash.GetHashBytes ())(deviceName);
string hashStr = lexical_cast<string> (hash);
if (ObjectDb::DoesExist (m_rootDir / ".chronoshare", deviceName, hashStr))
@@ -340,7 +342,7 @@
void
Dispatcher::Did_FetchManager_FileSegmentFetch_Execute (Ccnx::Name deviceName, Ccnx::Name fileSegmentBaseName, uint32_t segment, Ccnx::PcoPtr fileSegmentPco)
{
- const Bytes &hashBytes = fileSegmentBaseName.getCompFromBack (0);
+ const Bytes &hashBytes = fileSegmentBaseName.getComp (2);
Hash hash (head(hashBytes), hashBytes.size());
_LOG_DEBUG ("Received segment deviceName: " << deviceName << ", segmentBaseName: " << fileSegmentBaseName << ", segment: " << segment);
@@ -372,8 +374,9 @@
{
_LOG_DEBUG ("Finished fetching " << deviceName << ", fileBaseName: " << fileBaseName);
- const Bytes &hashBytes = fileBaseName.getCompFromBack (0);
+ const Bytes &hashBytes = fileBaseName.getComp (2);
Hash hash (head (hashBytes), hashBytes.size ());
+ _LOG_DEBUG ("Extracted hash: " << hash);
if (m_objectDbMap.find (hash) != m_objectDbMap.end())
{
diff --git a/src/fetch-manager.cc b/src/fetch-manager.cc
index 7e087ce..d152c30 100644
--- a/src/fetch-manager.cc
+++ b/src/fetch-manager.cc
@@ -72,9 +72,10 @@
FetchManager::~FetchManager ()
{
m_scheduler->shutdown ();
-
m_executor->shutdown();
+ m_ccnx.reset ();
+
m_fetchList.clear_and_dispose (fetcher_disposer ());
}
@@ -101,7 +102,10 @@
Name forwardingHint;
forwardingHint = m_mapping (deviceName);
- m_taskDb->addTask(deviceName, baseName, minSeqNo, maxSeqNo, priority);
+ if (m_taskDb)
+ {
+ m_taskDb->addTask(deviceName, baseName, minSeqNo, maxSeqNo, priority);
+ }
unique_lock<mutex> lock (m_parellelFetchMutex);
@@ -221,7 +225,11 @@
m_currentParallelFetches --;
_LOG_TRACE ("+++++ removing fetcher: " << fetcher.GetName ());
m_fetchList.erase_and_dispose (FetchList::s_iterator_to (fetcher), fetcher_disposer ());
- m_taskDb->deleteTask(deviceName, baseName);
+
+ if (m_taskDb)
+ {
+ m_taskDb->deleteTask(deviceName, baseName);
+ }
}
m_scheduler->rescheduleTaskAt (m_scheduleFetchesTask, 0);
diff --git a/src/fetcher.cc b/src/fetcher.cc
index 044c5ca..3334040 100644
--- a/src/fetcher.cc
+++ b/src/fetcher.cc
@@ -199,7 +199,10 @@
// tell FetchManager that we have finish our job
// m_onFetchComplete (*this);
// using executor, so we won't be deleted if there is scheduled FillPipeline call
- m_executor->execute (bind (m_onFetchComplete, ref(*this), m_deviceName, m_name));
+ if (!m_onFetchComplete.empty ())
+ {
+ m_executor->execute (bind (m_onFetchComplete, ref(*this), m_deviceName, m_name));
+ }
}
else
{
@@ -210,6 +213,7 @@
void
Fetcher::OnTimeout (uint64_t seqno, const Ccnx::Name &name, const Closure &closure, Selectors selectors)
{
+ _LOG_DEBUG (this << ", " << m_executor.get ());
m_executor->execute (bind (&Fetcher::OnTimeout_Execute, this, seqno, name, closure, selectors));
}
@@ -247,7 +251,10 @@
}
m_active = false;
- m_onFetchFailed (ref (*this));
+ if (!m_onFetchFailed.empty ())
+ {
+ m_onFetchFailed (ref (*this));
+ }
// this is not valid anymore, but we still should be able finish work
}
}
diff --git a/src/object-manager.cc b/src/object-manager.cc
index ee24c7f..7b02e16 100644
--- a/src/object-manager.cc
+++ b/src/object-manager.cc
@@ -42,9 +42,10 @@
const int MAX_FILE_SEGMENT_SIZE = 1024;
-ObjectManager::ObjectManager (Ccnx::CcnxWrapperPtr ccnx, const fs::path &folder)
+ObjectManager::ObjectManager (Ccnx::CcnxWrapperPtr ccnx, const fs::path &folder, const std::string &appName)
: m_ccnx (ccnx)
, m_folder (folder / ".chronoshare")
+ , m_appName (appName)
{
fs::create_directories (m_folder);
}
@@ -53,6 +54,7 @@
{
}
+// /<appname>/file/<hash>/<devicename>/<segment>
boost::tuple<HashPtr /*object-db name*/, size_t /* number of segments*/>
ObjectManager::localFileToObjects (const fs::path &file, const Ccnx::Name &deviceName)
{
@@ -71,7 +73,7 @@
break;
}
- Name name = Name (deviceName)("file")(fileHash->GetHash (), fileHash->GetHashBytes ())(segment);
+ Name name = Name ("/")(m_appName)("file")(fileHash->GetHash (), fileHash->GetHashBytes ())(deviceName)(segment);
// cout << *fileHash << endl;
// cout << name << endl;
@@ -84,7 +86,7 @@
}
if (segment == 0) // handle empty files
{
- Name name = Name (deviceName)("file")(fileHash->GetHash (), fileHash->GetHashBytes ())(0);
+ Name name = Name ("/")(m_appName)("file")(fileHash->GetHash (), fileHash->GetHashBytes ())(deviceName)(0);
Bytes data = m_ccnx->createContentObject (name, 0, 0);
fileDb.saveContentObject (deviceName, 0, data);
diff --git a/src/object-manager.h b/src/object-manager.h
index ff26f30..2d0cdd0 100644
--- a/src/object-manager.h
+++ b/src/object-manager.h
@@ -33,9 +33,14 @@
class ObjectManager
{
public:
- ObjectManager (Ccnx::CcnxWrapperPtr ccnx, const boost::filesystem::path &folder);
+ ObjectManager (Ccnx::CcnxWrapperPtr ccnx, const boost::filesystem::path &folder, const std::string &appName);
virtual ~ObjectManager ();
+ /**
+ * @brief Creates and saves local file in a local database file
+ *
+ * Format: /<appname>/file/<hash>/<devicename>/<segment>
+ */
boost::tuple<HashPtr /*object-db name*/, size_t /* number of segments*/>
localFileToObjects (const boost::filesystem::path &file, const Ccnx::Name &deviceName);
@@ -45,6 +50,7 @@
private:
Ccnx::CcnxWrapperPtr m_ccnx;
boost::filesystem::path m_folder;
+ std::string m_appName;
};
typedef boost::shared_ptr<ObjectManager> ObjectManagerPtr;
diff --git a/src/sync-core.cc b/src/sync-core.cc
index f744f26..e6a4b01 100644
--- a/src/sync-core.cc
+++ b/src/sync-core.cc
@@ -69,7 +69,7 @@
SyncCore::~SyncCore()
{
- // m_scheduler->shutdown ();
+ m_scheduler->shutdown ();
// need to "deregister" closures
}
diff --git a/src/sync-log.cc b/src/sync-log.cc
index 06eb1e7..3ce8cfe 100644
--- a/src/sync-log.cc
+++ b/src/sync-log.cc
@@ -157,7 +157,7 @@
<< errmsg_info_str ("Impossible thing in SyncLog::GetNextLocalSeqNo"));
}
- _LOG_DEBUG_COND (sqlite3_errcode (m_db) != SQLITE_OK, sqlite3_errmsg (m_db));
+ _LOG_DEBUG_COND (sqlite3_errcode (m_db) != SQLITE_DONE, sqlite3_errmsg (m_db));
sqlite3_int64 seq_no = sqlite3_column_int64 (stmt_seq, 0) + 1;
sqlite3_finalize (stmt_seq);
@@ -205,7 +205,7 @@
res += sqlite3_bind_int64 (insertStmt, 1, rowId);
sqlite3_step (insertStmt);
- _LOG_DEBUG_COND (sqlite3_errcode (m_db) != SQLITE_OK, "DbError: " << sqlite3_errmsg (m_db));
+ _LOG_DEBUG_COND (sqlite3_errcode (m_db) != SQLITE_DONE, "DbError: " << sqlite3_errmsg (m_db));
if (res != SQLITE_OK)
{
sqlite3_exec (m_db, "ROLLBACK TRANSACTION;", 0,0,0);
@@ -231,7 +231,7 @@
{
sqlite3_exec (m_db, "ROLLBACK TRANSACTION;", 0,0,0);
- _LOG_DEBUG_COND (sqlite3_errcode (m_db) != SQLITE_OK, "DbError: " << sqlite3_errmsg (m_db));
+ _LOG_ERROR ("DbError: " << sqlite3_errmsg (m_db));
BOOST_THROW_EXCEPTION (Error::Db ()
<< errmsg_info_str ("Not a valid hash in rememberStateInStateLog"));
}
diff --git a/test/test-action-log.cc b/test/test-action-log.cc
index 84c0cf9..1a1ff1f 100644
--- a/test/test-action-log.cc
+++ b/test/test-action-log.cc
@@ -47,7 +47,7 @@
SyncLogPtr syncLog = make_shared<SyncLog> (tmpdir, localName);
CcnxWrapperPtr ccnx = make_shared<CcnxWrapper> ();
- ActionLogPtr actionLog = make_shared<ActionLog> (ccnx, tmpdir, syncLog, "top-secret",
+ ActionLogPtr actionLog = make_shared<ActionLog> (ccnx, tmpdir, syncLog, "top-secret", "test-chronoshare",
ActionLog::OnFileAddedOrChangedCallback(), ActionLog::OnFileRemovedCallback ());
// const std::string &filename,
@@ -77,24 +77,27 @@
pco = actionLog->LookupActionPco (localName, 1);
BOOST_CHECK_EQUAL ((bool)pco, true);
- BOOST_CHECK_EQUAL (pco->name (), "/alex/action/top-secret/%00%01");
+ BOOST_CHECK_EQUAL (pco->name (), "/test-chronoshare/top-secret/action/alex/%00%01");
- ActionItemPtr action = actionLog->LookupAction (Name ("/alex/action/top-secret")(0));
+ ActionItemPtr action = actionLog->LookupAction (Name ("/test-chronoshare/top-secret/action/alex")(0));
BOOST_CHECK_EQUAL ((bool)action, false);
- action = actionLog->LookupAction (Name ("/alex/action/top-secret")(1));
+ action = actionLog->LookupAction (Name ("/test-chronoshare/top-secret/action/alex")(1));
BOOST_CHECK_EQUAL ((bool)action, true);
- BOOST_CHECK_EQUAL (action->version (), 0);
- BOOST_CHECK_EQUAL (action->action (), 0);
+ if (action)
+ {
+ BOOST_CHECK_EQUAL (action->version (), 0);
+ BOOST_CHECK_EQUAL (action->action (), 0);
- BOOST_CHECK_EQUAL (action->filename (), "file.txt");
- BOOST_CHECK_EQUAL (action->seg_num (), 10);
- BOOST_CHECK_EQUAL (action->file_hash ().size (), 32);
- BOOST_CHECK_EQUAL (action->mode (), 0755);
+ BOOST_CHECK_EQUAL (action->filename (), "file.txt");
+ BOOST_CHECK_EQUAL (action->seg_num (), 10);
+ BOOST_CHECK_EQUAL (action->file_hash ().size (), 32);
+ BOOST_CHECK_EQUAL (action->mode (), 0755);
- BOOST_CHECK_EQUAL (action->has_parent_device_name (), false);
- BOOST_CHECK_EQUAL (action->has_parent_seq_no (), false);
+ BOOST_CHECK_EQUAL (action->has_parent_device_name (), false);
+ BOOST_CHECK_EQUAL (action->has_parent_seq_no (), false);
+ }
actionLog->AddLocalActionUpdate ("file.txt", *Hash::FromString ("2ff304769cdb0125ac039e6fe7575f8576dceffc62618a431715aaf6eea2bf1c"),
time (NULL), 0755, 10);
@@ -105,11 +108,14 @@
action = actionLog->LookupAction (Name ("/alex"), 2);
BOOST_CHECK_EQUAL ((bool)action, true);
- BOOST_CHECK_EQUAL (action->has_parent_device_name (), true);
- BOOST_CHECK_EQUAL (action->has_parent_seq_no (), true);
+ if (action)
+ {
+ BOOST_CHECK_EQUAL (action->has_parent_device_name (), true);
+ BOOST_CHECK_EQUAL (action->has_parent_seq_no (), true);
- BOOST_CHECK_EQUAL (action->parent_seq_no (), 1);
- BOOST_CHECK_EQUAL (action->version (), 1);
+ BOOST_CHECK_EQUAL (action->parent_seq_no (), 1);
+ BOOST_CHECK_EQUAL (action->version (), 1);
+ }
BOOST_CHECK_NO_THROW (actionLog->AddRemoteAction (pco));
BOOST_CHECK_EQUAL (actionLog->LogSize (), 2);
@@ -122,7 +128,7 @@
item.set_timestamp (time (NULL));
BytesPtr item_msg = serializeMsg (item);
- Name actionName = Name ("/zhenkai")("action")("top-secret")(1);
+ Name actionName = Name ("/")("test-chronoshare")("top-secret")("action")("zhenkai")(1);
Bytes actionData = ccnx->createContentObject (actionName, head (*item_msg), item_msg->size ());
pco = make_shared<ParsedContentObject> (actionData);
diff --git a/test/test-executor.cc b/test/test-executor.cc
index 9bdd43a..a13f373 100644
--- a/test/test-executor.cc
+++ b/test/test-executor.cc
@@ -43,12 +43,13 @@
{
Executor executor (3);
+ executor.start ();
Executor::Job job = bind(timeConsumingJob);
executor.execute(job);
executor.execute(job);
- usleep(1000);
+ usleep(2000);
// both jobs should have been taken care of
BOOST_CHECK_EQUAL(executor.jobQueueSize(), 0);
@@ -72,6 +73,8 @@
// all jobs should have been fetched
usleep(501000);
BOOST_CHECK_EQUAL(executor.jobQueueSize(), 0);
+
+ executor.shutdown ();
} //separate scope to ensure that destructor is called
diff --git a/test/test-fetch-manager.cc b/test/test-fetch-manager.cc
index 24ca59c..5abe47d 100644
--- a/test/test-fetch-manager.cc
+++ b/test/test-fetch-manager.cc
@@ -24,6 +24,9 @@
#include "ccnx-wrapper.h"
#include <boost/test/unit_test.hpp>
#include <boost/make_shared.hpp>
+#include "logging.h"
+
+INIT_LOGGER ("Test.FetchManager");
using namespace Ccnx;
using namespace std;
@@ -51,6 +54,8 @@
void
onData (const Ccnx::Name &deviceName, const Ccnx::Name &basename, uint64_t seqno, Ccnx::PcoPtr pco)
{
+ _LOG_TRACE ("onData: " << seqno);
+
recvData.insert (seqno);
differentNames.insert (basename);
Name name = basename;
@@ -90,6 +95,8 @@
BOOST_AUTO_TEST_CASE (TestFetcher)
{
+ INIT_LOGGERS ();
+
CcnxWrapperPtr ccnx = make_shared<CcnxWrapper> ();
Name baseName ("/base");
@@ -111,6 +118,7 @@
FetcherTestData data;
ExecutorPtr executor = make_shared<Executor>(1);
+ executor->start ();
Fetcher fetcher (ccnx,
executor,
@@ -176,6 +184,8 @@
BOOST_CHECK_EQUAL (recvData.str (), "0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, ");
BOOST_CHECK_EQUAL (recvData.str (), recvContent.str ());
}
+
+ executor->shutdown ();
}
// BOOST_AUTO_TEST_CASE (CcnxWrapperSelector)
diff --git a/test/test-object-manager.cc b/test/test-object-manager.cc
index 8f2ef2c..36c5c57 100644
--- a/test/test-object-manager.cc
+++ b/test/test-object-manager.cc
@@ -49,7 +49,7 @@
Name deviceName ("/device");
CcnxWrapperPtr ccnx = make_shared<CcnxWrapper> ();
- ObjectManager manager (ccnx, tmpdir);
+ ObjectManager manager (ccnx, tmpdir, "test-chronoshare");
tuple<HashPtr,int> hash_semgents = manager.localFileToObjects (fs::path("test") / "test-object-manager.cc", deviceName);
diff --git a/test/test-serve-and-fetch.cc b/test/test-serve-and-fetch.cc
index a25fa89..9d71cfb 100644
--- a/test/test-serve-and-fetch.cc
+++ b/test/test-serve-and-fetch.cc
@@ -36,6 +36,10 @@
#include <stdio.h>
#include <ctime>
+#include "logging.h"
+
+INIT_LOGGER("Test.ServerAndFetch");
+
using namespace Ccnx;
using namespace std;
using namespace boost;
@@ -113,47 +117,55 @@
BOOST_AUTO_TEST_CASE (TestServeAndFetch)
{
- cout << "Setting up test environment ..." << endl;
+ INIT_LOGGERS ();
+
+ _LOG_DEBUG ("Setting up test environment ...");
setup();
CcnxWrapperPtr ccnx_serve = make_shared<CcnxWrapper>();
usleep(1000);
CcnxWrapperPtr ccnx_fetch = make_shared<CcnxWrapper>();
- ObjectManager om(ccnx_serve, root);
Name deviceName("/test/device");
Name localPrefix("/local");
Name broadcastPrefix("/broadcast");
+ const string APPNAME = "test-chronoshare";
+
time_t start = time(NULL);
- cout << "At time " << start << ", publish local file to database, this is extremely slow ..." << endl;
+ _LOG_DEBUG ("At time " << start << ", publish local file to database, this is extremely slow ...");
// publish file to db
+ ObjectManager om(ccnx_serve, root, APPNAME);
tuple<HashPtr, size_t> pub = om.localFileToObjects(filePath, deviceName);
time_t end = time(NULL);
- cout << "At time " << end <<", publish finally finished, used " << end - start << " seconds ..."<< endl;
+ _LOG_DEBUG ("At time " << end <<", publish finally finished, used " << end - start << " seconds ...");
ActionLogPtr dummyLog;
- ContentServer server(ccnx_serve, dummyLog, root, deviceName, "pentagon's secrets", 5);
+ ContentServer server(ccnx_serve, dummyLog, root, deviceName, "pentagon's secrets", APPNAME, 5);
server.registerPrefix(localPrefix);
server.registerPrefix(broadcastPrefix);
FetchManager fm(ccnx_fetch, bind(simpleMap, _1));
HashPtr hash = pub.get<0> ();
- Name baseName = Name (deviceName)("file")(hash->GetHash(), hash->GetHashBytes());
+ Name baseName = Name ("/")(APPNAME)("file")(hash->GetHash(), hash->GetHashBytes())(deviceName);
+
fm.Enqueue(deviceName, baseName, bind(segmentCallback, _1, _2, _3, _4), bind(finishCallback, _1, _2), 0, pub.get<1>() - 1);
unique_lock<mutex> lock(mut);
system_time timeout = get_system_time() + posix_time::milliseconds(5000);
while (!finished)
- {
- if (!cond.timed_wait(lock, timeout))
{
- BOOST_FAIL("Fetching has not finished after 5 seconds");
- break;
+ if (!cond.timed_wait(lock, timeout))
+ {
+ BOOST_FAIL ("Fetching has not finished after 5 seconds");
+ break;
+ }
}
- }
+ ccnx_fetch->shutdown ();
+ ccnx_serve->shutdown ();
- cout << "ack : " << ack << endl;
+ _LOG_DEBUG ("Finish");
+ usleep(100000);
teardown();
}