diff --git a/libretroshare/src/dbase/Makefile b/libretroshare/src/dbase/Makefile index f2d6edb1f..2464b9495 100644 --- a/libretroshare/src/dbase/Makefile +++ b/libretroshare/src/dbase/Makefile @@ -10,9 +10,11 @@ include $(RS_TOP_DIR)/scripts/config.mk RSOBJ = findex.o fimonitor.o cachestrapper.o fistore.o \ rsexpr.o -TESTOBJ = fitest2.o fisavetest.o ficachetest.o searchtest.o +TESTOBJ = fitest2.o fisavetest.o searchtest.o +#ficachetest.o -TESTS = fitest2 fisavetest ficachetest searchtest +TESTS = fitest2 fisavetest searchtest +#ficachetest ifeq ($(OS),Linux) TESTOBJ += fimontest.o diff --git a/libretroshare/src/dbase/cachestrapper.cc b/libretroshare/src/dbase/cachestrapper.cc index e5aec4e46..8e83d11ac 100644 --- a/libretroshare/src/dbase/cachestrapper.cc +++ b/libretroshare/src/dbase/cachestrapper.cc @@ -22,14 +22,17 @@ */ #include "dbase/cachestrapper.h" +#include "serialiser/rsconfigitems.h" +#include "pqi/p3connmgr.h" +#include "util/rsdir.h" #include #include #include -/** - * #define CS_DEBUG 1 - */ +/***/ +#define CS_DEBUG 1 +/***/ bool operator<(const CacheId &a, const CacheId &b) { @@ -59,8 +62,8 @@ std::ostream &operator<<(std::ostream &out, const CacheData &d) * ********************************* Cache Store / Source *************************/ -CacheSource::CacheSource(uint16_t t, bool m, std::string cachedir) - :cacheType(t), multiCache(m), cacheDir(cachedir) +CacheSource::CacheSource(uint16_t t, bool m, CacheStrapper *cs, std::string cachedir) + :cacheType(t), multiCache(m), mStrapper(cs), cacheDir(cachedir) { return; } @@ -108,6 +111,9 @@ bool CacheSource::refreshCache(const CacheData &data) } unlockData(); /* UNLOCK MUTEX */ + + if (mStrapper) /* allow testing without full feedback */ + mStrapper->refreshCache(data); return ret; } @@ -189,8 +195,10 @@ void CacheSource::listCaches(std::ostream &out) } -CacheStore::CacheStore(uint16_t t, bool m, CacheTransfer *cft, std::string cachedir) - :cacheType(t), multiCache(m), cacheTransfer(cft), cacheDir(cachedir) +CacheStore::CacheStore(uint16_t t, bool m, + CacheStrapper *cs, CacheTransfer *cft, std::string cachedir) + :cacheType(t), multiCache(m), mStrapper(cs), + cacheTransfer(cft), cacheDir(cachedir) { /* not much */ return; @@ -292,6 +300,28 @@ bool CacheStore::locked_getStoredCache(CacheData &data) +bool CacheStore::getAllStoredCaches(std::list &data) +{ + lockData(); /* LOCK MUTEX */ + + std::map::iterator pit; + for(pit = caches.begin(); pit != caches.end(); pit++) + { + CacheSet::iterator cit; + /* look for subid */ + for(cit = (pit->second).begin(); + cit != (pit->second).end(); cit++) + { + data.push_back(cit->second); + } + } + + unlockData(); /* UNLOCK MUTEX */ + + return true; +} + + /* input from CacheStrapper. * check if we want to download it... * determine the new name/path @@ -438,6 +468,12 @@ void CacheStore::locked_storeCacheEntry(const CacheData &data) { (pit->second)[0] = data; } + + /* tell the strapper we've loaded one */ + if (mStrapper) + { + mStrapper->refreshCacheStore(data); + } return; } @@ -447,11 +483,9 @@ void CacheStore::locked_storeCacheEntry(const CacheData &data) * ********************************* CacheStrapper ********************************/ -CacheStrapper::CacheStrapper(RsPeerId id, time_t period) - :ownId(id), queryPeriod(period) +CacheStrapper::CacheStrapper(p3AuthMgr *am, p3ConnectMgr *cm) + :p3Config(CONFIG_TYPE_CACHE), mAuthMgr(am), mConnMgr(cm) { - /* add OwnId */ - addPeerId(ownId); return; } @@ -467,12 +501,22 @@ void CacheStrapper::addCachePair(CachePair set) void CacheStrapper::statusChange(const std::list &plist) { std::list::const_iterator it; - std::map::iterator mit; for(it = plist.begin(); it != plist.end(); it++) { - if (status.end() == (mit = status.find(it->id))) + if (it->actions & RS_PEER_CONNECTED) { - addPeerId(it->id); + /* grab all the cache ids and add */ + + std::map hashs; + std::map::iterator cit; + + handleCacheQuery(it->id, hashs); + + RsStackMutex stack(csMtx); /******* LOCK STACK MUTEX *********/ + for(cit = hashs.begin(); cit != hashs.end(); cit++) + { + mCacheUpdates.push_back(std::make_pair(it->id, cit->second)); + } } } } @@ -480,50 +524,51 @@ void CacheStrapper::statusChange(const std::list &plist) /**************** from pqimonclient ********************/ -void CacheStrapper::addPeerId(RsPeerId pid) +void CacheStrapper::refreshCache(const CacheData &data) { - std::map::iterator it; + /* we've received an update + * send to all online peers + self + */ - /* just reset it for the moment */ - CacheTS ts; - ts.query = 0; - ts.answer = 0; + std::list ids; + std::list::iterator it; - status[pid] = ts; -} + mConnMgr->getOnlineList(ids); -bool CacheStrapper::removePeerId(RsPeerId pid) -{ - std::map::iterator it; - if (status.end() != (it = status.find(pid))) + RsStackMutex stack(csMtx); /******* LOCK STACK MUTEX *********/ + for(it = ids.begin(); it != ids.end(); it++) { - status.erase(it); - return true; + mCacheUpdates.push_back(std::make_pair(*it, data)); } - return false; + + mCacheUpdates.push_back(std::make_pair(mConnMgr->getOwnId(), data)); + + IndicateConfigChanged(); /**** INDICATE MSG CONFIG CHANGED! *****/ } +void CacheStrapper::refreshCacheStore(const CacheData &data) +{ + + /* indicate to save data */ + IndicateConfigChanged(); /**** INDICATE MSG CONFIG CHANGED! *****/ + +} + +bool CacheStrapper::getCacheUpdates(std::list > &updates) +{ + RsStackMutex stack(csMtx); /******* LOCK STACK MUTEX *********/ + updates = mCacheUpdates; + mCacheUpdates.clear(); + + return true; +} + + + /* pass to correct CacheSet */ void CacheStrapper::recvCacheResponse(CacheData &data, time_t ts) { - /* update internal data first */ - std::map::iterator it; - if (status.end() == status.find(data.pid)) - { - /* add it in */ - CacheTS d; - d.query = 0; - d.answer = 0; - - status[data.pid] = d; - } - - it = status.find(data.pid); /* will always succeed */ - - /* update status */ - (it -> second).answer = ts; - /* find cache store */ std::map::iterator it2; if (caches.end() == (it2 = caches.find(data.cid.type))) @@ -534,10 +579,12 @@ void CacheStrapper::recvCacheResponse(CacheData &data, time_t ts) /* notify the CacheStore */ (it2 -> second).store -> availableCache(data); + } /* generate periodically or at a change */ +#if 0 bool CacheStrapper::sendCacheQuery(std::list &id, time_t ts) { /* iterate through peers, and see who we haven't got an answer from recently */ @@ -553,6 +600,7 @@ bool CacheStrapper::sendCacheQuery(std::list &id, time_t ts) } return (id.size() > 0); } +#endif void CacheStrapper::handleCacheQuery(RsPeerId id, std::map &hashs) @@ -575,8 +623,8 @@ void CacheStrapper::listCaches(std::ostream &out) { /* can overwrite for more control! */ std::map::iterator it; - out << "CacheStrapper::listCaches() [" << ownId; - out << "] Total Peers: " << status.size() << " Total Caches: " << caches.size(); + out << "CacheStrapper::listCaches() [" << mConnMgr->getOwnId(); + out << "] " << " Total Caches: " << caches.size(); out << std::endl; for(it = caches.begin(); it != caches.end(); it++) { @@ -592,6 +640,7 @@ void CacheStrapper::listCaches(std::ostream &out) void CacheStrapper::listPeerStatus(std::ostream &out) { +#if 0 std::map::iterator it; out << "CacheStrapper::listPeerStatus() [" << ownId; out << "] Total Peers: " << status.size() << " Total Caches: " << caches.size(); @@ -604,6 +653,7 @@ void CacheStrapper::listPeerStatus(std::ostream &out) out << std::endl; } return; +#endif } @@ -620,6 +670,256 @@ bool CacheStrapper::findCache(std::string hash, CacheData &data) } return false; } + + + +/***************************************************************************/ +/****************************** CONFIGURATION HANDLING *********************/ +/***************************************************************************/ + +/**** OVERLOADED FROM p3Config ****/ + +RsSerialiser *CacheStrapper::setupSerialiser() +{ + RsSerialiser *rss = new RsSerialiser(); + + /* add in the types we need! */ + rss->addSerialType(new RsCacheConfigSerialiser()); + + return rss; +} + + +std::list CacheStrapper::saveList(bool &cleanup) +{ + std::list saveData; + + /* it can delete them! */ + cleanup = true; + +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::saveList()" << std::endl; +#endif + + /* iterate through the Caches (local first) */ + + std::list::iterator cit; + std::list ownCaches; + std::list remoteCaches; + std::string ownId = mConnMgr->getOwnId(); + + std::map::iterator it; + for(it = caches.begin(); it != caches.end(); it++) + { + std::map::iterator tit; + std::map ownTmp; + (it->second).source -> cachesAvailable(ownId, ownTmp); + (it->second).store -> getAllStoredCaches(remoteCaches); + + for(tit = ownTmp.begin(); tit != ownTmp.end(); tit++) + { + ownCaches.push_back(tit->second); + } + } + + for(cit = ownCaches.begin(); cit != ownCaches.end(); cit++) + { + RsCacheConfig *rscc = new RsCacheConfig(); + + rscc->pid = cit->pid; + //rscc->pname = cit->pname; + rscc->cachetypeid = cit->cid.type; + rscc->cachesubid = cit->cid.subid; + rscc->path = cit->path; + rscc->name = cit->name; + rscc->hash = cit->hash; + rscc->size = cit->size; + rscc->recvd = cit->recvd; + + saveData.push_back(rscc); + } + + for(cit = remoteCaches.begin(); cit != remoteCaches.end(); cit++) + { + if (cit->pid == ownId) + { +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::loadList() discarding Own Remote Cache"; + std::cerr << std::endl; +#endif + continue; /* skip own caches -> will get transferred anyway */ + } + + RsCacheConfig *rscc = new RsCacheConfig(); + + rscc->pid = cit->pid; + //rscc->pname = cit->pname; + rscc->cachetypeid = cit->cid.type; + rscc->cachesubid = cit->cid.subid; + rscc->path = cit->path; + rscc->name = cit->name; + rscc->hash = cit->hash; + rscc->size = cit->size; + rscc->recvd = cit->recvd; + + saveData.push_back(rscc); + } + + /* list completed! */ + return saveData; +} + + +bool CacheStrapper::loadList(std::list load) +{ + std::list::iterator it; + RsCacheConfig *rscc; + +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::loadList() Item Count: " << load.size(); + std::cerr << std::endl; +#endif + std::list ownCaches; + std::list remoteCaches; + std::string ownId = mConnMgr->getOwnId(); + + std::map > saveFiles; + std::map >::iterator sit; + + for(it = load.begin(); it != load.end(); it++) + { + /* switch on type */ + if (NULL != (rscc = dynamic_cast(*it))) + { +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::loadList() Item: "; + std::cerr << std::endl; + rscc->print(std::cerr, 10); + std::cerr << std::endl; +#endif + CacheData cd; + + cd.pid = rscc->pid; + cd.pname = mAuthMgr->getName(cd.pid); + cd.cid.type = rscc->cachetypeid; + cd.cid.subid = rscc->cachesubid; + cd.path = rscc->path; + cd.name = rscc->name; + cd.hash = rscc->hash; + cd.size = rscc->size; + cd.recvd = rscc->recvd; + + /* store files that we want to keep */ + (saveFiles[cd.path]).push_back(cd.name); + + std::map::iterator it2; + if (caches.end() == (it2 = caches.find(cd.cid.type))) + { + /* error - don't have this type of cache */ +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::loadList() Can't Find Cache discarding"; + std::cerr << std::endl; +#endif + } + else + { + if (cd.pid == ownId) + { + /* load local */ + (it2 -> second).source -> loadLocalCache(cd); +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::loadList() loaded Local"; + std::cerr << std::endl; +#endif + } + else + { + /* load remote */ + (it2 -> second).store -> loadCache(cd); +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::loadList() loaded Remote"; + std::cerr << std::endl; +#endif + } + } + + /* cleanup */ + delete (*it); + + } + else + { + /* cleanup */ + delete (*it); + } + } + + /* assemble a list of dirs to clean (union of cache dirs) */ + std::list cacheDirs; + std::list::iterator dit, fit; + std::map::iterator cit; + for(cit = caches.begin(); cit != caches.end(); cit++) + { + std::string lcdir = (cit->second).source->getCacheDir(); + std::string rcdir = (cit->second).store->getCacheDir(); + + if (cacheDirs.end() == std::find(cacheDirs.begin(), cacheDirs.end(), lcdir)) + { + cacheDirs.push_back(lcdir); + } + + if (cacheDirs.end() == std::find(cacheDirs.begin(), cacheDirs.end(), rcdir)) + { + cacheDirs.push_back(rcdir); + } + } + +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::loadList() Files To Save:" << std::endl; +#endif + + for(sit = saveFiles.begin(); sit != saveFiles.end(); sit++) + { +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::loadList() Files To Save in dir: <" << sit->first << ">" << std::endl; +#endif + for(fit = (sit->second).begin(); fit != (sit->second).end(); fit++) + { +#ifdef CS_DEBUG + std::cerr << "\tFile: " << *fit << std::endl; +#endif + } + } + + std::list emptyList; + for(dit = cacheDirs.begin(); dit != cacheDirs.end(); dit++) + { +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::loadList() Cleaning cache dir: <" << *dit << ">" << std::endl; +#endif + sit = saveFiles.find(*dit); + if (sit != saveFiles.end()) + { + for(fit = (sit->second).begin(); fit != (sit->second).end(); fit++) + { +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::loadList() Keeping File: " << *fit << std::endl; +#endif + } + RsDirUtil::cleanupDirectory(*dit, sit->second); + } + else + { +#ifdef CS_DEBUG + std::cerr << "CacheStrapper::loadList() No Files to save here!" << std::endl; +#endif + RsDirUtil::cleanupDirectory(*dit, emptyList); + } + } + + return true; + +} /********************************* CacheStrapper ********************************* @@ -631,6 +931,43 @@ bool CacheStrapper::findCache(std::string hash, CacheData &data) /* request from CacheStore */ bool CacheTransfer::RequestCache(CacheData &data, CacheStore *cbStore) { + /* check for a previous request -> and cancel + * + * - if duplicate pid, cid -> cancel old transfer + * - if duplicate hash -> Fail Transfer + */ + + std::map::iterator dit; + std::map::iterator sit; + + for(dit = cbData.begin(); dit != cbData.end(); dit++) + { + if (((dit->second).pid == data.pid) && + ((dit->second).cid.type == data.cid.type) && + ((dit->second).cid.subid == data.cid.subid)) + { + /* cancel old transfer */ + CancelCacheFile(dit->second.pid, dit->second.path, + dit->second.hash, dit->second.size); + + sit = cbStores.find(dit->second.hash); + cbData.erase(dit); + cbStores.erase(sit); + + break; + } + } + + /* find in store.... */ + sit = cbStores.find(data.hash); + if (sit != cbStores.end()) + { + /* Duplicate Current Request */ + cbStore -> failedCache(data); + return false; + } + + /* store request */ cbData[data.hash] = data; cbStores[data.hash] = cbStore; @@ -657,6 +994,18 @@ bool CacheTransfer::RequestCacheFile(RsPeerId id, std::string path, std::string return true; } +/* to be overloaded */ +bool CacheTransfer::CancelCacheFile(RsPeerId id, std::string path, std::string hash, uint64_t size) +{ + std::cerr << "CacheTransfer::CancelCacheFile() : from:" << id << " #"; + std::cerr << hash << " size: " << size; + std::cerr << " savepath: " << path << std::endl; + std::cerr << "CacheTransfer::CancelCacheFile() Dummy fn"; + std::cerr << std::endl; + + return true; +} + /* internal completion -> does cb */ bool CacheTransfer::CompletedCache(std::string hash) diff --git a/libretroshare/src/dbase/cachestrapper.h b/libretroshare/src/dbase/cachestrapper.h index 8ef9856a7..5fa9cf69f 100644 --- a/libretroshare/src/dbase/cachestrapper.h +++ b/libretroshare/src/dbase/cachestrapper.h @@ -113,6 +113,7 @@ bool RequestCache(CacheData &data, CacheStore *cbStore); /* request from CacheSt protected: /* to be overloaded */ virtual bool RequestCacheFile(RsPeerId id, std::string path, std::string hash, uint64_t size); +virtual bool CancelCacheFile(RsPeerId id, std::string path, std::string hash, uint64_t size); bool CompletedCache(std::string hash); /* internal completion -> does cb */ bool FailedCache(std::string hash); /* internal completion -> does cb */ @@ -134,7 +135,7 @@ typedef std::map CacheSet; class CacheSource { public: - CacheSource(uint16_t t, bool m, std::string cachedir); + CacheSource(uint16_t t, bool m, CacheStrapper *cs, std::string cachedir); virtual ~CacheSource() {} /* called to determine available cache for peer - @@ -165,8 +166,11 @@ bool findCache(std::string hash, CacheData &data); protected: - /*** MUTEX LOCKING - TODO - */ + uint16_t cacheType; /* for checking */ + bool multiCache; /* do we care about subid's */ + CacheStrapper *mStrapper; + + /*** MUTEX LOCKING */ void lockData(); void unlockData(); @@ -174,9 +178,6 @@ void unlockData(); private: - uint16_t cacheType; /* for checking */ - bool multiCache; /* do we care about subid's */ - std::string cacheDir; RsMutex cMutex; }; @@ -186,11 +187,12 @@ class CacheStore { public: - CacheStore(uint16_t t, bool m, CacheTransfer *cft, std::string cachedir); + CacheStore(uint16_t t, bool m, CacheStrapper *cs, CacheTransfer *cft, std::string cachedir); virtual ~CacheStore() {} /* current stored data */ bool getStoredCache(CacheData &data); /* use pid/cid in data */ +bool getAllStoredCaches(std::list &data); /* use pid/cid in data */ /* input from CacheStrapper -> store can then download new data */ void availableCache(const CacheData &data); @@ -232,12 +234,15 @@ bool locked_getStoredCache(CacheData &data); uint16_t cacheType; /* for checking */ bool multiCache; /* do we care about subid's */ + CacheStrapper *mStrapper; CacheTransfer *cacheTransfer; + std::string cacheDir; + RsMutex cMutex; + std::map caches; - RsMutex cMutex; }; @@ -263,54 +268,62 @@ class CachePair bool operator<(const CachePair &a, const CachePair &b); -class CacheTS -{ - public: - - time_t query; - time_t answer; -}; - #include "pqi/pqimonitor.h" +#include "pqi/p3cfgmgr.h" -class CacheStrapper: public pqiMonitor +class p3AuthMgr; + +class CacheStrapper: public pqiMonitor, public p3Config { public: - CacheStrapper(RsPeerId id, time_t period); + CacheStrapper(p3AuthMgr *am, p3ConnectMgr *cm); virtual ~CacheStrapper() { return; } /************* from pqiMonitor *******************/ virtual void statusChange(const std::list &plist); /************* from pqiMonitor *******************/ + /* Feedback from CacheSources */ +void refreshCache(const CacheData &data); +void refreshCacheStore(const CacheData &data); + + /* list of Caches to send out */ +bool getCacheUpdates(std::list > &updates); + void addCachePair(CachePair pair); -void addPeerId(RsPeerId pid); -bool removePeerId(RsPeerId pid); - - /*** I/O (1) ***/ - /* pass to correct CacheSet */ -void recvCacheResponse(CacheData &date, time_t ts); - /* generate periodically or at a change */ -bool sendCacheQuery(std::list &id, time_t ts); - /*** I/O (2) ***/ - /* handle a DirQuery */ +void recvCacheResponse(CacheData &data, time_t ts); void handleCacheQuery(RsPeerId id, std::map &data); + /* search through CacheSources. */ bool findCache(std::string hash, CacheData &data); /* display */ void listCaches(std::ostream &out); void listPeerStatus(std::ostream &out); - + + + /* Config */ + protected: + + /* Key Functions to be overloaded for Full Configuration */ +virtual RsSerialiser *setupSerialiser(); +virtual std::list saveList(bool &cleanup); +virtual bool loadList(std::list load); + private: - std::map status; + /* these are static - so shouldn't need mutex */ + p3AuthMgr *mAuthMgr; + p3ConnectMgr *mConnMgr; + std::map caches; - RsPeerId ownId; - time_t queryPeriod; + + RsMutex csMtx; /* protect below */ + + std::list > mCacheUpdates; }; diff --git a/libretroshare/src/dbase/cachetest.h b/libretroshare/src/dbase/cachetest.h index 3a28ed2ae..59d55b5be 100644 --- a/libretroshare/src/dbase/cachetest.h +++ b/libretroshare/src/dbase/cachetest.h @@ -32,8 +32,8 @@ class CacheTestSource: public CacheSource { public: - CacheTestSource(std::string dir) - :CacheSource(TESTID, false, dir) { return; } + CacheTestSource(CacheStrapper *cs, std::string dir) + :CacheSource(cs, TESTID, false, dir) { return; } }; class CacheTestStore: public CacheStore @@ -47,8 +47,8 @@ class CacheTestStore: public CacheStore class CacheTestMultiSource: public CacheSource { public: - CacheTestMultiSource(std::string dir) - :CacheSource(TESTID2, true, dir) { return; } + CacheTestMultiSource(CacheStrapper *cs, std::string dir) + :CacheSource(cs, TESTID2, true, dir) { return; } }; class CacheTestMultiStore: public CacheStore diff --git a/libretroshare/src/dbase/ficachetest.cc b/libretroshare/src/dbase/ficachetest.cc index c795ff95c..547776cfa 100644 --- a/libretroshare/src/dbase/ficachetest.cc +++ b/libretroshare/src/dbase/ficachetest.cc @@ -52,9 +52,13 @@ int main(int argc, char **argv) RsPeerId pid2("0x0102"); RsPeerId pid3("0x0103"); - CacheStrapper sc1(pid1, period); - CacheStrapper sc2(pid2, period); - CacheStrapper sc3(pid3, period); + p3ConnectMgr *connMgr1 = NULL; + p3ConnectMgr *connMgr2 = NULL; + p3ConnectMgr *connMgr3 = NULL; + + CacheStrapper sc1(connMgr1); + CacheStrapper sc2(connMgr2); + CacheStrapper sc3(connMgr3); CacheTransfer ctt1(&sc1); CacheTransfer ctt2(&sc2); CacheTransfer ctt3(&sc3); @@ -67,15 +71,15 @@ int main(int argc, char **argv) std::string nulldir = ""; - CacheSource *csrc1 = new CacheTestSource(nulldir); + CacheSource *csrc1 = new CacheTestSource(&sc1, nulldir); CacheStore *cstore1 = new CacheTestStore(&ctt1, nulldir); CacheId cid1(TESTID, 0); - CacheSource *csrc2 = new CacheTestSource(nulldir); + CacheSource *csrc2 = new CacheTestSource(&sc2, nulldir); CacheStore *cstore2 = new CacheTestStore(&ctt2, nulldir); CacheId cid2(TESTID, 0); - CacheSource *csrc3 = new CacheTestSource(nulldir); + CacheSource *csrc3 = new CacheTestSource(&sc3, nulldir); CacheStore *cstore3 = new CacheTestStore(&ctt3, nulldir); CacheId cid3(TESTID, 0); @@ -87,12 +91,6 @@ int main(int argc, char **argv) sc2.addCachePair(cp2); sc3.addCachePair(cp3); - - sc1.addPeerId(pid2); - sc2.addPeerId(pid1); - sc2.addPeerId(pid3); - sc3.addPeerId(pid2); - /* add in a cache to sc2 */ CacheData cdata; diff --git a/libretroshare/src/dbase/fimonitor.cc b/libretroshare/src/dbase/fimonitor.cc index 8e687247d..4f77bac59 100644 --- a/libretroshare/src/dbase/fimonitor.cc +++ b/libretroshare/src/dbase/fimonitor.cc @@ -38,8 +38,8 @@ #define FIM_DEBUG 1 -FileIndexMonitor::FileIndexMonitor(std::string cachedir, std::string pid) - :CacheSource(RS_SERVICE_TYPE_FILE_INDEX, false, cachedir), fi(pid), +FileIndexMonitor::FileIndexMonitor(CacheStrapper *cs, std::string cachedir, std::string pid) + :CacheSource(RS_SERVICE_TYPE_FILE_INDEX, false, cs, cachedir), fi(pid), pendingDirs(false), pendingForceCacheWrite(false) { diff --git a/libretroshare/src/dbase/fimonitor.h b/libretroshare/src/dbase/fimonitor.h index 34f2c440b..4d97afe06 100644 --- a/libretroshare/src/dbase/fimonitor.h +++ b/libretroshare/src/dbase/fimonitor.h @@ -68,7 +68,7 @@ std::string FileIndexMonitor::findRealRoot(std::string base); class FileIndexMonitor: public CacheSource, public RsThread { public: - FileIndexMonitor(std::string cachedir, std::string pid); + FileIndexMonitor(CacheStrapper *cs, std::string cachedir, std::string pid); virtual ~FileIndexMonitor(); /* external interface for filetransfer */ diff --git a/libretroshare/src/dbase/fimontest.cc b/libretroshare/src/dbase/fimontest.cc index 19d1b3bc8..4db1b8766 100644 --- a/libretroshare/src/dbase/fimontest.cc +++ b/libretroshare/src/dbase/fimontest.cc @@ -72,7 +72,7 @@ int main(int argc, char **argv) sleep(1); - FileIndexMonitor mon("", "OWN ID"); + FileIndexMonitor mon(NULL, "", "OWN ID"); /* setup monitor */ mon.setPeriod(period); diff --git a/libretroshare/src/dbase/fistore.cc b/libretroshare/src/dbase/fistore.cc index 05229abec..e3ab74acb 100644 --- a/libretroshare/src/dbase/fistore.cc +++ b/libretroshare/src/dbase/fistore.cc @@ -25,9 +25,9 @@ #include "rsiface/rsexpr.h" #include "serialiser/rsserviceids.h" -FileIndexStore::FileIndexStore(CacheTransfer *cft, +FileIndexStore::FileIndexStore(CacheStrapper *cs, CacheTransfer *cft, NotifyBase *cb_in, RsPeerId ownid, std::string cachedir) - :CacheStore(RS_SERVICE_TYPE_FILE_INDEX, false, cft, cachedir), + :CacheStore(RS_SERVICE_TYPE_FILE_INDEX, false, cs, cft, cachedir), localId(ownid), localindex(NULL), cb(cb_in) { return; diff --git a/libretroshare/src/dbase/fistore.h b/libretroshare/src/dbase/fistore.h index 8374e7640..d32e85121 100644 --- a/libretroshare/src/dbase/fistore.h +++ b/libretroshare/src/dbase/fistore.h @@ -63,7 +63,7 @@ class FileIndexStore: public CacheStore { public: - FileIndexStore(CacheTransfer *cft, NotifyBase *cb_in, + FileIndexStore(CacheStrapper *cs, CacheTransfer *cft, NotifyBase *cb_in, RsPeerId ownid, std::string cachedir); virtual ~FileIndexStore(); diff --git a/libretroshare/src/pqi/p3cfgmgr.h b/libretroshare/src/pqi/p3cfgmgr.h index 1eae01f37..dc860d35a 100644 --- a/libretroshare/src/pqi/p3cfgmgr.h +++ b/libretroshare/src/pqi/p3cfgmgr.h @@ -62,6 +62,7 @@ const uint32_t CONFIG_TYPE_GENERAL = 0x0001; const uint32_t CONFIG_TYPE_PEERS = 0x0002; const uint32_t CONFIG_TYPE_FSERVER = 0x0003; const uint32_t CONFIG_TYPE_MSGS = 0x0004; +const uint32_t CONFIG_TYPE_CACHE = 0x0005; class p3ConfigMgr; class p3AuthMgr; diff --git a/libretroshare/src/rsserver/p3face-startup.cc b/libretroshare/src/rsserver/p3face-startup.cc index 95c4bcc31..68a618a6c 100644 --- a/libretroshare/src/rsserver/p3face-startup.cc +++ b/libretroshare/src/rsserver/p3face-startup.cc @@ -474,18 +474,15 @@ int RsServer::StartupRetroShare(RsInit *config) ((AuthXPGP *) mAuthMgr) -> loadCertificates(oldFormat, oldConfigMap); - - /**************************************************************************/ /* setup classes / structures */ /**************************************************************************/ - uint32_t queryPeriod = 60; /* query every 1 minutes -> change later to 600+ */ mConnMgr = new p3ConnectMgr(mAuthMgr); p3UpnpMgr *mUpnpMgr = new upnphandler(); p3DhtMgr *mDhtMgr = new OpenDHTMgr(ownId, mConnMgr); - CacheStrapper *mCacheStrapper = new CacheStrapper(ownId, queryPeriod); + CacheStrapper *mCacheStrapper = new CacheStrapper(mAuthMgr, mConnMgr); ftfiler *mCacheTransfer = new ftfiler(mCacheStrapper); SecurityPolicy *none = secpolicy_create(); @@ -497,14 +494,11 @@ int RsServer::StartupRetroShare(RsInit *config) server->setConfigDir(config->basedir.c_str()); server->setSaveDir(config->homePath.c_str()); /* Default Save Dir - config will overwrite */ server->setSearchInterface(pqih, mAuthMgr, mConnMgr); + server->setFileCallback(ownId, mCacheStrapper, mCacheTransfer, &(getNotify())); mConfigMgr = new p3ConfigMgr(mAuthMgr, config->basedir, "rs-v0.4.cfg", "rs-v0.4.sgn"); mGeneralConfig = new p3GeneralConfig(); - - // Setup Peer Interface. - rsPeers = new p3Peers(mConnMgr, mAuthMgr); - /* create Services */ ad = new p3disc(mAuthMgr, mConnMgr); msgSrv = new p3MsgService(mConnMgr); @@ -516,13 +510,25 @@ int RsServer::StartupRetroShare(RsInit *config) pqih -> addService(chatSrv); pqih -> addService(gameLauncher); - /* so need to Monitor too! */ + /* create Cache Services */ + std::string config_dir = config->basedir; + std::string localcachedir = config_dir + "/cache/local"; + std::string remotecachedir = config_dir + "/cache/remote"; + + mRanking = new p3Ranking(RS_SERVICE_TYPE_RANK, + mCacheStrapper, mCacheTransfer, + localcachedir, remotecachedir, 3600 * 24 * 30); + + CachePair cp(mRanking, mRanking, CacheId(RS_SERVICE_TYPE_RANK, 0)); + mCacheStrapper -> addCachePair(cp); /**************************************************************************/ + mConnMgr->setDhtMgr(mDhtMgr); mConnMgr->setUpnpMgr(mUpnpMgr); /**************************************************************************/ + /* need to Monitor too! */ mConnMgr->addMonitor(pqih); mConnMgr->addMonitor(mCacheStrapper); @@ -535,7 +541,8 @@ int RsServer::StartupRetroShare(RsInit *config) mConfigMgr->addConfiguration("peers.cfg", mConnMgr); mConfigMgr->addConfiguration("general.cfg", mGeneralConfig); mConfigMgr->addConfiguration("msgs.cfg", msgSrv); - + mConfigMgr->addConfiguration("cache.cfg", mCacheStrapper); + /**************************************************************************/ @@ -570,6 +577,9 @@ int RsServer::StartupRetroShare(RsInit *config) /* trigger generalConfig loading for classes that require it */ /**************************************************************************/ + pqih->setConfig(mGeneralConfig); + + pqih->load_config(); /**************************************************************************/ /* Force Any Configuration before Startup (After Load) */ @@ -621,29 +631,7 @@ int RsServer::StartupRetroShare(RsInit *config) /* Force Any Last Configuration Options */ /**************************************************************************/ - - - - - - - - - - - - /****************** setup new stuff ***************/ - - pqih->load_config(); - - /* Must be after server->setSearchInterface() - * and. - * Must be before other Caches are added to the Strapper! - * */ - - server->setFileCallback(ownId, mCacheStrapper, mCacheTransfer, &(getNotify())); - - + server->StartupMonitor(); #ifdef PQI_USE_CHANNELS server->setP3Channel(pqih->getP3Channel()); @@ -660,20 +648,9 @@ int RsServer::StartupRetroShare(RsInit *config) pqih->AddSearchModule(mod); + /* Setup GUI Interfaces. */ - - /* create Cache Services */ - std::string config_dir = config->basedir; - std::string localcachedir = config_dir + "/cache/local"; - std::string remotecachedir = config_dir + "/cache/remote"; - - mRanking = new p3Ranking(RS_SERVICE_TYPE_RANK, - mCacheTransfer, localcachedir, remotecachedir, 3600 * 24 * 30); - - CachePair cp(mRanking, mRanking, CacheId(RS_SERVICE_TYPE_RANK, 0)); - mCacheStrapper -> addCachePair(cp); - - /* setup the gui */ + rsPeers = new p3Peers(mConnMgr, mAuthMgr); rsGameLauncher = gameLauncher; rsRanks = new p3Rank(mRanking); rsMsgs = new p3Msgs(mAuthMgr, msgSrv, chatSrv); diff --git a/libretroshare/src/serialiser/rsconfigitems.cc b/libretroshare/src/serialiser/rsconfigitems.cc index 817b63e09..774f07871 100644 --- a/libretroshare/src/serialiser/rsconfigitems.cc +++ b/libretroshare/src/serialiser/rsconfigitems.cc @@ -843,7 +843,7 @@ RsPeerStunItem *RsPeerConfigSerialiser::deserialiseStun(void *data, uint32_t *si } -/****************************************************************************/ +/****************************************************************************/ RsCacheConfig::~RsCacheConfig() @@ -851,43 +851,51 @@ RsCacheConfig::~RsCacheConfig() return; } -void RsCacheConfig::clear() -{ - - cacheid = 0; - path = ""; - name = ""; - hash = ""; - recvd = 0; - -} - -std::ostream &RsCacheConfig::print(std::ostream &out, uint16_t indent) -{ - printRsItemBase(out, "RsCacheConfig", indent); // begin 'WRITE' check - uint16_t int_Indent = indent + 2; - - printIndent(out, int_Indent); //indent - out << "cacheid: " << cacheid << std::endl; // display value of cacheid - - printIndent(out, int_Indent); - out << "path: " << path << std::endl; // display value of path - - printIndent(out, int_Indent); - out << "name: " << name << std::endl; // display value of name - - printIndent(out, int_Indent); - out << "hash: " << hash << std::endl; // display value of hash - - printIndent(out, int_Indent); - out << "recvd: " << recvd << std::endl; // display value of recvd - - printRsItemEnd(out, "RsCacheConfig", indent); // end of 'WRITE' check - return out; -} - -/**************************************************************************/ - +void RsCacheConfig::clear() +{ + pid.clear(); + cachetypeid = 0; + cachesubid = 0; + path = ""; + name = ""; + hash = ""; + size = 0; + recvd = 0; + +} + +std::ostream &RsCacheConfig::print(std::ostream &out, uint16_t indent) +{ + printRsItemBase(out, "RsCacheConfig", indent); + uint16_t int_Indent = indent + 2; + + printIndent(out, int_Indent); //indent + out << "pid: " << pid << std::endl; // display value of peerid + + printIndent(out, int_Indent); //indent + out << "cacheid: " << cachetypeid << ":" << cachesubid << std::endl; // display value of cacheid + + printIndent(out, int_Indent); + out << "path: " << path << std::endl; // display value of path + + printIndent(out, int_Indent); + out << "name: " << name << std::endl; // display value of name + + printIndent(out, int_Indent); + out << "hash: " << hash << std::endl; // display value of hash + + printIndent(out, int_Indent); + out << "size: " << size << std::endl; // display value of size + + printIndent(out, int_Indent); + out << "recvd: " << recvd << std::endl; // display value of recvd + + printRsItemEnd(out, "RsCacheConfig", indent); // end of 'WRITE' check + return out; +} + +/**************************************************************************/ + RsCacheConfigSerialiser::~RsCacheConfigSerialiser() { @@ -895,140 +903,148 @@ RsCacheConfigSerialiser::~RsCacheConfigSerialiser() } uint32_t RsCacheConfigSerialiser::size(RsItem *i) -{ +{ RsCacheConfig *item = (RsCacheConfig *) i; - - uint32_t s = 8; // to store calculated size, initiailize with size of header - - s += 2; /* cacheid */ - s += GetTlvStringSize(item->path); - s += GetTlvStringSize(item->name); - s += GetTlvStringSize(item->hash); - s += 2; /* recvd */ - - return s; -} - -bool RsCacheConfigSerialiser::serialise(RsItem *i, void *data, uint32_t *size) -{ - RsCacheConfig *item = (RsCacheConfig *) i; - uint32_t tlvsize = RsCacheConfigSerialiser::size(item); - uint32_t offset = 0; - - if(*size < tlvsize) - return false; /* not enough space */ - - *size = tlvsize; - - bool ok = true; - - ok &=setRsItemHeader(data, tlvsize, item->PacketId(), tlvsize); - - std::cerr << "RsCacheConfigSerialiser::serialise() Header: " << ok << std::endl; - std::cerr << "RsCacheConfigSerialiser::serialise() Size: " << size << std::endl; - - /* skip the header */ - offset += 8; - - /* add the mandatory parts first */ - - ok &= setRawUInt32(data, tlvsize, &offset, item->cacheid); - std::cerr << "RsCacheConfigSerialiser::serialise() cacheid: " << ok << std::endl; - - ok &= setRawUInt32(data, tlvsize, &offset, item->recvd); - std::cerr << "RsCacheConfigSerialiser::serialise() recvd: " << ok << std::endl; - - ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_PATH, item->path); - std::cerr << "RsCacheConfigSerialiser::serialise() path: " << ok << std::endl; - - ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_NAME, item->name); - std::cerr << "RsCacheConfigSerialiser::serialise() name: " << ok << std::endl; - - ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_HASH_SHA1, item->hash); - std::cerr << "RsCacheConfigSerialiser::serialise() hash: " << ok << std::endl; - - - if (offset !=tlvsize) - { - ok = false; - std::cerr << "RsConfigSerialiser::serialisertransfer() Size Error! " << std::endl; - } - - return ok; -} - -RsItem *RsCacheConfigSerialiser::deserialise(void *data, uint32_t *size) -{/* get the type and size */ - uint32_t rstype = getRsItemId(data); - uint32_t rssize = getRsItemSize(data); - - uint32_t *offset; - *offset = 0; - - uint32_t tlvend = *offset + rssize; - - - - - if ((RS_PKT_VERSION1 != getRsItemVersion(rstype)) || - (RS_PKT_CLASS_CONFIG != getRsItemClass(rstype)) || - (RS_PKT_TYPE_CACHE_CONFIG != getRsItemType(rstype)) || - (RS_PKT_SUBTYPE_DEFAULT != getRsItemSubType(rstype))) - { - return NULL; /* wrong type */ - } - - if (*size < rssize) /* check size */ - return NULL; /* not enough data */ - - /* set the packet length */ - *size = rssize; - - bool ok = true; - - /* ready to load */ - RsCacheConfig *item = new RsCacheConfig(); - item->clear(); - - /* skip the header */ - offset += 8; - - /* get mandatory parts first */ - - ok &= getRawUInt32(data, rssize, offset, &(item->cacheid)); - std::cerr << "RsCacheConfigSerialiser::deserialise() cacheid: " << ok << std::endl; - - ok &= getRawUInt32(data, rssize, offset, &(item->recvd)); - std::cerr << "RsCacheConfigSerialiser::deserialise() recvd: " << ok << std::endl; - - while((*offset) + 2 < tlvend) - { - /* get the next type */ - uint16_t tlvsubtype = GetTlvType( &(((uint8_t *) data)[*offset]) ); - - switch(tlvsubtype) - { - case TLV_TYPE_STR_PATH: - ok &= GetTlvString(data, tlvend, offset, tlvsubtype, item->path); - break; - case TLV_TYPE_STR_NAME: - ok &= GetTlvString(data, tlvend, offset, tlvsubtype, item->name); - break; - case TLV_TYPE_STR_HASH_SHA1: - ok &= GetTlvString(data, tlvend, offset, tlvsubtype, item->hash); - break; - default: - break; - } - if (!ok) - { - return false; - } + uint32_t s = 8; // to store calculated size, initiailize with size of header + + + s += GetTlvStringSize(item->pid); + s += 2; /* cachetypeid */ + s += 2; /* cachesubid */ + s += GetTlvStringSize(item->path); + s += GetTlvStringSize(item->name); + s += GetTlvStringSize(item->hash); + s += 8; /* size */ + s += 4; /* recvd */ + + return s; +} + +bool RsCacheConfigSerialiser::serialise(RsItem *i, void *data, uint32_t *size) +{ + RsCacheConfig *item = (RsCacheConfig *) i; + uint32_t tlvsize = RsCacheConfigSerialiser::size(item); + uint32_t offset = 0; + + if(*size < tlvsize) + return false; /* not enough space */ + + *size = tlvsize; + + bool ok = true; + + ok &=setRsItemHeader(data, tlvsize, item->PacketId(), tlvsize); + + std::cerr << "RsCacheConfigSerialiser::serialise() Header: " << ok << std::endl; + std::cerr << "RsCacheConfigSerialiser::serialise() Size: " << size << std::endl; + + /* skip the header */ + offset += 8; + + /* add the mandatory parts first */ + + ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_PEERID, item->pid); + std::cerr << "RsCacheConfigSerialiser::serialise() peerid: " << ok << std::endl; + + ok &= setRawUInt16(data, tlvsize, &offset, item->cachetypeid); + std::cerr << "RsCacheConfigSerialiser::serialise() cacheTypeId: " << ok << std::endl; + + ok &= setRawUInt16(data, tlvsize, &offset, item->cachesubid); + std::cerr << "RsCacheConfigSerialiser::serialise() cacheSubId: " << ok << std::endl; + + ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_PATH, item->path); + std::cerr << "RsCacheConfigSerialiser::serialise() path: " << ok << std::endl; + + ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_NAME, item->name); + std::cerr << "RsCacheConfigSerialiser::serialise() name: " << ok << std::endl; + + ok &= SetTlvString(data, tlvsize, &offset, TLV_TYPE_STR_HASH_SHA1, item->hash); + std::cerr << "RsCacheConfigSerialiser::serialise() hash: " << ok << std::endl; + + ok &= setRawUInt64(data, tlvsize, &offset, item->size); + std::cerr << "RsCacheConfigSerialiser::serialise() size: " << ok << std::endl; + + ok &= setRawUInt32(data, tlvsize, &offset, item->recvd); + std::cerr << "RsCacheConfigSerialiser::serialise() recvd: " << ok << std::endl; + + + if (offset !=tlvsize) + { + ok = false; + std::cerr << "RsConfigSerialiser::serialisertransfer() Size Error! " << std::endl; + } + + return ok; +} + +RsItem *RsCacheConfigSerialiser::deserialise(void *data, uint32_t *size) +{/* get the type and size */ + uint32_t rstype = getRsItemId(data); + uint32_t rssize = getRsItemSize(data); + + uint32_t offset; + offset = 0; + + if ((RS_PKT_VERSION1 != getRsItemVersion(rstype)) || + (RS_PKT_CLASS_CONFIG != getRsItemClass(rstype)) || + (RS_PKT_TYPE_CACHE_CONFIG != getRsItemType(rstype)) || + (RS_PKT_SUBTYPE_DEFAULT != getRsItemSubType(rstype))) + { + return NULL; /* wrong type */ + } + + if (*size < rssize) /* check size */ + return NULL; /* not enough data */ + + /* set the packet length */ + *size = rssize; + + bool ok = true; + + /* ready to load */ + RsCacheConfig *item = new RsCacheConfig(); + item->clear(); + + /* skip the header */ + offset += 8; + + /* get mandatory parts first */ + + ok &= GetTlvString(data, rssize, &offset, TLV_TYPE_STR_PEERID, item->pid); + std::cerr << "RsCacheConfigSerialiser::deserialise() peerid: " << ok << std::endl; + + ok &= getRawUInt16(data, rssize, &offset, &(item->cachetypeid)); + std::cerr << "RsCacheConfigSerialiser::serialise() cacheTypeId: " << ok << std::endl; + + ok &= getRawUInt16(data, rssize, &offset, &(item->cachesubid)); + std::cerr << "RsCacheConfigSerialiser::serialise() cacheSubId: " << ok << std::endl; + + ok &= GetTlvString(data, rssize, &offset, TLV_TYPE_STR_PATH, item->path); + std::cerr << "RsCacheConfigSerialiser::serialise() path: " << ok << std::endl; + + ok &= GetTlvString(data, rssize, &offset, TLV_TYPE_STR_NAME, item->name); + std::cerr << "RsCacheConfigSerialiser::serialise() name: " << ok << std::endl; + + ok &= GetTlvString(data, rssize, &offset, TLV_TYPE_STR_HASH_SHA1, item->hash); + std::cerr << "RsCacheConfigSerialiser::deserialise() hash: " << ok << std::endl; + + ok &= getRawUInt64(data, rssize, &offset, &(item->size)); + std::cerr << "RsCacheConfigSerialiser::deserialise() size: " << ok << std::endl; + + ok &= getRawUInt32(data, rssize, &offset, &(item->recvd)); + std::cerr << "RsCacheConfigSerialiser::deserialise() recvd: " << ok << std::endl; + + + if (offset != rssize) + { + + /* error */ + delete item; + return NULL; } return item; - } diff --git a/libretroshare/src/serialiser/rsconfigitems.h b/libretroshare/src/serialiser/rsconfigitems.h index 2f6b961c1..4dfe69834 100644 --- a/libretroshare/src/serialiser/rsconfigitems.h +++ b/libretroshare/src/serialiser/rsconfigitems.h @@ -126,12 +126,14 @@ virtual ~RsCacheConfig(); virtual void clear(); std::ostream &print(std::ostream &out, uint16_t indent = 0); - //RsTlvPeerId peerid; /* Mandatory */ - uint32_t cacheid; /* Mandatory */ + std::string pid; /* Mandatory */ + uint16_t cachetypeid; /* Mandatory */ + uint16_t cachesubid; /* Mandatory */ std::string path; /* Mandatory */ std::string name; /* Mandatory */ std::string hash; /* Mandatory */ + uint64_t size; /* Mandatory */ uint32_t recvd; /* Mandatory */ }; diff --git a/libretroshare/src/server/Makefile b/libretroshare/src/server/Makefile index fdf19bd03..ed1bf8e5c 100644 --- a/libretroshare/src/server/Makefile +++ b/libretroshare/src/server/Makefile @@ -9,9 +9,9 @@ include $(RS_TOP_DIR)/scripts/config.mk RSOBJ = ft.o ftfiler.o hashsearch.o filedexserver.o -TESTOBJ = ftcachetest.o +TESTOBJ = #ftcachetest.o -TESTS = ftcachetest +TESTS = #ftcachetest all: librs tests diff --git a/libretroshare/src/server/filedexserver.cc b/libretroshare/src/server/filedexserver.cc index 53813c1f8..27215b8bd 100644 --- a/libretroshare/src/server/filedexserver.cc +++ b/libretroshare/src/server/filedexserver.cc @@ -217,10 +217,10 @@ void filedexserver::setFileCallback(std::string ownId, CacheStrapper *strappe /* setup FiStore/Monitor */ std::string localcachedir = config_dir + "/cache/local"; std::string remotecachedir = config_dir + "/cache/remote"; - fiStore = new FileIndexStore(ftFiler, cb, ownId, remotecachedir); + fiStore = new FileIndexStore(strapper, ftFiler, cb, ownId, remotecachedir); /* now setup the FiMon */ - fimon = new FileIndexMonitor(localcachedir, ownId); + fimon = new FileIndexMonitor(strapper, localcachedir, ownId); /* setup ftFiler * to find peer info / savedir @@ -234,10 +234,7 @@ void filedexserver::setFileCallback(std::string ownId, CacheStrapper *strappe CachePair cp(fimon, fiStore, CacheId(RS_SERVICE_TYPE_FILE_INDEX, 0)); mCacheStrapper -> addCachePair(cp); - /* now we can load the cache configuration */ - //std::string cacheconfig = config_dir + "/caches.cfg"; - //mCacheStrapper -> loadCaches(cacheconfig); - +#if 0 /************ TMP HACK LOAD until new serialiser is finished */ /* get filename and hash from configuration */ std::string localCacheFile; // = getSSLRoot()->getSetting(LOCAL_CACHE_FILE_KEY); @@ -270,8 +267,14 @@ void filedexserver::setFileCallback(std::string ownId, CacheStrapper *strappe RsDirUtil::cleanupDirectory(remotecachedir, saveRemoteCaches); /* clean up all */ /************ TMP HACK LOAD until new serialiser is finished */ +#endif + + return; +} +void filedexserver::StartupMonitor() +{ /* startup the FileMonitor (after cache load) */ fimon->setPeriod(600); /* 10 minutes */ /* start it up */ @@ -296,9 +299,10 @@ void filedexserver::setFileCallback(std::string ownId, CacheStrapper *strappe delete rsft; } mResumeTransferList.clear(); - } + + int filedexserver::FileCacheSave() { /************ TMP HACK SAVE until new serialiser is finished */ @@ -441,29 +445,28 @@ int filedexserver::handleInputQueues() i_init = i; while((cr = pqisi -> RequestedSearch()) != NULL) { - //std::cerr << "filedexserver::handleInputQueues() Recvd RequestedSearch (CacheQuery!)" << std::endl; + /* just delete these */ std::ostringstream out; - if (i++ == i_init) - { - out << "Requested Search:" << std::endl; - } + out << "Requested Search:" << std::endl; cr -> print(out); pqioutput(PQL_DEBUG_BASIC, fldxsrvrzone, out.str()); + delete cr; + } - /* these go to the CacheStrapper (handled immediately) */ - std::map::iterator it; - std::map answer; - RsPeerId id; - mCacheStrapper->handleCacheQuery(id, answer); - for(it = answer.begin(); it != answer.end(); it++) + // Now handle it replacement (pushed cache results) + { + std::list > cacheUpdates; + std::list >::iterator it; + + mCacheStrapper->getCacheUpdates(cacheUpdates); + for(it = cacheUpdates.begin(); it != cacheUpdates.end(); it++) { - //std::cerr << "filedexserver::handleInputQueues() Sending (CacheAnswer!)" << std::endl; /* construct reply */ RsCacheItem *ci = new RsCacheItem(); /* id from incoming */ - ci -> PeerId(cr->PeerId()); + ci -> PeerId(it->first); ci -> file.hash = (it->second).hash; ci -> file.name = (it->second).name; @@ -473,13 +476,13 @@ int filedexserver::handleInputQueues() ci -> cacheSubId = (it->second).cid.subid; std::ostringstream out2; - out2 << "Outgoing CacheStrapper reply -> RsCacheItem:" << std::endl; + out2 << "Outgoing CacheStrapper Update -> RsCacheItem:" << std::endl; ci -> print(out2); - pqioutput(PQL_DEBUG_BASIC, fldxsrvrzone, out2.str()); + std::cerr << out2.str() << std::endl; + + //pqioutput(PQL_DEBUG_BASIC, fldxsrvrzone, out2.str()); pqisi -> SendSearchResult(ci); } - - delete cr; } // now File Input. @@ -545,11 +548,13 @@ int filedexserver::handleOutputQueues() //std::cerr << "filedexserver::handleOutputQueues()" << std::endl; int i = 0; + +#if 0 /* no more cache queries -> results are pushed */ + std::list ids; std::list::iterator pit; mCacheStrapper->sendCacheQuery(ids, time(NULL)); - for(pit = ids.begin(); pit != ids.end(); pit++) { //std::cerr << "filedexserver::handleOutputQueues() Cache Query for: " << (*pit) << std::endl; @@ -569,6 +574,8 @@ int filedexserver::handleOutputQueues() /* send it off */ pqisi -> SearchSpecific(cr); } +#endif + /* now see if the filer has any data */ ftFileRequest *ftr; diff --git a/libretroshare/src/server/filedexserver.h b/libretroshare/src/server/filedexserver.h index 120ec9182..d003a29cf 100644 --- a/libretroshare/src/server/filedexserver.h +++ b/libretroshare/src/server/filedexserver.h @@ -154,10 +154,13 @@ bool loadConfigMap(std::map &configMap); int FileStoreTick(); int FileCacheSave(); + /* Setup */ void initialiseFileStore(); void setFileCallback(std::string ownId, CacheStrapper *strapper, ftfiler *ft, NotifyBase *cb); +void StartupMonitor(); + /* Controls */ int RequestDirDetails(std::string uid, std::string path, DirDetails &details); int RequestDirDetails(void *ref, DirDetails &details, uint32_t flags); diff --git a/libretroshare/src/server/ftfiler.cc b/libretroshare/src/server/ftfiler.cc index aab966218..e7bd27866 100644 --- a/libretroshare/src/server/ftfiler.cc +++ b/libretroshare/src/server/ftfiler.cc @@ -173,6 +173,46 @@ bool ftfiler::RequestCacheFile(std::string id, std::string destpath, std::str return 1; } +bool ftfiler::CancelCacheFile(RsPeerId id, std::string path, + std::string hash, uint64_t size) +{ + /* clean up old transfer - just remove it (no callback) */ + { + std::ostringstream out; + out << "ftfiler::CancelCacheFile() Looking for: " << hash; + pqioutput(PQL_DEBUG_BASIC, ftfilerzone, out.str()); + } + + /* iterate through fileItems and check for this one */ + std::list::iterator it; + for(it = recvFiles.begin(); it != recvFiles.end(); it++) + { + if ((hash==(*it)->hash) && + (size==(*it)->size) && + ((*it)->ftMode == FT_MODE_CACHE)) + { + std::ostringstream out; + out << "ftfiler::CancelCacheFile() "; + out << "Match ftFileStatus: " << hash; + pqioutput(PQL_DEBUG_BASIC, ftfilerzone, out.str()); + /* same */ + + std::cerr << "Clearing Failed Cache Transfer: " << (*it)->name; + std::cerr << std::endl; + delete (*it); + it = recvFiles.erase(it); + + return true; + } + } + + std::cerr << "************* ERROR *****************"; + std::cerr << std::endl; + std::cerr << "ftfiler::CancelCacheFile() Failed to Find: " << hash; + std::cerr << std::endl; + return false; +} + ftFileStatus *ftfiler::findRecvFileItem(std::string hash) diff --git a/libretroshare/src/server/ftfiler.h b/libretroshare/src/server/ftfiler.h index e492d5b36..7542ce90a 100644 --- a/libretroshare/src/server/ftfiler.h +++ b/libretroshare/src/server/ftfiler.h @@ -159,6 +159,9 @@ virtual ~ftfiler() { return; } virtual bool RequestCacheFile(std::string id, std::string path, std::string hash, uint64_t size); +virtual bool CancelCacheFile(RsPeerId id, std::string path, + std::string hash, uint64_t size); + virtual int getFile(std::string name, std::string hash, uint64_t size, std::string destpath); diff --git a/libretroshare/src/services/p3ranking.cc b/libretroshare/src/services/p3ranking.cc index ed0afc9f0..672d55d07 100644 --- a/libretroshare/src/services/p3ranking.cc +++ b/libretroshare/src/services/p3ranking.cc @@ -42,11 +42,11 @@ std::string generateRandomLinkId(); #define RANK_DEBUG 1 -p3Ranking::p3Ranking(uint16_t type, CacheTransfer *cft, +p3Ranking::p3Ranking(uint16_t type, CacheStrapper *cs, CacheTransfer *cft, std::string sourcedir, std::string storedir, uint32_t storePeriod) - :CacheSource(type, true, sourcedir), - CacheStore(type, true, cft, storedir), + :CacheSource(type, true, cs, sourcedir), + CacheStore(type, true, cs, cft, storedir), mStorePeriod(storePeriod) { @@ -64,7 +64,35 @@ p3Ranking::p3Ranking(uint16_t type, CacheTransfer *cft, bool p3Ranking::loadLocalCache(const CacheData &data) { - /* ignore Local Cache -> just use remote caches */ + std::string filename = data.path + '/' + data.name; + std::string hash = data.hash; + //uint64_t size = data.size; + std::string source = data.pid; + +#ifdef RANK_DEBUG + std::cerr << "p3Ranking::loadLocalCache()"; + std::cerr << std::endl; + std::cerr << "\tSource: " << source; + std::cerr << std::endl; + std::cerr << "\tFilename: " << filename; + std::cerr << std::endl; + std::cerr << "\tHash: " << hash; + std::cerr << std::endl; + std::cerr << "\tSize: " << data.size; + std::cerr << std::endl; +#endif + + loadRankFile(filename, source); + + { + RsStackMutex stack(mRankMtx); /********** STACK LOCKED MTX ******/ + mRepublish = false; + } + + if (data.size > 0) /* don't refresh zero sized caches */ + { + refreshCache(data); + } return true; } @@ -90,6 +118,11 @@ int p3Ranking::loadCache(const CacheData &data) loadRankFile(filename, source); + + CacheStore::lockData(); /***** LOCK ****/ + locked_storeCacheEntry(data); + CacheStore::unlockData(); /***** UNLOCK ****/ + return 1; } diff --git a/libretroshare/src/services/p3ranking.h b/libretroshare/src/services/p3ranking.h index 784049b5e..3a8513be9 100644 --- a/libretroshare/src/services/p3ranking.h +++ b/libretroshare/src/services/p3ranking.h @@ -65,7 +65,7 @@ class p3Ranking: public CacheSource, public CacheStore { public: - p3Ranking(uint16_t type, CacheTransfer *cft, + p3Ranking(uint16_t type, CacheStrapper *cs, CacheTransfer *cft, std::string sourcedir, std::string storedir, uint32_t storePeriod);