SunnyQjm/ChronoSync
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Remove outdated parts

Browse Source
This commit is contained in:
Alexander Afanasyev committed 2013-11-07 13:37:14 -08:00
1 parent 1baf6e084d
commit 559c37049c
10 files changed
-1556
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

disable/sync-app-socket-c.cc
-146
View File
@@ -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);
}
disable/sync-app-socket-c.h
-52
View File
@@ -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
disable/sync-app-socket.cc
-115
View File
@@ -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;
}
}
disable/sync-app-socket.h
-112
View File
@@ -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
disable/sync-ccnx-wrapper.cc
-597
View File
@@ -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);
}
}
}
disable/sync-ccnx-wrapper.h
-198
View File
@@ -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
disable/test_app_socket.cc
-227
View File
@@ -1,227 +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 <boost/test/unit_test.hpp>
#include <boost/test/output_test_stream.hpp>
using boost::test_tools::output_test_stream;
#include <boost/make_shared.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include "sync-logging.h"
#include "ccnx/sync-app-socket.h"
extern "C" {
#include <unistd.h>
}
using namespace Sync;
using namespace std;
using namespace boost;
INIT_LOGGER ("Test.AppSocket");
#define PRINT
//std::cout << "Line: " << __LINE__ << std::endl;
class TestSocketApp {
public:
map<string, string> data;
void set(string str1, string str2) {
// _LOG_FUNCTION (this << ", " << str1);
data.insert(make_pair(str1, str2));
// cout << str1 << ", " << str2 << endl;
}
void setNum(string str1, const char *buf, size_t len) {
int n = len / 4;
int *numbers = new int [n];
memcpy(numbers, buf, len);
for (int i = 0; i < n; i++) {
sum += numbers[i];
}
delete numbers;
}
int sum;
void fetchAll(const vector<MissingDataInfo> &v, SyncAppSocket *socket) {
int n = v.size();
PRINT
for (int i = 0; i < n; i++) {
for(SeqNo s = v[i].low; s <= v[i].high; ++s) {
//PRINT
socket->fetchString(v[i].prefix, s, bind(&TestSocketApp::set, this, _1, _2));
}
}
}
void fetchNumbers(const vector<MissingDataInfo> &v, SyncAppSocket *socket) {
int n = v.size();
PRINT
std::cout << "In fetchNumbers. size of v is: " << n << std::endl;
for (int i = 0; i < n; i++) {
std::cout << "In fetchNumbers. v[i].low is (" <<v[i].low.getSession() <<", " << v[i].low.getSeq() << ") v[i].high is ("<<v[i].high.getSession() <<", " <<v[i].high.getSeq()<<")" << std::endl;
for(SeqNo s = v[i].low; s <= v[i].high; ++s) {
PRINT
socket->fetchRaw(v[i].prefix, s, bind(&TestSocketApp::setNum, this, _1, _2, _3));
}
}
}
void pass(const string &prefix) {
}
string toString(){
map<string, string>::iterator it = data.begin();
string str = "\n";
for (; it != data.end(); ++it){
str += "<";
str += it->first;
str += "|";
str += it->second;
str += ">";
str += "\n";
}
return str;
}
};
BOOST_AUTO_TEST_CASE (AppSocketTest)
{
INIT_LOGGERS ();
TestSocketApp a1, a2, a3;
string syncPrefix("/let/us/sync");
string p1("/irl.cs.ucla.edu"), p2("/yakshi.org"), p3("/google.com");
_LOG_DEBUG ("s1");
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::fetchAll, &a2, _1, _2), bind(&TestSocketApp::pass, &a2, _1));
this_thread::sleep (posix_time::milliseconds (50));
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.publishString (p1, 0, data0, 10);
this_thread::sleep (posix_time::milliseconds (1000));
// from code logic, we won't be fetching our own data
a1.set(p1 + "/0/0", data0);
BOOST_CHECK_EQUAL(a1.toString(), a2.toString());
BOOST_CHECK_EQUAL(a2.toString(), a3.toString());
// single source, multiple data at once
string data1 = "Yes, give me that ketchup";
string data2 = "Don't look conspicuous, it draws fire";
_LOG_DEBUG ("s1 publish");
s1.publishString (p1, 0, data1, 10);
_LOG_DEBUG ("s1 publish");
s1.publishString (p1, 0, data2, 10);
this_thread::sleep (posix_time::milliseconds (1000));
// from code logic, we won't be fetching our own data
a1.set(p1 + "/0/1", data1);
a1.set(p1 + "/0/2", data2);
BOOST_CHECK_EQUAL(a1.toString(), a2.toString());
BOOST_CHECK_EQUAL(a2.toString(), a3.toString());
// another single source
string data3 = "You surf the Internet, I surf the real world";
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.publishString(p3, 0, data3, 10);
this_thread::sleep (posix_time::milliseconds (200));
// another single source, multiple data at once
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
a3.set(p3 + "/0/0", data3);
a2.set(p2 + "/0/0", data4);
a2.set(p2 + "/0/1", data5);
BOOST_CHECK_EQUAL(a1.toString(), a2.toString());
BOOST_CHECK_EQUAL(a2.toString(), a3.toString());
// not sure weither this is simultanous data generation from multiple sources
_LOG_DEBUG ("Simultaneous publishing");
string data6 = "Shakespeare says: 'Prose before hos.'";
string data7 = "Pick good people, talent never wears out";
s1.publishString(p1, 0, data6, 10);
// this_thread::sleep (posix_time::milliseconds (1000));
s2.publishString(p2, 0, data7, 10);
this_thread::sleep (posix_time::milliseconds (1500));
// from code logic, we won't be fetching our own data
a1.set(p1 + "/0/3", data6);
a2.set(p2 + "/0/2", data7);
// a1.set(p1 + "/0/1", data6);
// a2.set(p2 + "/0/0", data7);
BOOST_CHECK_EQUAL(a1.toString(), a2.toString());
BOOST_CHECK_EQUAL(a2.toString(), a3.toString());
_LOG_DEBUG("Begin new test");
std::cout << "Begin new Test " << std::endl;
string syncRawPrefix = "/this/is/the/prefix";
a1.sum = 0;
a2.sum = 0;
SyncAppSocket s4 (syncRawPrefix, bind(&TestSocketApp::fetchNumbers, &a1, _1, _2), bind(&TestSocketApp::pass, &a1, _1));
SyncAppSocket s5 (syncRawPrefix, bind(&TestSocketApp::fetchNumbers, &a2, _1, _2), bind(&TestSocketApp::pass, &a2, _1));
int num[5] = {0, 1, 2, 3, 4};
string p4 = "/xiaonei.com";
string p5 = "/mitbbs.com";
s4.publishRaw(p4, 0,(const char *) num, sizeof(num), 10);
a1.setNum(p4, (const char *) num, sizeof (num));
this_thread::sleep (posix_time::milliseconds (1000));
BOOST_CHECK(a1.sum == a2.sum && a1.sum == 10);
int newNum[5] = {9, 7, 2, 1, 1};
s5.publishRaw(p5, 0,(const char *) newNum, sizeof(newNum), 10);
a2.setNum(p5, (const char *)newNum, sizeof (newNum));
this_thread::sleep (posix_time::milliseconds (1000));
BOOST_CHECK_EQUAL(a1.sum, a2.sum);
BOOST_CHECK_EQUAL(a1.sum, 30);
_LOG_DEBUG ("Finish");
}
disable/test_ccnx_wrapper.cc
-109
View File
@@ -1,109 +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 <boost/test/unit_test.hpp>
#include <boost/test/output_test_stream.hpp>
using boost::test_tools::output_test_stream;
#include <boost/make_shared.hpp>
#include "ccnx/sync-ccnx-wrapper.h"
using namespace Sync;
using namespace std;
using namespace boost;
string echoStr = "";
void echo(string str) {
echoStr = str;
}
struct TestStruct {
string s_str1, s_str2;
void set(string str1, string str2) {
s_str1 = str1;
s_str2 = str2;
}
int * num;
int n;
void rawSet(string str1, const char *buf, size_t len) {
std::cout << "In rawSet" << std::endl;
s_str1 = str1;
n = len / 4;
num = new int [n];
memcpy(num, buf, len);
}
};
BOOST_AUTO_TEST_CASE (CcnxWrapperTest)
{
CcnxWrapper ha;
CcnxWrapper hb;
TestStruct foo;
boost::function<void (string)> globalFunc = echo;
boost::function<void (string, string)> memberFunc =
bind(&TestStruct::set, &foo, _1, _2);
boost::function<void (string, const char *, size_t)> rawFunc =
bind(&TestStruct::rawSet, &foo, _1, _2, _3);
string prefix = "/ucla.edu";
ha.setInterestFilter(prefix, globalFunc);
this_thread::sleep (posix_time::milliseconds (10));
string interest = "/ucla.edu/0";
hb.sendInterestForString(interest, memberFunc);
// give time for ccnd to react
sleep(1);
this_thread::sleep (posix_time::milliseconds (5));
BOOST_CHECK_EQUAL(echoStr, interest);
string name = "/ucla.edu/0";
string data = "random bits: !#$!@#$!";
ha.publishStringData(name, data, 5);
hb.sendInterestForString(interest, memberFunc);
// give time for ccnd to react
this_thread::sleep (posix_time::milliseconds (5));
BOOST_CHECK_EQUAL(foo.s_str1, name);
BOOST_CHECK_EQUAL(foo.s_str2, data);
string rawDataName = "/ucla.edu/1";
int num[5] = {0, 1, 2, 3, 4};
ha.publishRawData(rawDataName, (const char *)num, sizeof(num), 30);
hb.sendInterest(rawDataName, rawFunc);
this_thread::sleep (posix_time::milliseconds (5));
BOOST_CHECK_EQUAL(foo.s_str1, rawDataName);
for (int i = 0; i < 5; i++) {
BOOST_CHECK(foo.num[i] == num[i]);
}
}
src/ccnx/sync-socket.cc → src/sync-socket.cc
View File
File renamed without changes.
src/ccnx/sync-socket.h → src/sync-socket.h
View File
File renamed without changes.