Adapt sync to use new ndn.cxx api
diff --git a/src/ccnx/sync-app-socket-c.cc b/src/ccnx/sync-app-socket-c.cc
deleted file mode 100644
index e23c8d1..0000000
--- a/src/ccnx/sync-app-socket-c.cc
+++ /dev/null
@@ -1,146 +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-socket.h"
-#include <boost/shared_array.hpp>
-using namespace std;
-using namespace Sync;
-
-struct SyncAppSocketStruct;
-struct MissingDataInfoC
-{
- const char *prefix;
- int session;
- int low;
- int high;
-};
-
-class CallbackHolder {
-private:
- void (*m_callback)(const char *, const char *);
-
-public:
- CallbackHolder(void (*callback)(const char*, const char*))
- :m_callback(callback){};
-
- void callbackWrapper(const string &name, const string &data) {
- if (m_callback != NULL)
- (*m_callback)(name.c_str(), data.c_str());
- }
-};
-
-class UpdateCallbackHolder {
-private:
- void (*m_callback)(const struct MissingDataInfoC*, const int, const SyncAppSocketStruct*);
-
-public:
- UpdateCallbackHolder(void (*callback)(
- const MissingDataInfoC*,
- const int,
- const SyncAppSocketStruct*))
- :m_callback(callback){};
-
- void callbackWrapper(const vector<MissingDataInfo> &mdi, SyncAppSocket* socket) {
- boost::shared_array<MissingDataInfoC> mdic(new MissingDataInfoC[mdi.size()]);
- int i;
-
- for (i = 0; i < mdi.size(); i++)
- {
- mdic[i].prefix = mdi[i].prefix.c_str();
- mdic[i].session = mdi[i].high.getSession();
- if (mdi[i].low.getSession() != mdi[i].high.getSession())
- mdic[i].low = 0;
- else
- mdic[i].low = mdi[i].low.getSeq();
- mdic[i].high = mdi[i].high.getSeq();
- }
- if (m_callback != NULL)
- (*m_callback)(mdic.get(), i, (const SyncAppSocketStruct*) socket);
- }
-};
-
-class RemoveCallbackHolder {
-private:
- void (*m_callback)(const char*);
-
-public:
- RemoveCallbackHolder(void (*callback)(const char*)):m_callback(callback){};
- void callbackWrapper(const string &prefix) {
- if (m_callback != NULL)
- (*m_callback)(prefix.c_str());
- }
-};
-
-extern "C"
-SyncAppSocketStruct *
-create_sync_app_socket(const char *prefix,
- void (*updatecallback)(const MissingDataInfoC*, const int, const SyncAppSocketStruct*),
- void (*removecallback)(const char *))
-{
- boost::shared_ptr<UpdateCallbackHolder> uh(new UpdateCallbackHolder(updatecallback));
- boost::shared_ptr<RemoveCallbackHolder> rh(new RemoveCallbackHolder(removecallback));
- boost::function<void (const vector<MissingDataInfo>&, SyncAppSocket*)> ucb = bind(&UpdateCallbackHolder::callbackWrapper, uh, _1, _2);
- boost::function<void (const string&)> rcb = bind(&RemoveCallbackHolder::callbackWrapper, rh, _1);
- SyncAppSocket *sock = new SyncAppSocket(prefix, ucb, rcb);
- return (SyncAppSocketStruct *) sock;
-}
-
-extern "C"
-void
-delete_sync_app_socket(SyncAppSocketStruct **sock) {
- SyncAppSocket *temp = *((SyncAppSocket **)sock);
- delete temp;
- temp = NULL;
-}
-
-// assume char *buf ends with '\0', or otherwise it's going to crash;
-// should fix this "feature"
-extern "C"
-int
-sync_app_socket_publish(const SyncAppSocketStruct *sock, const char *prefix, int session, const char *buf, int freshness)
-{
- SyncAppSocket *temp = (SyncAppSocket*) sock;
- return temp->publishString(prefix, session, buf, freshness);
-}
-
-extern "C"
-void
-sync_app_socket_remove(const SyncAppSocketStruct *sock, const char *prefix)
-{
- SyncAppSocket *temp = (SyncAppSocket*) sock;
- temp->remove(prefix);
-}
-
-extern "C"
-void
-sync_app_socket_fetch(const SyncAppSocketStruct *sock, const char *prefix, int session, int sequence,
- void (*callback)(const char*, const char*), int retry)
-{
- SyncAppSocket *temp = (SyncAppSocket*) sock;
- string s(prefix);
- SeqNo seq(session, sequence);
- boost::shared_ptr<CallbackHolder> h(new CallbackHolder(callback));
- boost::function<void (const string&, const string&)> cb = bind(&CallbackHolder::callbackWrapper, h, _1, _2);
-
- temp->fetchString(s, seq, cb, retry);
-}
-
diff --git a/src/ccnx/sync-app-socket-c.h b/src/ccnx/sync-app-socket-c.h
deleted file mode 100644
index 7ab9e70..0000000
--- a/src/ccnx/sync-app-socket-c.h
+++ /dev/null
@@ -1,52 +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_SOCKET_C_H
-#define SYNC_APP_SOCKET_C_H
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
- typedef struct SyncAppSocketStruct SyncAppSocketStruct;
-
- struct MissingDataInfoC
- {
- const char *prefix;
- int session;
- int low;
- int high;
- };
-
- SyncAppSocketStruct *create_sync_app_socket(const char *prefix,
- void (*updatecallback)(const struct MissingDataInfoC*, const int, const SyncAppSocketStruct*),
- void (*removecallback)(const char *));
- void delete_sync_app_socket(SyncAppSocketStruct **sock);
- int sync_app_socket_publish(const SyncAppSocketStruct *sock, const char *prefix, int session, const char *buf, int freshness);
- void sync_app_socket_remove(const SyncAppSocketStruct *sock, const char *prefix);
- void sync_app_socket_fetch(const SyncAppSocketStruct *sock, const char *prefix, int session, int seq,
- void (*callback)(const char*, const char*), int retry);
-#ifdef __cplusplus
-}
-#endif
-
-#endif // SYNC_APP_SOCKET_C_H
diff --git a/src/ccnx/sync-app-socket.cc b/src/ccnx/sync-app-socket.cc
deleted file mode 100644
index 0e4e829..0000000
--- a/src/ccnx/sync-app-socket.cc
+++ /dev/null
@@ -1,115 +0,0 @@
-/* -*- Mode: C32++; 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-socket.h"
-
-using namespace std;
-using namespace boost;
-
-namespace Sync
-{
-
-SyncAppSocket::SyncAppSocket (const string &syncPrefix, NewDataCallback dataCallback, RemoveCallback rmCallback )
- : m_newDataCallback(dataCallback)
- , m_ccnxHandle (new CcnxWrapper())
- , m_syncLogic (syncPrefix,
- bind(&SyncAppSocket::passCallback, this, _1),
- rmCallback)
-{
-}
-
-SyncAppSocket::~SyncAppSocket()
-{
-}
-
-std::string
-SyncAppSocket::GetLocalPrefix()
-{
- // this handle is supposed to be short lived
- CcnxWrapperPtr handle( new CcnxWrapper());
- return handle->getLocalPrefix();
-}
-
-bool
-SyncAppSocket::publishString (const string &prefix, uint32_t session, const string &dataBuffer, int 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);
- return true;
-}
-
-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);
- return true;
-}
-
-
-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();
- //std::cout << "Socket " << this << " Send Interest <" << interestName.str() << "> for raw data " << endl;
- 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/src/ccnx/sync-app-socket.h b/src/ccnx/sync-app-socket.h
deleted file mode 100644
index ceeb5ce..0000000
--- a/src/ccnx/sync-app-socket.h
+++ /dev/null
@@ -1,112 +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_SOCKET_H
-#define SYNC_APP_SOCKET_H
-
-#include "sync-logic.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 {
-
-/**
- * \ingroup sync
- * @brief A simple interface to interact with client code
- */
-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
- * parameter should be passed to the constructor of m_fetcher; furthermore,
- * the fetch function of m_fetcher should be a second paramter passed to
- * the constructor of m_syncAppWrapper, so that m_syncAppWrapper can tell
- * m_fetcher to fetch the actual app data after it learns the names
- *
- * @param syncPrefix the name prefix for Sync Interest
- * @param dataCallback the callback to process data
- */
- SyncAppSocket (const std::string &syncPrefix, NewDataCallback dataCallback, RemoveCallback rmCallback);
- ~SyncAppSocket ();
-
- /**
- * @brief publish data from local client and tell SyncLogic to update
- * the sync tree by adding the local names
- *
- * @param prefix the name prefix for the data
- * @param session session to which data is published
- * @param dataBuffer the data itself
- * @param freshness the freshness time for the data (in seconds)
- */
- 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
- * this is just a wrapper
- *
- * @param prefix the prefix for the participant
- */
- void remove (const std::string &prefix) {m_syncLogic.remove(prefix);}
-
- 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);
-
- // for sync-demo
- std::string getRootDigest() {return m_syncLogic.getRootDigest();}
- uint32_t
- getNextSeq (const std::string &prefix, uint32_t session);
-
- SyncLogic &
- getLogic () { return m_syncLogic; }
-
- // make this a static function so we don't have to create socket instance without
- // knowing the local prefix. it's a wrong place for this function anyway
- static std::string
- GetLocalPrefix ();
-
-private:
- void
- passCallback(const std::vector<MissingDataInfo> &v) {m_newDataCallback(v, this);}
-
-private:
- typedef boost::unordered_map<std::string, SeqNo> SequenceLog;
- NewDataCallback m_newDataCallback;
- SequenceLog m_sequenceLog;
- CcnxWrapperPtr m_ccnxHandle;
- SyncLogic m_syncLogic;
-};
-
-} // Sync
-
-#endif // SYNC_APP_SOCKET_H
diff --git a/src/ccnx/sync-ccnx-wrapper.cc b/src/ccnx/sync-ccnx-wrapper.cc
deleted file mode 100644
index c824e72..0000000
--- a/src/ccnx/sync-ccnx-wrapper.cc
+++ /dev/null
@@ -1,597 +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-ccnx-wrapper.h"
-
-extern "C" {
-#include <ccn/fetch.h>
-}
-
-#include "sync-logging.h"
-#include <poll.h>
-#include <boost/throw_exception.hpp>
-#include <boost/date_time/posix_time/posix_time.hpp>
-#include <boost/random.hpp>
-
-#include "sync-scheduler.h"
-
-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;
-
-INIT_LOGGER ("CcnxWrapper");
-
-namespace Sync {
-
-#ifdef _DEBUG_WRAPPER_
-CcnxWrapper::CcnxWrapper(char c)
-#else
-CcnxWrapper::CcnxWrapper()
-#endif
- : m_handle (0)
- , m_keyStore (0)
- , m_keyLoactor (0)
- , m_running (true)
- , m_connected (false)
-{
-#ifdef _DEBUG_WRAPPER_
- m_c = c;
-#endif
- connectCcnd();
- initKeyStore ();
- createKeyLocator ();
- m_thread = thread (&CcnxWrapper::ccnLoop, this);
-}
-
-void
-CcnxWrapper::connectCcnd()
-{
- recursive_mutex::scoped_lock lock (m_mutex);
-
- if (m_handle != 0) {
- ccn_disconnect (m_handle);
- ccn_destroy (&m_handle);
- }
-
- m_handle = ccn_create ();
- _LOG_DEBUG("<<< connecting to ccnd");
- if (ccn_connect(m_handle, NULL) < 0)
- {
- _LOG_DEBUG("<<< connecting to ccnd failed");
- BOOST_THROW_EXCEPTION (CcnxOperationException() << errmsg_info_str("connection to ccnd failed"));
- }
- m_connected = true;
-
- if (!m_registeredInterests.empty())
- {
- for (map<std::string, InterestCallback>::const_iterator it = m_registeredInterests.begin(); it != m_registeredInterests.end(); ++it)
- {
- // clearInterestFilter(it->first);
- setInterestFilter(it->first, it->second);
- _LOG_DEBUG("<<< registering interest filter for: " << it->first);
- }
- }
-}
-
-CcnxWrapper::~CcnxWrapper()
-{
- // std::cout << "CcnxWrapper::~CcnxWrapper()" << std::endl;
- {
- recursive_mutex::scoped_lock lock(m_mutex);
- m_running = false;
- }
-
- m_thread.join ();
- ccn_disconnect (m_handle);
- ccn_destroy (&m_handle);
- ccn_charbuf_destroy (&m_keyLoactor);
- ccn_keystore_destroy (&m_keyStore);
-}
-
-/// @cond include_hidden
-
-void
-CcnxWrapper::createKeyLocator ()
-{
- m_keyLoactor = ccn_charbuf_create();
- ccn_charbuf_append_tt (m_keyLoactor, CCN_DTAG_KeyLocator, CCN_DTAG);
- ccn_charbuf_append_tt (m_keyLoactor, CCN_DTAG_Key, CCN_DTAG);
- int res = ccn_append_pubkey_blob (m_keyLoactor, ccn_keystore_public_key(m_keyStore));
- if (res >= 0)
- {
- ccn_charbuf_append_closer (m_keyLoactor); /* </Key> */
- ccn_charbuf_append_closer (m_keyLoactor); /* </KeyLocator> */
- }
-}
-
-const ccn_pkey*
-CcnxWrapper::getPrivateKey ()
-{
- return ccn_keystore_private_key (m_keyStore);
-}
-
-const unsigned char*
-CcnxWrapper::getPublicKeyDigest ()
-{
- return ccn_keystore_public_key_digest(m_keyStore);
-}
-
-ssize_t
-CcnxWrapper::getPublicKeyDigestLength ()
-{
- return ccn_keystore_public_key_digest_length(m_keyStore);
-}
-
-void
-CcnxWrapper::initKeyStore ()
-{
- m_keyStore = ccn_keystore_create ();
- string keyStoreFile = string(getenv("HOME")) + string("/.ccnx/.ccnx_keystore");
- if (ccn_keystore_init (m_keyStore, (char *)keyStoreFile.c_str(), (char*)"Th1s1sn0t8g00dp8ssw0rd.") < 0)
- BOOST_THROW_EXCEPTION(CcnxOperationException() << errmsg_info_str(keyStoreFile.c_str()));
-}
-
-void
-CcnxWrapper::ccnLoop ()
-{
- _LOG_FUNCTION (this);
- static boost::mt19937 randomGenerator (static_cast<unsigned int> (std::time (0)));
- static boost::variate_generator<boost::mt19937&, boost::uniform_int<> > rangeUniformRandom (randomGenerator, uniform_int<> (0,1000));
-
- while (m_running)
- {
- try
- {
-#ifdef _DEBUG_WRAPPER_
- std::cout << m_c << flush;
-#endif
- int res = 0;
- {
- recursive_mutex::scoped_lock lock (m_mutex);
- res = ccn_run (m_handle, 0);
- }
-
- if (!m_running) break;
-
- if (res < 0)
- BOOST_THROW_EXCEPTION (CcnxOperationException()
- << errmsg_info_str("ccn_run returned error"));
-
-
- pollfd pfds[1];
- {
- recursive_mutex::scoped_lock lock (m_mutex);
-
- pfds[0].fd = ccn_get_connection_fd (m_handle);
- pfds[0].events = POLLIN;
- if (ccn_output_is_pending (m_handle))
- pfds[0].events |= POLLOUT;
- }
-
- int ret = poll (pfds, 1, 1);
- if (ret < 0)
- {
- BOOST_THROW_EXCEPTION (CcnxOperationException() << errmsg_info_str("ccnd socket failed (probably ccnd got stopped)"));
- }
- }
- catch (CcnxOperationException &e)
- {
- // do not try reconnect for now
- throw e;
- /*
- m_connected = false;
- // probably ccnd has been stopped
- // try reconnect with sleep
- int interval = 1;
- int maxInterval = 32;
- while (m_running)
- {
- try
- {
- this_thread::sleep (boost::get_system_time () + TIME_SECONDS(interval) + TIME_MILLISECONDS (rangeUniformRandom ()));
-
- connectCcnd();
- _LOG_DEBUG("reconnect to ccnd succeeded");
- break;
- }
- catch (CcnxOperationException &e)
- {
- this_thread::sleep (boost::get_system_time () + TIME_SECONDS(interval) + TIME_MILLISECONDS (rangeUniformRandom ()));
-
- // do exponential backup for reconnect interval
- if (interval < maxInterval)
- {
- interval *= 2;
- }
- }
- }
- */
- }
- catch (const std::exception &exc)
- {
- // catch anything thrown within try block that derives from std::exception
- std::cerr << exc.what();
- }
- catch (...)
- {
- cout << "UNKNOWN EXCEPTION !!!" << endl;
- }
-
- }
-}
-
-/// @endcond
-int
-CcnxWrapper::publishStringData (const string &name, const string &dataBuffer, int freshness) {
- return publishRawData(name, dataBuffer.c_str(), dataBuffer.length(), freshness);
-}
-
-int
-CcnxWrapper::publishRawData (const string &name, const char *buf, size_t len, int freshness)
-{
-
- recursive_mutex::scoped_lock lock(m_mutex);
- if (!m_running || !m_connected)
- return -1;
-
- // cout << "Publish: " << name << endl;
- ccn_charbuf *pname = ccn_charbuf_create();
- ccn_charbuf *signed_info = ccn_charbuf_create();
- ccn_charbuf *content = ccn_charbuf_create();
-
- ccn_name_from_uri(pname, name.c_str());
- ccn_signed_info_create(signed_info,
- getPublicKeyDigest(),
- getPublicKeyDigestLength(),
- NULL,
- CCN_CONTENT_DATA,
- freshness,
- NULL,
- m_keyLoactor);
- if(ccn_encode_ContentObject(content, pname, signed_info,
- (const unsigned char *)buf, len,
- NULL, getPrivateKey()) < 0)
- {
- // BOOST_THROW_EXCEPTION(CcnxOperationException() << errmsg_info_str("encode content failed"));
- _LOG_ERROR("<<< Encode content failed " << name);
- }
-
- if (ccn_put(m_handle, content->buf, content->length) < 0)
- {
- // BOOST_THROW_EXCEPTION(CcnxOperationException() << errmsg_info_str("ccnput failed"));
- _LOG_ERROR("<<< ccnput content failed " << name);
- }
-
- ccn_charbuf_destroy (&pname);
- ccn_charbuf_destroy (&signed_info);
- ccn_charbuf_destroy (&content);
- return 0;
-}
-
-
-static ccn_upcall_res
-incomingInterest(ccn_closure *selfp,
- ccn_upcall_kind kind,
- ccn_upcall_info *info)
-{
- CcnxWrapper::InterestCallback *f = static_cast<CcnxWrapper::InterestCallback*> (selfp->data);
-
- switch (kind)
- {
- case CCN_UPCALL_FINAL: // effective in unit tests
- delete f;
- delete selfp;
- return CCN_UPCALL_RESULT_OK;
-
- case CCN_UPCALL_INTEREST:
- break;
-
- default:
- return CCN_UPCALL_RESULT_OK;
- }
-
- string interest;
- for (int i = 0; i < info->interest_comps->n - 1; i++)
- {
- char *comp;
- size_t size;
- interest += "/";
- ccn_name_comp_get(info->interest_ccnb, info->interest_comps, i, (const unsigned char **)&comp, &size);
- string compStr(comp, size);
- interest += compStr;
- }
- (*f) (interest);
- _LOG_DEBUG("<<< processed interest: " << interest);
- return CCN_UPCALL_RESULT_OK;
-}
-
-static ccn_upcall_res
-incomingData(ccn_closure *selfp,
- ccn_upcall_kind kind,
- ccn_upcall_info *info)
-{
- ClosurePass *cp = static_cast<ClosurePass *> (selfp->data);
-
- switch (kind)
- {
- case CCN_UPCALL_FINAL: // effecitve in unit tests
- delete cp;
- cp = NULL;
- delete selfp;
- return CCN_UPCALL_RESULT_OK;
-
- case CCN_UPCALL_CONTENT:
- break;
-
- case CCN_UPCALL_INTEREST_TIMED_OUT: {
- if (cp != NULL && cp->getRetry() > 0) {
- cp->decRetry();
- return CCN_UPCALL_RESULT_REEXPRESS;
- }
- return CCN_UPCALL_RESULT_OK;
- }
-
- default:
- return CCN_UPCALL_RESULT_OK;
- }
-
- char *pcontent;
- size_t len;
- if (ccn_content_get_value(info->content_ccnb, info->pco->offset[CCN_PCO_E], info->pco, (const unsigned char **)&pcontent, &len) < 0)
- {
- // BOOST_THROW_EXCEPTION(CcnxOperationException() << errmsg_info_str("decode ContentObject failed"));
- _LOG_ERROR("<<< Decode content failed ");
- }
-
- string name;
- for (int i = 0; i < info->content_comps->n - 1; i++)
- {
- char *comp;
- size_t size;
- name += "/";
- ccn_name_comp_get(info->content_ccnb, info->content_comps, i, (const unsigned char **)&comp, &size);
- string compStr(comp, size);
- name += compStr;
- }
-
- cp->runCallback(name, pcontent, len);
-
- return CCN_UPCALL_RESULT_OK;
-}
-
-int CcnxWrapper::sendInterestForString (const string &strInterest, const StringDataCallback &strDataCallback, int retry)
-{
- DataClosurePass * pass = new DataClosurePass(STRING_FORM, retry, strDataCallback);
- return sendInterest(strInterest, pass);
-}
-
-int CcnxWrapper::sendInterest (const string &strInterest, const RawDataCallback &rawDataCallback, int retry)
-{
- RawDataClosurePass * pass = new RawDataClosurePass(RAW_DATA, retry, rawDataCallback);
- return sendInterest(strInterest, pass);
-}
-
-int CcnxWrapper::sendInterest (const string &strInterest, void *dataPass)
-{
- recursive_mutex::scoped_lock lock(m_mutex);
- if (!m_running || !m_connected)
- return -1;
-
- // std::cout << "Send interests for " << strInterest << std::endl;
- ccn_charbuf *pname = ccn_charbuf_create();
- ccn_closure *dataClosure = new ccn_closure;
-
- ccn_name_from_uri (pname, strInterest.c_str());
- dataClosure->data = dataPass;
-
- dataClosure->p = &incomingData;
- if (ccn_express_interest (m_handle, pname, dataClosure, NULL) < 0)
- {
- // BOOST_THROW_EXCEPTION(CcnxOperationException() << errmsg_info_str("express interest failed"));
- _LOG_ERROR("<<< Express interest failed: " << strInterest);
- }
-
- _LOG_DEBUG("<<< Sending interest: " << strInterest);
-
- ccn_charbuf_destroy (&pname);
- return 0;
-}
-
-int CcnxWrapper::setInterestFilter (const string &prefix, const InterestCallback &interestCallback)
-{
- recursive_mutex::scoped_lock lock(m_mutex);
- if (!m_running || !m_connected)
- return -1;
-
- ccn_charbuf *pname = ccn_charbuf_create();
- ccn_closure *interestClosure = new ccn_closure;
-
- ccn_name_from_uri (pname, prefix.c_str());
- interestClosure->data = new InterestCallback (interestCallback); // should be removed when closure is removed
- interestClosure->p = &incomingInterest;
- int ret = ccn_set_interest_filter (m_handle, pname, interestClosure);
- if (ret < 0)
- {
- BOOST_THROW_EXCEPTION(CcnxOperationException() << errmsg_info_str("set interest filter failed") << errmsg_info_int (ret));
- }
-
- m_registeredInterests.insert(pair<std::string, InterestCallback>(prefix, interestCallback));
- ccn_charbuf_destroy(&pname);
-
- return 0;
-}
-
-void
-CcnxWrapper::clearInterestFilter (const std::string &prefix)
-{
- recursive_mutex::scoped_lock lock(m_mutex);
- if (!m_running || !m_connected)
- return;
-
- std::cout << "clearInterestFilter" << std::endl;
- ccn_charbuf *pname = ccn_charbuf_create();
-
- ccn_name_from_uri (pname, prefix.c_str());
- int ret = ccn_set_interest_filter (m_handle, pname, 0);
- if (ret < 0)
- {
- BOOST_THROW_EXCEPTION(CcnxOperationException() << errmsg_info_str("set interest filter failed") << errmsg_info_int (ret));
- }
-
- m_registeredInterests.erase(prefix);
- ccn_charbuf_destroy(&pname);
-}
-
-string
-CcnxWrapper::getLocalPrefix ()
-{
- struct ccn * tmp_handle = ccn_create ();
- int res = ccn_connect (tmp_handle, NULL);
- if (res < 0)
- {
- _LOG_ERROR ("connecting to ccnd failed");
- return "";
- }
-
- string retval = "";
-
- struct ccn_charbuf *templ = ccn_charbuf_create();
- ccn_charbuf_append_tt(templ, CCN_DTAG_Interest, CCN_DTAG);
- ccn_charbuf_append_tt(templ, CCN_DTAG_Name, CCN_DTAG);
- ccn_charbuf_append_closer(templ); /* </Name> */
- // XXX - use pubid if possible
- ccn_charbuf_append_tt(templ, CCN_DTAG_MaxSuffixComponents, CCN_DTAG);
- ccnb_append_number(templ, 1);
- ccn_charbuf_append_closer(templ); /* </MaxSuffixComponents> */
- ccnb_tagged_putf(templ, CCN_DTAG_Scope, "%d", 2);
- ccn_charbuf_append_closer(templ); /* </Interest> */
-
- struct ccn_charbuf *name = ccn_charbuf_create ();
- res = ccn_name_from_uri (name, "/local/ndn/prefix");
- if (res < 0) {
- _LOG_ERROR ("Something wrong with `ccn_name_from_uri` call");
- }
- else
- {
- struct ccn_fetch *fetch = ccn_fetch_new (tmp_handle);
-
- struct ccn_fetch_stream *stream = ccn_fetch_open (fetch, name, "/local/ndn/prefix",
- NULL, 4, CCN_V_HIGHEST, 0);
- if (stream == NULL) {
- _LOG_ERROR ("Cannot create ccn_fetch_stream");
- }
- else
- {
- ostringstream os;
-
- int counter = 0;
- char buf[256];
- while (true) {
- res = ccn_fetch_read (stream, buf, sizeof(buf));
-
- if (res == 0) {
- break;
- }
-
- if (res > 0) {
- os << string(buf, res);
- } else if (res == CCN_FETCH_READ_NONE) {
- if (counter < 2)
- {
- ccn_run(tmp_handle, 1000);
- counter ++;
- }
- else
- {
- break;
- }
- } else if (res == CCN_FETCH_READ_END) {
- break;
- } else if (res == CCN_FETCH_READ_TIMEOUT) {
- break;
- } else {
- _LOG_ERROR ("fatal error. should be reported");
- break;
- }
- }
- retval = os.str ();
- stream = ccn_fetch_close(stream);
- }
- fetch = ccn_fetch_destroy(fetch);
- }
-
- ccn_charbuf_destroy (&name);
-
- ccn_disconnect (tmp_handle);
- ccn_destroy (&tmp_handle);
-
- return retval;
-}
-
-
-
-
-
-DataClosurePass::DataClosurePass (CallbackType type, int retry, const CcnxWrapper::StringDataCallback &strDataCallback): ClosurePass(type, retry), m_callback(NULL)
-{
- m_callback = new CcnxWrapper::StringDataCallback (strDataCallback);
-}
-
-DataClosurePass::~DataClosurePass ()
-{
- delete m_callback;
- m_callback = NULL;
-}
-
-void
-DataClosurePass::runCallback(std::string name, const char *data, size_t len)
-{
- string content(data, len);
- if (m_callback != NULL) {
- (*m_callback)(name, content);
- }
-}
-
-
-RawDataClosurePass::RawDataClosurePass (CallbackType type, int retry, const CcnxWrapper::RawDataCallback &rawDataCallback): ClosurePass(type, retry), m_callback(NULL)
-{
- m_callback = new CcnxWrapper::RawDataCallback (rawDataCallback);
-}
-
-RawDataClosurePass::~RawDataClosurePass ()
-{
- delete m_callback;
- m_callback = NULL;
-}
-
-void
-RawDataClosurePass::runCallback(std::string name, const char *data, size_t len)
-{
- if (m_callback != NULL) {
- (*m_callback)(name, data, len);
- }
-}
-
-}
diff --git a/src/ccnx/sync-ccnx-wrapper.h b/src/ccnx/sync-ccnx-wrapper.h
deleted file mode 100644
index f0f8d8c..0000000
--- a/src/ccnx/sync-ccnx-wrapper.h
+++ /dev/null
@@ -1,198 +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_CCNX_WRAPPER_H
-#define SYNC_CCNX_WRAPPER_H
-
-extern "C" {
-#include <ccn/ccn.h>
-#include <ccn/charbuf.h>
-#include <ccn/keystore.h>
-#include <ccn/uri.h>
-#include <ccn/bloom.h>
-#include <ccn/signing.h>
-}
-
-#include <boost/exception/all.hpp>
-#include <boost/thread/recursive_mutex.hpp>
-#include <boost/thread/thread.hpp>
-#include <boost/function.hpp>
-#include <string>
-#include <sstream>
-#include <map>
-
-/**
- * \defgroup sync SYNC protocol
- *
- * Implementation of SYNC protocol
- */
-namespace Sync {
-
-struct CcnxOperationException : virtual boost::exception, virtual std::exception { };
-/**
- * \ingroup sync
- * @brief A wrapper for ccnx library; clients of this code do not need to deal
- * with ccnx library
- */
-class CcnxWrapper {
-public:
- 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;
-
-public:
-
-#ifdef _DEBUG_WRAPPER_
- CcnxWrapper(char c='.');
- char m_c;
-#else
- CcnxWrapper();
-#endif
-
- ~CcnxWrapper();
-
- /**
- * @brief send Interest; need to grab lock m_mutex first
- *
- * @param strInterest the Interest name
- * @param dataCallback the callback function to deal with the returned data
- * @return the return code of ccn_express_interest
- */
- int
- sendInterestForString (const std::string &strInterest, const StringDataCallback &strDataCallback, int retry = 0);
-
- int
- sendInterest (const std::string &strInterest, const RawDataCallback &rawDataCallback, int retry = 0);
-
- /**
- * @brief set Interest filter (specify what interest you want to receive)
- *
- * @param prefix the prefix of Interest
- * @param interestCallback the callback function to deal with the returned data
- * @return the return code of ccn_set_interest_filter
- */
- int
- setInterestFilter (const std::string &prefix, const InterestCallback &interestCallback);
-
- /**
- * @brief clear Interest filter
- * @param prefix the prefix of Interest
- */
- void
- clearInterestFilter (const std::string &prefix);
-
- /**
- * @brief publish data and put it to local ccn content store; need to grab
- * lock m_mutex first
- *
- * @param name the name for the data object
- * @param dataBuffer the data to be published
- * @param freshness the freshness time for the data object
- * @return code generated by ccnx library calls, >0 if success
- */
- int
- 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);
-
- std::string
- getLocalPrefix ();
-
-protected:
- void
- connectCcnd();
-
- /// @cond include_hidden
- void
- createKeyLocator ();
-
- void
- initKeyStore ();
-
- const ccn_pkey *
- getPrivateKey ();
-
- const unsigned char *
- getPublicKeyDigest ();
-
- ssize_t
- getPublicKeyDigestLength ();
-
- void
- ccnLoop ();
-
- int
- sendInterest (const std::string &strInterest, void *dataPass);
- /// @endcond
-protected:
- ccn* m_handle;
- ccn_keystore *m_keyStore;
- ccn_charbuf *m_keyLoactor;
- // to lock, use "boost::recursive_mutex::scoped_lock scoped_lock(mutex);
- boost::recursive_mutex m_mutex;
- boost::thread m_thread;
- bool m_running;
- bool m_connected;
- std::map<std::string, InterestCallback> m_registeredInterests;
- // std::list< std::pair<std::string, InterestCallback> > m_registeredInterests;
-};
-
-typedef boost::shared_ptr<CcnxWrapper> CcnxWrapperPtr;
-
-enum CallbackType { STRING_FORM, RAW_DATA};
-
-class ClosurePass {
-public:
- ClosurePass(CallbackType type, int retry): m_retry(retry), m_type(type) {}
- int getRetry() {return m_retry;}
- void decRetry() { m_retry--;}
- CallbackType getCallbackType() {return m_type;}
- virtual ~ClosurePass(){}
- virtual void runCallback(std::string name, const char *data, size_t len) = 0;
-
-protected:
- int m_retry;
- CallbackType m_type;
-};
-
-class DataClosurePass: public ClosurePass {
-public:
- DataClosurePass(CallbackType type, int retry, const CcnxWrapper::StringDataCallback &strDataCallback);
- virtual ~DataClosurePass();
- virtual void runCallback(std::string name, const char *, size_t len);
-private:
- CcnxWrapper::StringDataCallback * m_callback;
-};
-
-class RawDataClosurePass: public ClosurePass {
-public:
- RawDataClosurePass(CallbackType type, int retry, const CcnxWrapper::RawDataCallback &RawDataCallback);
- virtual ~RawDataClosurePass();
- virtual void runCallback(std::string name, const char *, size_t len);
-private:
- CcnxWrapper::RawDataCallback * m_callback;
-};
-
-} // Sync
-
-#endif // SYNC_CCNX_WRAPPER_H
diff --git a/src/ccnx/sync-socket.cc b/src/ccnx/sync-socket.cc
new file mode 100644
index 0000000..48d4ad0
--- /dev/null
+++ b/src/ccnx/sync-socket.cc
@@ -0,0 +1,126 @@
+/* -*- Mode: C32++; 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: Yingdi Yu <yingdi@cs.ucla.edu>
+ */
+
+#include "sync-socket.h"
+
+using namespace std;
+using namespace ndn;
+
+namespace Sync {
+
+SyncSocket::SyncSocket (const string &syncPrefix,
+ Ptr<SyncPolicyManager> syncPolicyManager,
+ NewDataCallback dataCallback,
+ RemoveCallback rmCallback )
+ : m_newDataCallback(dataCallback)
+ , m_syncPolicyManager(syncPolicyManager)
+ , m_syncLogic (syncPrefix,
+ syncPolicyManager,
+ bind(&SyncSocket::passCallback, this, _1),
+ rmCallback)
+{
+ Ptr<security::Keychain> keychain = Ptr<security::Keychain>(new security::Keychain(Ptr<security::IdentityManager>::Create(), m_syncPolicyManager, NULL));
+ m_handler = Ptr<Wrapper>(new Wrapper(keychain));
+}
+
+SyncSocket::~SyncSocket()
+{}
+
+bool
+SyncSocket::publishData(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;
+
+ Name dataName(contentNameWithSeqno.str ());
+ Blob blob(buf, len);
+ Name signingIdentity = m_syncPolicyManager->inferSigningIdentity(dataName);
+
+ m_handler->publishDataByIdentity (dataName,
+ blob,
+ signingIdentity,
+ freshness);
+
+ SeqNo s(session, sequence + 1);
+ m_sequenceLog[prefix] = s;
+ m_syncLogic.addLocalNames (prefix, session, sequence);
+ return true;
+}
+
+void
+SyncSocket::fetchData(const string &prefix, const SeqNo &seq, const DataCallback& callback, int retry)
+{
+ ostringstream interestName;
+ interestName << prefix << "/" << seq.getSession() << "/" << seq.getSeq();
+ //std::cout << "Socket " << this << " Send Interest <" << interestName.str() << "> for raw data " << endl;
+
+
+ Ptr<ndn::Interest> interestPtr = Ptr<ndn::Interest>(new ndn::Interest(interestName.str()));
+ Ptr<Closure> closure = Ptr<Closure> (new Closure(callback,
+ boost::bind(&SyncSocket::onChatDataTimeout,
+ this,
+ _1,
+ _2,
+ retry),
+ boost::bind(&SyncSocket::onChatDataUnverified,
+ this,
+ _1)));
+ m_handler->sendInterest(interestPtr, closure);
+}
+
+void
+SyncSocket::onChatDataTimeout(Ptr<Closure> closure, Ptr<ndn::Interest> interest, int retry)
+{
+ if(retry > 0)
+ {
+ Ptr<Closure> newClosure = Ptr<Closure>(new Closure(closure->m_dataCallback,
+ boost::bind(&SyncSocket::onChatDataTimeout,
+ this,
+ _1,
+ _2,
+ retry - 1),
+ closure->m_unverifiedCallback,
+ closure->m_stepCount)
+ );
+ m_handler->sendInterest(interest, newClosure);
+ }
+}
+
+void
+SyncSocket::onChatDataUnverified(Ptr<Data> data)
+{}
+
+
+uint32_t
+SyncSocket::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;
+}
+
+}//Sync
diff --git a/src/ccnx/sync-socket.h b/src/ccnx/sync-socket.h
new file mode 100644
index 0000000..a4530ba
--- /dev/null
+++ b/src/ccnx/sync-socket.h
@@ -0,0 +1,111 @@
+/* -*- 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: Yingdi Yu <yingdi@cs.ucla.edu>
+ */
+
+#ifndef SYNC_SOCKET_H
+#define SYNC_SOCKET_H
+
+#include "sync-logic.h"
+#include <boost/function.hpp>
+#include <boost/unordered_map.hpp>
+#include "sync-seq-no.h"
+#include <ndn.cxx/wrapper/wrapper.h>
+#include <utility>
+#include <map>
+#include <vector>
+#include <sstream>
+
+namespace Sync {
+
+/**
+ * \ingroup sync
+ * @brief A simple interface to interact with client code
+ */
+class SyncSocket
+{
+public:
+ typedef boost::function< void (const std::vector<MissingDataInfo> &, SyncSocket * ) > 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
+ * parameter should be passed to the constructor of m_fetcher; furthermore,
+ * the fetch function of m_fetcher should be a second paramter passed to
+ * the constructor of m_syncAppWrapper, so that m_syncAppWrapper can tell
+ * m_fetcher to fetch the actual app data after it learns the names
+ *
+ * @param syncPrefix the name prefix for Sync Interest
+ * @param dataCallback the callback to process data
+ */
+ SyncSocket (const std::string &syncPrefix,
+ ndn::Ptr<SyncPolicyManager> syncPolicyManager,
+ NewDataCallback dataCallback,
+ RemoveCallback rmCallback);
+
+ ~SyncSocket ();
+
+ bool
+ publishData(const std::string &prefix, uint32_t session, const char *buf, size_t len, int freshness);
+
+ void
+ remove (const std::string &prefix)
+ { m_syncLogic.remove(prefix); }
+
+ void
+ fetchData(const std::string &prefix, const SeqNo &seq, const ndn::DataCallback& callback, int retry = 0);
+
+ std::string
+ getRootDigest()
+ { return m_syncLogic.getRootDigest(); }
+
+ uint32_t
+ getNextSeq (const std::string &prefix, uint32_t session);
+
+ SyncLogic &
+ getLogic ()
+ { return m_syncLogic; }
+
+ // make this a static function so we don't have to create socket instance without
+ // knowing the local prefix. it's a wrong place for this function anyway
+ static std::string
+ GetLocalPrefix ();
+
+private:
+ void
+ passCallback(const std::vector<MissingDataInfo> &v)
+ { m_newDataCallback(v, this); }
+
+ void
+ onChatDataTimeout(ndn::Ptr<ndn::Closure> closure, ndn::Ptr<ndn::Interest> interest, int retry);
+
+ void
+ onChatDataUnverified(ndn::Ptr<ndn::Data> data);
+
+private:
+ typedef boost::unordered_map<std::string, SeqNo> SequenceLog;
+ NewDataCallback m_newDataCallback;
+ SequenceLog m_sequenceLog;
+ ndn::Ptr<SyncPolicyManager> m_syncPolicyManager;
+ ndn::Ptr<ndn::Wrapper> m_handler;
+ SyncLogic m_syncLogic;
+};
+
+} // Sync
+
+#endif // SYNC_SOCKET_H