get rid of publisher and fetcher
modify SyncAppSocket API
modify SyncLogic API
Tweak SeqNo
diff --git a/ccnx/sync-app-data-fetch.cc b/ccnx/sync-app-data-fetch.cc
deleted file mode 100644
index 5cfabce..0000000
--- a/ccnx/sync-app-data-fetch.cc
+++ /dev/null
@@ -1,67 +0,0 @@
-/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
-/*
- * Copyright (c) 2012 University of California, Los Angeles
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License version 2 as
- * published by the Free Software Foundation;
- *
- * This program 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 this program; if not, write to the Free Software
- * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
- *
- * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
- * Chaoyi Bian <bcy@pku.edu.cn>
- * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
- */
-
-#include "sync-app-data-fetch.h"
-#include "sync-log.h"
-
-using namespace std;
-using namespace boost;
-
-namespace Sync
-{
-
-INIT_LOGGER ("AppDataFetch");
-
-void
-AppDataFetch::onUpdate (const std::string &prefix, const SeqNo &newSeq, const SeqNo &oldSeq)
-{
- _LOG_FUNCTION (this << ", " << prefix << ", " << newSeq << ", " << oldSeq);
-
- // sequence number logic here
- uint32_t start = 0;
- if (oldSeq.isValid () && oldSeq.getSession() == newSeq.getSession())
- {
- start = oldSeq.getSeq () + 1;
- }
- uint32_t end = newSeq.getSeq ();
-
- //
- // add logic for wrap around
- //
-
- for (uint32_t i = start; i <= end; i++)
- {
- ostringstream interestName;
- interestName << prefix << "/" << newSeq.getSession () << "/" << i;
- m_ccnxHandle->sendInterest (interestName.str (), m_dataCallback);
- }
-}
-
-void
-AppDataFetch::onRemove (const std::string &prefix)
-{
- _LOG_FUNCTION (this << ", " << prefix);
-
- // I guess this should be somewhere in app
-}
-
-}
diff --git a/ccnx/sync-app-data-fetch.h b/ccnx/sync-app-data-fetch.h
deleted file mode 100644
index 1e42860..0000000
--- a/ccnx/sync-app-data-fetch.h
+++ /dev/null
@@ -1,86 +0,0 @@
-/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
-/*
- * Copyright (c) 2012 University of California, Los Angeles
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License version 2 as
- * published by the Free Software Foundation;
- *
- * This program 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 this program; if not, write to the Free Software
- * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
- *
- * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
- * Chaoyi Bian <bcy@pku.edu.cn>
- * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
- */
-
-#ifndef SYNC_APP_DATA_FETCH_H
-#define SYNC_APP_DATA_FETCH_H
-
-#include "sync-ccnx-wrapper.h"
-#include "sync-seq-no.h"
-
-namespace Sync {
-
-/**
- * \ingroup sync
- * @brief fetch application data, by default it will try to fetch every piece
- * of data
- */
-class AppDataFetch
-{
-public:
- /**
- * @brief Constructor
- * @param ccnxHandle handle for CCNx
- * @param dataCallback the callback function to process data
- */
- AppDataFetch (CcnxWrapperPtr ccnxHandle,
- CcnxWrapper::DataCallback dataCallback)
- : m_ccnxHandle (ccnxHandle)
- , m_dataCallback (dataCallback)
- { }
-
- /**
- * @brief Set data callback
- * @param dataCallback the callback function to process data
- */
- void
- setDataCallback(CcnxWrapper::DataCallback dataCallback)
- {
- m_dataCallback = dataCallback;
- }
-
- /**
- * @brief Fired from SyncLogic when new data is available
- *
- * @param prefix the prefix for the data
- * @param newSeq old session ID/sequence number
- * @param oldSeq new session ID/sequence number
- */
- void
- onUpdate (const std::string &prefix, const SeqNo &newSeq, const SeqNo &oldSeq);
-
- /**
- * @brief Fired from SyncLogic when data is removed
- *
- * @param prefix the prefix for the data
- */
- void
- onRemove (const std::string &prefix);
-
-private:
- CcnxWrapperPtr m_ccnxHandle;
- CcnxWrapper::DataCallback m_dataCallback;
-};
-
-
-} // Sync
-
-#endif // SYNC_APP_DATA_FETCH_H
diff --git a/ccnx/sync-app-data-publish.cc b/ccnx/sync-app-data-publish.cc
deleted file mode 100644
index e7bee97..0000000
--- a/ccnx/sync-app-data-publish.cc
+++ /dev/null
@@ -1,81 +0,0 @@
-/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
-/*
- * Copyright (c) 2012 University of California, Los Angeles
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License version 2 as
- * published by the Free Software Foundation;
- *
- * This program 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 this program; if not, write to the Free Software
- * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
- *
- * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
- * Chaoyi Bian <bcy@pku.edu.cn>
- * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
- */
-
-#include "sync-app-data-publish.h"
-#include <boost/throw_exception.hpp>
-typedef boost::error_info<struct tag_errmsg, std::string> errmsg_info_str;
-typedef boost::error_info<struct tag_errmsg, int> errmsg_info_int;
-
-using namespace std;
-using namespace boost;
-
-namespace Sync
-{
-
-
-string
-AppDataPublish::getRecentData (const string &prefix, uint32_t session)
-{
- if (m_recentData.find(make_pair(prefix, session)) != m_recentData.end())
- return m_recentData[make_pair(prefix, session)];
- else
- return "";
-}
-
-uint32_t
-AppDataPublish::getNextSeq (const string &prefix, uint32_t session)
-{
- SequenceLog::iterator i = m_sequenceLog.find (prefix);
-
- if (i != m_sequenceLog.end ())
- {
- Seq s = i->second;
- if (s.session == session)
- return s.seq;
- }
- return 0;
-}
-
-uint32_t
-AppDataPublish::publishData (const string &name, uint32_t session, const string &dataBuffer, int freshness)
-{
- uint32_t seq = getNextSeq (name, session);
-
- ostringstream contentNameWithSeqno;
- contentNameWithSeqno << name << "/" << session << "/" << seq;
-
- m_ccnxHandle->publishData (contentNameWithSeqno.str (), dataBuffer, freshness);
-
- Seq s;
- s.session = session;
- s.seq = seq + 1;
- m_sequenceLog[name] = s;
-
- unordered_map<pair<string, uint32_t>, string>::iterator it = m_recentData.find(make_pair(name, session));
- if (it != m_recentData.end())
- m_recentData.erase(it);
- m_recentData.insert(make_pair(make_pair(name, session), dataBuffer));
-
- return seq;
-}
-
-} // Sync
diff --git a/ccnx/sync-app-data-publish.h b/ccnx/sync-app-data-publish.h
deleted file mode 100644
index ec95a60..0000000
--- a/ccnx/sync-app-data-publish.h
+++ /dev/null
@@ -1,97 +0,0 @@
-/* -*- Mode: C++; c-file-style: "gnu"; indent-tabs-mode:nil -*- */
-/*
- * Copyright (c) 2012 University of California, Los Angeles
- *
- * This program is free software; you can redistribute it and/or modify
- * it under the terms of the GNU General Public License version 2 as
- * published by the Free Software Foundation;
- *
- * This program 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 this program; if not, write to the Free Software
- * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
- *
- * Author: Zhenkai Zhu <zhenkai@cs.ucla.edu>
- * Chaoyi Bian <bcy@pku.edu.cn>
- * Alexander Afanasyev <alexander.afanasyev@ucla.edu>
- */
-
-#ifndef SYNC_APP_DATA_PUBLISH_H
-#define SYNC_APP_DATA_PUBLISH_H
-
-#include <boost/unordered_map.hpp>
-#include "sync-seq-no.h"
-#include "sync-ccnx-wrapper.h"
-#include <utility>
-#include <map>
-
-namespace Sync {
-
-struct Seq
-{
- uint32_t session;
- uint32_t seq;
-};
-
-struct GetSeqException : virtual boost::exception, virtual std::exception { };
-
-/**
- * \ingroup sync
- * @brief publishes application data using incrementing sequence number (for
- * each sequence namber and keeps track of most recently published data for
- * each name prefix
- */
-class AppDataPublish
-{
-public:
- AppDataPublish (CcnxWrapperPtr ccnxHandle)
- : m_ccnxHandle (ccnxHandle)
- { }
-
- /**
- * @brief get the name (including sequence number) and the content
- * (unencoded, just XML stanza) of the most recent published data
- *
- * @param prefix the name prefix to look for
- * @param session session
- * @return the pair of name and content
- */
- std::string
- getRecentData (const std::string &prefix, uint32_t session);
-
- /**
- * @brief get the most recent sequence number for a name prefix
- * @param prefix the name prefix to look for
- * @param session session
- */
- uint32_t
- getNextSeq (const std::string &prefix, uint32_t session);
-
- /**
- * @brief publish data for a name prefix, updates the corresponding
- * sequence number and recent data
- *
- * @param name data prefix
- * @param session session to which data is published
- * @param dataBuffer the data itself
- * @param freshness the freshness for the data object
- * @return published sequence number, will throw an exception if something wrong happened
- */
- uint32_t publishData (const std::string &name, uint32_t session, const std::string &dataBuffer, int freshness);
-
-private:
- typedef boost::unordered_map<std::string, Seq> SequenceLog;
- typedef boost::unordered_map<std::pair<std::string, uint32_t>, std::string> RecentData;
-
- CcnxWrapperPtr m_ccnxHandle;
- SequenceLog m_sequenceLog;
- RecentData m_recentData;
-};
-
-} // Sync
-
-#endif // SYNC_APP_DATA_PUBLISH_H
diff --git a/ccnx/sync-app-socket.cc b/ccnx/sync-app-socket.cc
index 848a6d0..ce65674 100644
--- a/ccnx/sync-app-socket.cc
+++ b/ccnx/sync-app-socket.cc
@@ -28,13 +28,12 @@
namespace Sync
{
-SyncAppSocket::SyncAppSocket (const string &syncPrefix, CcnxWrapper::DataCallback dataCallback)
- : m_appHandle (new CcnxWrapper())
- , m_fetcher (m_appHandle, dataCallback)
- , m_publisher (m_appHandle)
+SyncAppSocket::SyncAppSocket (const string &syncPrefix, NewDataCallback dataCallback, RemoveCallback rmCallback )
+ : m_ccnxHandle (new CcnxWrapper())
+ , m_newDataCallback(dataCallback)
, m_syncLogic (syncPrefix,
- bind (&AppDataFetch::onUpdate, m_fetcher, _1, _2, _3),
- bind (&AppDataFetch::onRemove, m_fetcher, _1))
+ bind(&SyncAppSocket::passCallback, this, _1),
+ rmCallback)
{
}
@@ -42,10 +41,62 @@
{
}
-bool SyncAppSocket::publish (const string &prefix, uint32_t session, const string &dataBuffer, int freshness)
+bool
+SyncAppSocket::publishString (const string &prefix, uint32_t session, const string &dataBuffer, int freshness)
{
- uint32_t sequence = m_publisher.publishData (prefix, session, dataBuffer, freshness);
+ uint32_t sequence = getNextSeq(prefix, session);
+ ostringstream contentNameWithSeqno;
+ contentNameWithSeqno << prefix << "/" << session << "/" << sequence;
+ m_ccnxHandle->publishStringData (contentNameWithSeqno.str (), dataBuffer, freshness);
+
+ SeqNo s(session, sequence + 1);
+ m_sequenceLog[prefix] = s;
+
m_syncLogic.addLocalNames (prefix, session, sequence);
}
+bool
+SyncAppSocket::publishRaw(const std::string &prefix, uint32_t session, const char *buf, size_t len, int freshness)
+{
+ uint32_t sequence = getNextSeq(prefix, session);
+ ostringstream contentNameWithSeqno;
+ contentNameWithSeqno << prefix << "/" << session << "/" << sequence;
+ m_ccnxHandle->publishRawData (contentNameWithSeqno.str (), buf, len, freshness);
+
+ SeqNo s(session, sequence + 1);
+ m_sequenceLog[prefix] = s;
+ m_syncLogic.addLocalNames (prefix, session, sequence);
+}
+
+
+void
+SyncAppSocket::fetchString(const std::string &prefix, const SeqNo &seq, CcnxWrapper::StringDataCallback callback, int retry)
+{
+ ostringstream interestName;
+ interestName << prefix << "/" << seq.getSession() << "/" << seq.getSeq();
+ m_ccnxHandle->sendInterestForString(interestName.str(), callback, retry);
+}
+
+void
+SyncAppSocket::fetchRaw(const std::string &prefix, const SeqNo &seq, CcnxWrapper::RawDataCallback callback, int retry)
+{
+ ostringstream interestName;
+ interestName << prefix << "/" << seq.getSession() << "/" << seq.getSeq();
+ m_ccnxHandle->sendInterest(interestName.str(), callback, retry);
+}
+
+uint32_t
+SyncAppSocket::getNextSeq (const string &prefix, uint32_t session)
+{
+ SequenceLog::iterator i = m_sequenceLog.find (prefix);
+
+ if (i != m_sequenceLog.end ())
+ {
+ SeqNo s = i->second;
+ if (s.getSession() == session)
+ return s.getSeq();
+ }
+ return 0;
+}
+
}
diff --git a/ccnx/sync-app-socket.h b/ccnx/sync-app-socket.h
index f5e5e0f..07c0aa0 100644
--- a/ccnx/sync-app-socket.h
+++ b/ccnx/sync-app-socket.h
@@ -24,9 +24,14 @@
#define SYNC_APP_SOCKET_H
#include "sync-logic.h"
-#include "sync-app-data-fetch.h"
-#include "sync-app-data-publish.h"
#include <boost/function.hpp>
+#include <boost/unordered_map.hpp>
+#include "sync-seq-no.h"
+#include "sync-ccnx-wrapper.h"
+#include <utility>
+#include <map>
+#include <vector>
+#include <sstream>
namespace Sync {
@@ -37,6 +42,8 @@
class SyncAppSocket
{
public:
+ typedef boost::function< void (const std::vector<MissingDataInfo> &, SyncAppSocket * ) > NewDataCallback;
+ typedef boost::function< void ( const std::string &/*prefix*/ ) > RemoveCallback;
/**
* @brief the constructor for SyncAppSocket; the parameter syncPrefix
* should be passed to the constructor of m_syncAppWrapper; the other
@@ -48,7 +55,7 @@
* @param syncPrefix the name prefix for Sync Interest
* @param dataCallback the callback to process data
*/
- SyncAppSocket (const std::string &syncPrefix, CcnxWrapper::DataCallback dataCallback);
+ SyncAppSocket (const std::string &syncPrefix, NewDataCallback dataCallback, RemoveCallback rmCallback);
~SyncAppSocket ();
/**
@@ -60,7 +67,9 @@
* @param dataBuffer the data itself
* @param freshness the freshness time for the data (in seconds)
*/
- bool publish (const std::string &prefix, uint32_t session, const std::string &dataBuffer, int freshness);
+ bool publishString (const std::string &prefix, uint32_t session, const std::string &dataBuffer, int freshness);
+
+ bool publishRaw(const std::string &prefix, uint32_t session, const char *buf, size_t len, int freshness);
/**
* @brief delete a participant's subtree from the sync tree; SyncLogic will do the work
@@ -70,13 +79,20 @@
*/
void remove (const std::string &prefix) {m_syncLogic.remove(prefix);}
- int getSeq();
+ void fetchString(const std::string &prefix, const SeqNo &seq, CcnxWrapper::StringDataCallback callback, int retry = 0);
+ void fetchRaw(const std::string &prefix, const SeqNo &seq, CcnxWrapper::RawDataCallback callback, int retry = 0);
+
+ void passCallback(std::vector<MissingDataInfo> &v) {m_newDataCallback(v, this);}
private:
- CcnxWrapperPtr m_appHandle;
+ uint32_t
+ getNextSeq (const std::string &prefix, uint32_t session);
- AppDataFetch m_fetcher;
- AppDataPublish m_publisher;
+private:
+ typedef boost::unordered_map<std::string, SeqNo> SequenceLog;
+ NewDataCallback m_newDataCallback;
+ SequenceLog m_sequenceLog;
+ CcnxWrapperPtr m_ccnxHandle;
SyncLogic m_syncLogic;
};
diff --git a/ccnx/sync-ccnx-wrapper.cc b/ccnx/sync-ccnx-wrapper.cc
index 85ffbd6..559cfad 100644
--- a/ccnx/sync-ccnx-wrapper.cc
+++ b/ccnx/sync-ccnx-wrapper.cc
@@ -158,9 +158,13 @@
}
/// @endcond
+int
+CcnxWrapper::publishStringData (const string &name, const string &dataBuffer, int freshness) {
+ publishRawData(name, dataBuffer.c_str(), dataBuffer.length(), freshness);
+}
int
-CcnxWrapper::publishData (const string &name, const string &dataBuffer, int freshness)
+CcnxWrapper::publishRawData (const string &name, const char *buf, size_t len, int freshness)
{
recursive_mutex::scoped_lock lock(m_mutex);
if (!m_running)
@@ -181,7 +185,7 @@
NULL,
m_keyLoactor);
if(ccn_encode_ContentObject(content, pname, signed_info,
- dataBuffer.c_str(), dataBuffer.length (),
+ (const unsigned char *)buf, len,
NULL, getPrivateKey()) < 0)
BOOST_THROW_EXCEPTION(CcnxOperationException() << errmsg_info_str("encode content failed"));
@@ -235,7 +239,6 @@
ccn_upcall_kind kind,
ccn_upcall_info *info)
{
- //CcnxWrapper::DataCallback *f = static_cast<CcnxWrapper::DataCallback*> (selfp->data);
ClosurePass *cp = static_cast<ClosurePass *> (selfp->data);
switch (kind)
@@ -282,13 +285,13 @@
return CCN_UPCALL_RESULT_OK;
}
-int CcnxWrapper::sendInterest (const string &strInterest, const DataCallback &dataCallback, int retry)
+int CcnxWrapper::sendInterestForString (const string &strInterest, const StringDataCallback &strDataCallback, int retry)
{
- DataClosurePass * pass = new DataClosurePass(STRING_FORM, retry, dataCallback);
+ DataClosurePass * pass = new DataClosurePass(STRING_FORM, retry, strDataCallback);
sendInterest(strInterest, pass);
}
-int CcnxWrapper::sendInterestForRawData (const string &strInterest, const RawDataCallback &rawDataCallback, int retry)
+int CcnxWrapper::sendInterest (const string &strInterest, const RawDataCallback &rawDataCallback, int retry)
{
RawDataClosurePass * pass = new RawDataClosurePass(RAW_DATA, retry, rawDataCallback);
sendInterest(strInterest, pass);
@@ -356,9 +359,9 @@
ccn_charbuf_destroy(&pname);
}
-DataClosurePass::DataClosurePass (CallbackType type, int retry, const CcnxWrapper::DataCallback &dataCallback): ClosurePass(type, retry), m_callback(NULL)
+DataClosurePass::DataClosurePass (CallbackType type, int retry, const CcnxWrapper::StringDataCallback &strDataCallback): ClosurePass(type, retry), m_callback(NULL)
{
- m_callback = new CcnxWrapper::DataCallback (dataCallback);
+ m_callback = new CcnxWrapper::StringDataCallback (strDataCallback);
}
DataClosurePass::~DataClosurePass ()
diff --git a/ccnx/sync-ccnx-wrapper.h b/ccnx/sync-ccnx-wrapper.h
index b8c3c93..b51c629 100644
--- a/ccnx/sync-ccnx-wrapper.h
+++ b/ccnx/sync-ccnx-wrapper.h
@@ -37,6 +37,7 @@
#include <boost/thread/thread.hpp>
#include <boost/function.hpp>
#include <string>
+#include <sstream>
/**
* \defgroup sync SYNC protocol
@@ -53,7 +54,7 @@
*/
class CcnxWrapper {
public:
- typedef boost::function<void (std::string, std::string)> DataCallback;
+ typedef boost::function<void (std::string, std::string)> StringDataCallback;
typedef boost::function<void (std::string, const char *buf, size_t len)> RawDataCallback;
typedef boost::function<void (std::string)> InterestCallback;
@@ -78,10 +79,10 @@
* @return the return code of ccn_express_interest
*/
int
- sendInterest (const std::string &strInterest, const DataCallback &dataCallback, int retry = 0);
+ sendInterestForString (const std::string &strInterest, const StringDataCallback &strDataCallback, int retry = 0);
int
- sendInterestForRawData (const std::string &strInterest, const RawDataCallback &rawDataCallback, int retry = 0);
+ sendInterest (const std::string &strInterest, const RawDataCallback &rawDataCallback, int retry = 0);
/**
* @brief set Interest filter (specify what interest you want to receive)
@@ -110,7 +111,10 @@
* @return code generated by ccnx library calls, >0 if success
*/
int
- publishData (const std::string &name, const std::string &dataBuffer, int freshness);
+ publishStringData (const std::string &name, const std::string &dataBuffer, int freshness);
+
+ int
+ publishRawData (const std::string &name, const char *buf, size_t len, int freshness);
private:
/// @cond include_hidden
@@ -165,11 +169,11 @@
class DataClosurePass: public ClosurePass {
public:
- DataClosurePass(CallbackType type, int retry, const CcnxWrapper::DataCallback &dataCallback);
+ DataClosurePass(CallbackType type, int retry, const CcnxWrapper::StringDataCallback &strDataCallback);
virtual ~DataClosurePass();
virtual void runCallback(std::string name, const char *, size_t len);
private:
- CcnxWrapper::DataCallback * m_callback;
+ CcnxWrapper::StringDataCallback * m_callback;
};
class RawDataClosurePass: public ClosurePass {
diff --git a/model/sync-logic.cc b/model/sync-logic.cc
index 3bae752..b44c2e8 100644
--- a/model/sync-logic.cc
+++ b/model/sync-logic.cc
@@ -279,6 +279,7 @@
DiffState diff;
istringstream ss (dataBuffer);
ss >> diff;
+ vector<MissingDataInfo> v;
BOOST_FOREACH (LeafConstPtr leaf, diff.getLeaves().get<ordered>())
{
DiffLeafConstPtr diffLeaf = dynamic_pointer_cast<const DiffLeaf> (leaf);
@@ -297,7 +298,12 @@
if (inserted || updated)
{
diffLog->update (info, seq);
- m_onUpdate (info->toString (), seq, oldSeq);
+ //m_onUpdate (info->toString (), seq, oldSeq);
+ MissingDataInfo mdi;
+ mdi.prefix = info->toString();
+ mdi.low = oldSeq;
+ mdi.high = seq;
+ v.push_back(mdi);
}
}
else if (diffLeaf->getOperation() == REMOVE)
@@ -314,6 +320,11 @@
}
}
+ if (!v.empty())
+ {
+ m_onUpdate(v);
+ }
+
insertToDiffLog (diffLog);
}
catch (Error::SyncXmlDecodingFailure &e)
@@ -470,7 +481,7 @@
bind (&SyncLogic::sendSyncInterest, this),
REEXPRESSING_INTEREST);
- m_ccnxHandle->sendInterest (os.str (),
+ m_ccnxHandle->sendInterestForString (os.str (),
bind (&SyncLogic::respondSyncData, this, _1, _2));
}
@@ -492,7 +503,7 @@
REEXPRESSING_RECOVERY_INTEREST);
}
- m_ccnxHandle->sendInterest (os.str (),
+ m_ccnxHandle->sendInterestForString (os.str (),
bind (&SyncLogic::respondSyncData, this, _1, _2));
}
@@ -502,7 +513,7 @@
{
_LOG_TRACE (">> D " << name);
// sending
- m_ccnxHandle->publishData (name,
+ m_ccnxHandle->publishStringData (name,
lexical_cast<string> (*state),
m_syncResponseFreshness); // in NS-3 it doesn't have any effect... yet
diff --git a/model/sync-logic.h b/model/sync-logic.h
index e205fc0..daa6e2f 100644
--- a/model/sync-logic.h
+++ b/model/sync-logic.h
@@ -50,6 +50,12 @@
namespace Sync {
+struct MissingDataInfo {
+ std::string prefix;
+ SeqNo low;
+ SeqNo high;
+};
+
/**
* \ingroup sync
* @brief A wrapper for SyncApp, which handles ccnx related things (process
@@ -61,7 +67,8 @@
#endif
{
public:
- typedef boost::function< void ( const std::string &/*prefix*/, const SeqNo &/*newSeq*/, const SeqNo &/*oldSeq*/ ) > LogicUpdateCallback;
+ //typedef boost::function< void ( const std::string &/*prefix*/, const SeqNo &/*newSeq*/, const SeqNo &/*oldSeq*/ ) > LogicUpdateCallback;
+ typedef boost::function< void (const std::vector<MissingDataInfo> & ) > LogicUpdateCallback;
typedef boost::function< void ( const std::string &/*prefix*/ ) > LogicRemoveCallback;
/**
diff --git a/model/sync-seq-no.h b/model/sync-seq-no.h
index 5dbe490..4f84482 100644
--- a/model/sync-seq-no.h
+++ b/model/sync-seq-no.h
@@ -122,6 +122,26 @@
}
bool
+ operator <= (const SeqNo &seq) const
+ {
+ return m_session <= seq.m_session || (m_session == seq.m_session && m_seq <= seq.m_seq);
+ }
+
+ SeqNo &
+ operator ++ ()
+ {
+ m_seq ++;
+ return *this;
+ }
+
+ SeqNo &
+ operator --()
+ {
+ m_seq --;
+ return *this;
+ }
+
+ bool
isValid () const
{
return m_valid;
diff --git a/test/test_app_socket.cc b/test/test_app_socket.cc
index 3e4394e..8a3f864 100644
--- a/test/test_app_socket.cc
+++ b/test/test_app_socket.cc
@@ -49,6 +49,19 @@
// cout << str1 << ", " << str2 << endl;
}
+ void fetchAll(vector<MissingDataInfo> &v, SyncAppSocket *socket) {
+ int n = v.size();
+ for (int i = 0; i < n; i++) {
+ SeqNo s = ++v[i].low;
+ for(; s <= v[i].high; ++s) {
+ socket->fetchString(v[i].prefix, s, bind(&TestSocketApp::set, this, _1, _2));
+ }
+ }
+ }
+
+ void pass(const string &prefix) {
+ }
+
string toString(){
map<string, string>::iterator it = data.begin();
string str = "\n";
@@ -76,18 +89,18 @@
string p1("/irl.cs.ucla.edu"), p2("/yakshi.org"), p3("/google.com");
_LOG_DEBUG ("s1");
- SyncAppSocket s1 (syncPrefix, bind(&TestSocketApp::set, &a1, _1, _2));
+ SyncAppSocket s1 (syncPrefix, bind(&TestSocketApp::fetchAll, &a1, _1, _2), bind(&TestSocketApp::pass, &a1, _1));
this_thread::sleep (posix_time::milliseconds (50));
_LOG_DEBUG ("s2");
- SyncAppSocket s2 (syncPrefix, bind(&TestSocketApp::set, &a2, _1, _2));
+ SyncAppSocket s2 (syncPrefix, bind(&TestSocketApp::fetchAll, &a2, _1, _2), bind(&TestSocketApp::pass, &a2, _1));
this_thread::sleep (posix_time::milliseconds (50));
- SyncAppSocket s3 (syncPrefix, bind(&TestSocketApp::set, &a3, _1, _2));
+ SyncAppSocket s3 (syncPrefix, bind(&TestSocketApp::fetchAll, &a3, _1, _2), bind(&TestSocketApp::pass, &a3, _1));
this_thread::sleep (posix_time::milliseconds (50));
// single source
string data0 = "Very funny Scotty, now beam down my clothes";
_LOG_DEBUG ("s1 publish");
- s1.publish (p1, 0, data0, 10);
+ s1.publishString (p1, 0, data0, 10);
this_thread::sleep (posix_time::milliseconds (120));
// from code logic, we won't be fetching our own data
@@ -100,9 +113,9 @@
string data2 = "Don't look conspicuous, it draws fire";
_LOG_DEBUG ("s1 publish");
- s1.publish (p1, 0, data1, 10);
+ s1.publishString (p1, 0, data1, 10);
_LOG_DEBUG ("s1 publish");
- s1.publish (p1, 0, data2, 10);
+ s1.publishString (p1, 0, data2, 10);
this_thread::sleep (posix_time::milliseconds (1000));
// from code logic, we won't be fetching our own data
@@ -116,12 +129,12 @@
string data4 = "I got a fortune cookie once that said 'You like Chinese food'";
string data5 = "Real men wear pink. Why? Because their wives make them";
_LOG_DEBUG ("s3 publish");
- s3.publish(p3, 0, data3, 10);
+ s3.publishString(p3, 0, data3, 10);
this_thread::sleep (posix_time::milliseconds (200));
// another single source, multiple data at once
- s2.publish(p2, 0, data4, 10);
- s2.publish(p2, 0, data5, 10);
+ s2.publishString(p2, 0, data4, 10);
+ s2.publishString(p2, 0, data5, 10);
this_thread::sleep (posix_time::milliseconds (1000));
// from code logic, we won't be fetching our own data
@@ -135,9 +148,9 @@
_LOG_DEBUG ("Simultaneous publishing");
string data6 = "Shakespeare says: 'Prose before hos.'";
string data7 = "Pick good people, talent never wears out";
- s1.publish(p1, 0, data6, 10);
+ s1.publishString(p1, 0, data6, 10);
// this_thread::sleep (posix_time::milliseconds (1000));
- s2.publish(p2, 0, data7, 10);
+ s2.publishString(p2, 0, data7, 10);
this_thread::sleep (posix_time::milliseconds (1500));
// from code logic, we won't be fetching our own data
diff --git a/test/test_ccnx_wrapper.cc b/test/test_ccnx_wrapper.cc
index 6934584..3db44ba 100644
--- a/test/test_ccnx_wrapper.cc
+++ b/test/test_ccnx_wrapper.cc
@@ -62,7 +62,7 @@
this_thread::sleep (posix_time::milliseconds (10));
string interest = "/ucla.edu/0";
- hb.sendInterest(interest, memberFunc);
+ hb.sendInterestForString(interest, memberFunc);
// give time for ccnd to react
sleep(1);
@@ -71,9 +71,9 @@
string name = "/ucla.edu/0";
string data = "random bits: !#$!@#$!";
- ha.publishData(name, data, 5);
+ ha.publishStringData(name, data, 5);
- hb.sendInterest(interest, memberFunc);
+ hb.sendInterestForString(interest, memberFunc);
// give time for ccnd to react
this_thread::sleep (posix_time::milliseconds (5));
diff --git a/test/test_data_fetch_and_publish.cc b/test/test_data_fetch_and_publish.cc
index 157ce7f..e75b274 100644
--- a/test/test_data_fetch_and_publish.cc
+++ b/test/test_data_fetch_and_publish.cc
@@ -20,6 +20,7 @@
* Alexander Afanasyev <alexander.afanasyev@ucla.edu>
*/
+/*
#include <boost/test/unit_test.hpp>
#include <boost/test/output_test_stream.hpp>
#include <map>
@@ -71,7 +72,7 @@
string str[5] = {"panda", "express", "tastes", "so", "good"};
for (int i = 0; i < 5; i++) {
- foo.set(interest + "/" + "0/" /*session*/ + seq[i], str[i]);
+ foo.set(interest + "/" + "0/" + seq[i], str[i]);
}
boost::function<void (string, string)> setFunc =
@@ -107,5 +108,6 @@
BOOST_CHECK_EQUAL(poo.toString(), bar.toString());
}
+*/
diff --git a/test/test_sync_logic.cc b/test/test_sync_logic.cc
index f6b6808..0d77cd0 100644
--- a/test/test_sync_logic.cc
+++ b/test/test_sync_logic.cc
@@ -43,7 +43,14 @@
: instance (_instance)
{
}
-
+
+ void wrapper (vector<MissingDataInfo> &v) {
+ int n = v.size();
+ for (int i = 0; i < n; i++) {
+ onUpdate (v[i].prefix, v[i].high, v[i].low);
+ }
+ }
+
void onUpdate (const string &p/*prefix*/, const SeqNo &seq/*newSeq*/, const SeqNo &oldSeq/*oldSeq*/)
{
m_map[p] = seq.getSeq ();
@@ -68,7 +75,7 @@
{
Handler h1 ("1");
- SyncLogic l1 ("/bcast", bind (&Handler::onUpdate, &h1, _1, _2, _3), bind (&Handler::onRemove, &h1, _1));
+ SyncLogic l1 ("/bcast", bind (&Handler::wrapper, &h1, _1), bind (&Handler::onRemove, &h1, _1));
l1.addLocalNames ("/one", 1, 2);
BOOST_CHECK_EQUAL (h1.m_map.size (), 0);
@@ -76,7 +83,7 @@
BOOST_CHECK_EQUAL (h1.m_map.size (), 0);
Handler h2 ("2");
- SyncLogic l2 ("/bcast", bind (&Handler::onUpdate, &h2, _1, _2, _3), bind (&Handler::onRemove, &h2, _1));
+ SyncLogic l2 ("/bcast", bind (&Handler::wrapper, &h2, _1), bind (&Handler::onRemove, &h2, _1));
sleep (1);
BOOST_CHECK_EQUAL (h1.m_map.size (), 0);
diff --git a/wscript b/wscript
index c4ece42..0bcb242 100644
--- a/wscript
+++ b/wscript
@@ -122,9 +122,9 @@
'ccnx/sync-ccnx-wrapper.cc',
'ccnx/sync-scheduler.cc',
'ccnx/sync-log.cc',
- 'ccnx/sync-app-data-fetch.cc',
- 'ccnx/sync-app-data-publish.cc',
- 'ccnx/sync-app-socket-c.cc',
+ #'ccnx/sync-app-data-fetch.cc',
+ #'ccnx/sync-app-data-publish.cc',
+ #'ccnx/sync-app-socket-c.cc',
'ccnx/sync-app-socket.cc',
'model/sync-diff-leaf.cc',