Merge branch 'master' of github.com:jedmccaleb/NewCoin

src/cpp/ripple/Application.cpp

@@ -52,6 +52,7 @@ Application::Application() :
 	mJobQueue.setThreadCount();
 	mSweepTimer.expires_from_now(boost::posix_time::seconds(10));
 	mSweepTimer.async_wait(boost::bind(&Application::sweep, this));
+	mLoadMgr.init();
 }
 
 extern const char *RpcDBInit[], *TxnDBInit[], *LedgerDBInit[], *WalletDBInit[], *HashNodeDBInit[], *NetNodeDBInit[];
@@ -147,6 +148,8 @@ void Application::setup()
 	else
 		startNewLedger();
 
+	mOrderBookDB.setup(theApp->getLedgerMaster().getCurrentLedger()); // TODO: We need to update this if the ledger jumps
+
 	//
 	// Begin validation and ip maintenance.
 	// - Wallet maintains local information: including identity and network connection persistence information.

src/cpp/ripple/Application.h

@@ -24,6 +24,7 @@
 #include "ProofOfWork.h"
 #include "LoadManager.h"
 #include "TransactionQueue.h"
+#include "OrderBookDB.h"
 
 class RPCDoor;
 class PeerDoor;
@@ -66,6 +67,7 @@ class Application
 	LoadManager				mLoadMgr;
 	LoadFeeTrack			mFeeTrack;
 	TXQueue					mTxnQueue;
+	OrderBookDB				mOrderBookDB;
 
 	DatabaseCon				*mRpcDB, *mTxnDB, *mLedgerDB, *mWalletDB, *mHashNodeDB, *mNetNodeDB;
 
@@ -115,6 +117,7 @@ public:
 	LoadFeeTrack& getFeeTrack()						{ return mFeeTrack; }
 	TXQueue& getTxnQueue()							{ return mTxnQueue; }
 	PeerDoor& getPeerDoor()							{ return *mPeerDoor; }
+	OrderBookDB& getOrderBookDB()					{ return mOrderBookDB; }
 
 
 	bool isNew(const uint256& s)					{ return mSuppressions.addSuppression(s); }

src/cpp/ripple/JobQueue.cpp

@@ -181,6 +181,16 @@ Json::Value JobQueue::getJson(int)
 	return ret;
 }
 
+int JobQueue::isOverloaded()
+{
+	int count = 0;
+	boost::mutex::scoped_lock sl(mJobLock);
+	for (int i = 0; i < NUM_JOB_TYPES; ++i)
+		if (mJobLoads[i].isOver())
+			++count;
+	return count;
+}
+
 void JobQueue::shutdown()
 { // shut down the job queue without completing pending jobs
 	cLog(lsINFO) << "Job queue shutting down";

src/cpp/ripple/JobQueue.h

@@ -108,6 +108,7 @@ public:
 	LoadEvent::autoptr getLoadEventAP(JobType t)
 	{ return LoadEvent::autoptr(new LoadEvent(mJobLoads[t], true, 1)); }
 
+	int isOverloaded();
 	Json::Value getJson(int c = 0);
 };
 

src/cpp/ripple/KeyCache.h

@@ -6,11 +6,13 @@
 #include <boost/unordered_map.hpp>
 #include <boost/thread/mutex.hpp>
 
+extern int upTime();
+
 template <typename c_Key> class KeyCache
 { // Maintains a cache of keys with no associated data
 public:
 	typedef c_Key									key_type;
-	typedef boost::unordered_map<key_type, time_t>	map_type;
+	typedef boost::unordered_map<key_type, int>		map_type;
 	typedef typename map_type::iterator				map_iterator;
 
 protected:
@@ -65,7 +67,7 @@ public:
 		if (it == mCache.end())
 			return false;
 		if (refresh)
-			it->second = time(NULL);
+			it->second = upTime();
 		return true;
 	}
 
@@ -88,19 +90,19 @@ public:
 		map_iterator it = mCache.find(key);
 		if (it != mCache.end())
 		{
-			it->second = time(NULL);
+			it->second = upTime();
 			return false;
 		}
-		mCache.insert(std::make_pair(key, time(NULL)));
+		mCache.insert(std::make_pair(key, upTime()));
 		return true;
 	}
 
 	void sweep()
 	{ // Remove stale entries from the cache 
-		time_t now = time(NULL);
+		int now = upTime();
 		boost::mutex::scoped_lock sl(mNCLock);
 
-		time_t target;
+		int target;
 		if ((mTargetSize == 0) || (mCache.size() <= mTargetSize))
 			target = now - mTargetAge;
 		else

src/cpp/ripple/LedgerAcquire.cpp

@@ -22,7 +22,7 @@ DECLARE_INSTANCE(LedgerAcquire);
 PeerSet::PeerSet(const uint256& hash, int interval) : mHash(hash), mTimerInterval(interval), mTimeouts(0),
 	mComplete(false), mFailed(false), mProgress(true), mAggressive(true), mTimer(theApp->getIOService())
 {
-	mLastAction = time(NULL);
+	mLastAction = upTime();
 	assert((mTimerInterval > 10) && (mTimerInterval < 30000));
 }
 
@@ -890,7 +890,7 @@ void LedgerAcquireMaster::sweep()
 {
 	mRecentFailures.sweep();
 
-	time_t now = time(NULL);
+	int now = upTime();
 	boost::mutex::scoped_lock sl(mLock);
 
 	std::map<uint256, LedgerAcquire::pointer>::iterator it = mLedgers.begin();

src/cpp/ripple/LedgerAcquire.h

@@ -32,7 +32,7 @@ protected:
 	uint256 mHash;
 	int mTimerInterval, mTimeouts;
 	bool mComplete, mFailed, mProgress, mAggressive;
-	time_t mLastAction;
+	int mLastAction;
 
 	boost::recursive_mutex					mLock;
 	boost::asio::deadline_timer				mTimer;
@@ -53,8 +53,8 @@ public:
 	bool isActive();
 	void progress()						{ mProgress = true; mAggressive = false; }
 	bool isProgress()					{ return mProgress; }
-	void touch()						{ mLastAction = time(NULL); }
+	void touch()						{ mLastAction = upTime(); }
-	time_t getLastAction()				{ return mLastAction; }
+	int getLastAction()					{ return mLastAction; }
 
 	void peerHas(Peer::ref);
 	void badPeer(Peer::ref);

src/cpp/ripple/LoadManager.cpp

@@ -1,15 +1,28 @@
 #include "LoadManager.h"
 
 #include <boost/test/unit_test.hpp>
+#include <boost/thread.hpp>
+#include <boost/date_time/posix_time/posix_time.hpp>
 
 #include "Log.h"
 #include "Config.h"
+#include "Application.h"
 
 SETUP_LOG();
 
+static volatile int* uptimePtr = NULL;
+
+int upTime()
+{
+	static time_t firstCall = time(NULL);
+	if (uptimePtr != NULL)
+		return *uptimePtr;
+	return static_cast<int>(time(NULL) - firstCall);
+}
+
 LoadManager::LoadManager(int creditRate, int creditLimit, int debitWarn, int debitLimit) :
 		mCreditRate(creditRate), mCreditLimit(creditLimit), mDebitWarn(debitWarn), mDebitLimit(debitLimit),
-		mCosts(LT_MAX)
+		mShutdown(false), mUptime(0), mCosts(LT_MAX)
 {
 	addLoadCost(LoadCost(LT_InvalidRequest,		10,		LC_CPU | LC_Network));
 	addLoadCost(LoadCost(LT_RequestNoReply,		1,		LC_CPU | LC_Disk));
@@ -23,6 +36,35 @@ LoadManager::LoadManager(int creditRate, int creditLimit, int debitWarn, int deb
 
 	addLoadCost(LoadCost(LT_RequestData,		5,		LC_Disk | LC_Network));
 	addLoadCost(LoadCost(LT_CheapQuery,			1,		LC_CPU));
+
+}
+
+void LoadManager::init()
+{
+	if (uptimePtr == NULL)
+		uptimePtr = static_cast<volatile int *>(&mUptime);
+	boost::thread(boost::bind(&LoadManager::threadEntry, this)).detach();
+}
+
+LoadManager::~LoadManager()
+{
+	if (uptimePtr == &mUptime)
+		uptimePtr = NULL;
+	{
+		boost::mutex::scoped_lock sl(mLock);
+		mShutdown = true;
+	}
+
+	do
+	{
+		boost::this_thread::sleep(boost::posix_time::milliseconds(100));
+		{
+			boost::mutex::scoped_lock sl(mLock);
+			if (!mShutdown)
+				return;
+		}
+	}
+	while (1);
 }
 
 
@@ -74,7 +116,7 @@ void LoadManager::setDebitLimit(int r)
 	mDebitLimit = r;
 }
 
-void LoadManager::canonicalize(LoadSource& source, const time_t now) const
+void LoadManager::canonicalize(LoadSource& source, int now) const
 {
 	if (source.mLastUpdate != now)
 	{
@@ -90,9 +132,9 @@ void LoadManager::canonicalize(LoadSource& source, const time_t now) const
 
 bool LoadManager::shouldWarn(LoadSource& source) const
 {
-	time_t now = time(NULL);
 	boost::mutex::scoped_lock sl(mLock);
 
+	int now = upTime();
 	canonicalize(source, now);
 	if (source.isPrivileged() || (source.mBalance < mDebitWarn) || (source.mLastWarning == now))
 		return false;
@@ -103,9 +145,9 @@ bool LoadManager::shouldWarn(LoadSource& source) const
 
 bool LoadManager::shouldCutoff(LoadSource& source) const
 {
-	time_t now = time(NULL);
 	boost::mutex::scoped_lock sl(mLock);
 
+	int now = upTime();
 	canonicalize(source, now);
 	return !source.isPrivileged() && (source.mBalance < mDebitLimit);
 }
@@ -118,10 +160,10 @@ bool LoadManager::adjust(LoadSource& source, LoadType t) const
 
 bool LoadManager::adjust(LoadSource& source, int credits) const
 { // return: true = need to warn/cutoff
-	time_t now = time(NULL);
 	boost::mutex::scoped_lock sl(mLock);
 
 	// We do it this way in case we want to add exponential decay later
+	int now = upTime();
 	canonicalize(source, now);
 	source.mBalance += credits;
 	if (source.mBalance > mCreditLimit)
@@ -239,6 +281,45 @@ Json::Value LoadFeeTrack::getJson(uint64 baseFee, uint32 referenceFeeUnits)
 	return j;
 }
 
+int LoadManager::getUptime()
+{
+	boost::mutex::scoped_lock sl(mLock);
+	return mUptime;
+}
+
+void LoadManager::threadEntry()
+{
+	boost::posix_time::ptime t = boost::posix_time::microsec_clock::universal_time();
+	while (1)
+	{
+		{
+			boost::mutex::scoped_lock sl(mLock);
+			if (mShutdown)
+			{
+				mShutdown = false;
+				return;
+			}
+			++mUptime;
+		}
+
+		if (theApp->getJobQueue().isOverloaded())
+			theApp->getFeeTrack().raiseLocalFee();
+		else
+			theApp->getFeeTrack().lowerLocalFee();
+
+		t += boost::posix_time::seconds(1);
+		boost::posix_time::time_duration when = t - boost::posix_time::microsec_clock::universal_time();
+
+		if ((when.is_negative()) || (when.total_seconds() > 1))
+		{
+			cLog(lsWARNING) << "time jump";
+			t = boost::posix_time::microsec_clock::universal_time();
+		}
+		else
+			boost::this_thread::sleep(when);
+	}
+}
+
 BOOST_AUTO_TEST_SUITE(LoadManager_test)
 
 BOOST_AUTO_TEST_CASE(LoadFeeTrack_test)

src/cpp/ripple/LoadManager.h

@@ -9,6 +9,8 @@
 
 #include "types.h"
 
+extern int upTime();
+
 enum LoadType
 { // types of load that can be placed on the server
 
@@ -61,12 +63,12 @@ public:
 protected:
 	int		mBalance;
 	int		mFlags;
-	time_t	mLastUpdate;
+	int		mLastUpdate;
-	time_t	mLastWarning;
+	int		mLastWarning;
 
 public:
 	LoadSource() : mBalance(0), mFlags(0), mLastWarning(0)
-									{ mLastUpdate = time(NULL); }
+									{ mLastUpdate = upTime(); }
 
 	bool	isPrivileged() const	{ return (mFlags & lsfPrivileged) != 0; }
 	void	setPrivileged()			{ mFlags |= lsfPrivileged; }
@@ -86,17 +88,27 @@ protected:
 	int mDebitWarn;				// when a source drops below this, we warn
 	int mDebitLimit;			// when a source drops below this, we cut it off (should be negative)
 
+	bool mShutdown;
+
+	int mSpace1[4];				// We want mUptime to have its own cache line
+	int mUptime;
+	int mSpace2[4];
+
 	mutable boost::mutex mLock;
 
-	void canonicalize(LoadSource&, const time_t now) const;
+	void canonicalize(LoadSource&, int upTime) const;
 
 	std::vector<LoadCost>	mCosts;
 
 	void addLoadCost(const LoadCost& c) { mCosts[static_cast<int>(c.mType)] = c; }
 
+	void threadEntry();
+
 public:
 
 	LoadManager(int creditRate = 10, int creditLimit = 50, int debitWarn = -50, int debitLimit = -100);
+	~LoadManager();
+	void init();
 
 	int getCreditRate() const;
 	int getCreditLimit() const;
@@ -113,6 +125,7 @@ public:
 	bool adjust(LoadSource&, LoadType l) const;
 
 	int getCost(LoadType t)		{ return mCosts[static_cast<int>(t)].mCost; }
+	int getUptime();
 };
 
 class LoadFeeTrack

src/cpp/ripple/LoadMonitor.cpp

@@ -2,7 +2,7 @@
 
 void LoadMonitor::update()
 { // call with the mutex
-	time_t now = time(NULL);
+	int now = upTime();
 
 	if (now == mLastUpdate) // current
 		return;
@@ -69,6 +69,18 @@ void LoadMonitor::addCountAndLatency(int counts, int latency)
 		mLatencyMSPeak = lp;
 }
 
+bool LoadMonitor::isOver()
+{
+	boost::mutex::scoped_lock sl(mLock);
+
+	update();
+
+	if (mLatencyEvents == 0)
+		return 0;
+
+	return isOverTarget(mLatencyMSAvg / (mLatencyEvents * 4), mLatencyMSPeak / (mLatencyEvents * 4));
+}
+
 void LoadMonitor::getCountAndLatency(uint64& count, uint64& latencyAvg, uint64& latencyPeak, bool& isOver)
 {
 	boost::mutex::scoped_lock sl(mLock);
@@ -89,3 +101,5 @@ void LoadMonitor::getCountAndLatency(uint64& count, uint64& latencyAvg, uint64&
 	}
 	isOver = isOverTarget(latencyAvg, latencyPeak);
 }
+
+// vim:ts=4

src/cpp/ripple/LoadMonitor.h

@@ -7,6 +7,7 @@
 #include <boost/shared_ptr.hpp>
 
 #include "types.h"
+extern int upTime();
 
 // Monitors load levels and response times
 
@@ -19,7 +20,7 @@ protected:
 	uint64				mLatencyMSPeak;
 	uint64				mTargetLatencyAvg;
 	uint64				mTargetLatencyPk;
-	time_t				mLastUpdate;
+	int					mLastUpdate;
 	boost::mutex		mLock;
 
 	void update();
@@ -27,7 +28,7 @@ protected:
 public:
 	LoadMonitor() : mCounts(0), mLatencyEvents(0), mLatencyMSAvg(0), mLatencyMSPeak(0),
 		mTargetLatencyAvg(0), mTargetLatencyPk(0)
-	{ mLastUpdate = time(NULL); }
+	{ mLastUpdate = upTime(); }
 
 	void addCount(int counts);
 	void addLatency(int latency);
@@ -35,8 +36,8 @@ public:
 
 	void setTargetLatency(uint64 avg, uint64 pk)
 	{
-		mTargetLatencyAvg = avg * 4;
+		mTargetLatencyAvg  = avg;
-		mTargetLatencyPk = pk * 4;
+		mTargetLatencyPk = pk;
 	}
 
 	bool isOverTarget(uint64 avg, uint64 peak)
@@ -46,6 +47,7 @@ public:
 	}
 
 	void getCountAndLatency(uint64& count, uint64& latencyAvg, uint64& latencyPeak, bool& isOver);
+	bool isOver();
 };
 
 class LoadEvent

src/cpp/ripple/NetworkOPs.cpp

@@ -1178,7 +1178,7 @@ Json::Value NetworkOPs::getServerInfo(bool human, bool admin)
 		}
 		else
 		{
-			l["base_fee_xrp"]		= static_cast<double>(Json::UInt(baseFee)) / SYSTEM_CURRENCY_PARTS;
+			l["base_fee_xrp"]		= static_cast<double>(baseFee) / SYSTEM_CURRENCY_PARTS;
 			l["reserve_base_xrp"]	=
 				static_cast<double>(Json::UInt(lpClosed->getReserve(0) * baseFee / baseRef)) / SYSTEM_CURRENCY_PARTS;
 			l["reserve_inc_xrp"]	=
@@ -1321,6 +1321,7 @@ void NetworkOPs::pubAcceptedTransaction(Ledger::ref lpCurrent, const SerializedT
 			ispListener->send(jvObj);
 		}
 	}
+	theApp->getOrderBookDB().processTxn(stTxn, terResult, meta, jvObj);
 
 	pubAccountTransaction(lpCurrent, stTxn, terResult, true, meta);
 }
@@ -1451,6 +1452,20 @@ void NetworkOPs::unsubAccount(InfoSub* ispListener, const boost::unordered_set<R
 	}
 }
 
+bool NetworkOPs::subBook(InfoSub* ispListener, uint160 currencyIn, uint160 currencyOut, uint160 issuerIn, uint160 issuerOut)
+{
+	BookListeners::pointer listeners=theApp->getOrderBookDB().makeBookListeners(currencyIn, currencyOut, issuerIn, issuerOut);
+	if(listeners) listeners->addSubscriber(ispListener);
+	return(true);
+}
+
+bool NetworkOPs::unsubBook(InfoSub* ispListener, uint160 currencyIn, uint160 currencyOut, uint160 issuerIn, uint160 issuerOut)
+{
+	BookListeners::pointer listeners=theApp->getOrderBookDB().getBookListeners(currencyIn, currencyOut, issuerIn, issuerOut);
+	if(listeners) listeners->removeSubscriber(ispListener);
+	return(true);
+}
+
 void NetworkOPs::newLCL(int proposers, int convergeTime, const uint256& ledgerHash)
 {
 	assert(convergeTime);

src/cpp/ripple/NetworkOPs.h

@@ -71,6 +71,7 @@ protected:
 	typedef boost::unordered_map<uint160,boost::unordered_set<InfoSub*> >::iterator		subInfoMapIterator;
 
 	typedef boost::unordered_map<uint160,std::pair<InfoSub*,uint32> >					subSubmitMapType;
+	//typedef boost::unordered_map<OrderBook::pointer,boost::unordered_set<InfoSub*> >	subOrderMap;
 
 	typedef boost::unordered_map<std::string, RPCSub* >									subRpcMapType;
 
@@ -103,6 +104,7 @@ protected:
 	subInfoMapType										mSubAccount;
 	subInfoMapType										mSubRTAccount;
 	subSubmitMapType									mSubmitMap;   // TODO: probably dump this
+	
 
 	subRpcMapType										mRpcSubMap;
 
@@ -283,6 +285,9 @@ public:
 	bool subServer(InfoSub* ispListener, Json::Value& jvResult);
 	bool unsubServer(InfoSub* ispListener);
 
+	bool subBook(InfoSub* ispListener, uint160 currencyIn, uint160 currencyOut, uint160 issuerIn, uint160 issuerOut);
+	bool unsubBook(InfoSub* ispListener, uint160 currencyIn, uint160 currencyOut, uint160 issuerIn, uint160 issuerOut);
+
 	bool subTransactions(InfoSub* ispListener);
 	bool unsubTransactions(InfoSub* ispListener);
 

src/cpp/ripple/OrderBook.cpp

@@ -20,4 +20,6 @@ OrderBook::OrderBook(SerializedLedgerEntry::pointer ledgerEntry)
 
 	mBookBase=Ledger::getBookBase(mCurrencyIn, mIssuerIn, mCurrencyOut, mIssuerOut);
 }
+
+
 // vim:ts=4

src/cpp/ripple/OrderBook.h

@@ -1,5 +1,7 @@
 #include "SerializedLedger.h"
+#include "NetworkOPs.h"
 #include <boost/shared_ptr.hpp>
+
 /*
 	Encapsulates the SLE for an orderbook
 */

src/cpp/ripple/OrderBookDB.cpp

@@ -5,9 +5,18 @@
 
 SETUP_LOG();
 
-// TODO: this would be way faster if we could just look under the order dirs
+OrderBookDB::OrderBookDB()
-OrderBookDB::OrderBookDB(Ledger::pointer ledger)
 {
+
+}
+
+// TODO: this would be way faster if we could just look under the order dirs
+void OrderBookDB::setup(Ledger::pointer ledger)
+{
+	mXRPOrders.clear();
+	mIssuerMap.clear();
+	mKnownMap.clear();
+
 	// walk through the entire ledger looking for orderbook entries
 	uint256 currentIndex = ledger->getFirstLedgerIndex();
 
@@ -70,4 +79,151 @@ void OrderBookDB::getBooks(const uint160& issuerID, const uint160& currencyID, s
 	}
 }
 
+BookListeners::pointer OrderBookDB::makeBookListeners(uint160 currencyIn, uint160 currencyOut, uint160 issuerIn, uint160 issuerOut)
+{
+	BookListeners::pointer ret=getBookListeners(currencyIn, currencyOut, issuerIn, issuerOut);
+	if(!ret)
+	{
+		ret=BookListeners::pointer(new BookListeners);
+		mListeners[issuerIn][issuerOut][currencyIn][currencyOut]=ret;
+	}
+	return(ret);
+}
+
+BookListeners::pointer OrderBookDB::getBookListeners(uint160 currencyIn, uint160 currencyOut, uint160 issuerIn, uint160 issuerOut)
+{
+	std::map<uint160, std::map<uint160, std::map<uint160, std::map<uint160, BookListeners::pointer> > > >::iterator it0=mListeners.find(issuerIn);
+	if(it0 != mListeners.end())
+	{
+		std::map<uint160, std::map<uint160, std::map<uint160, BookListeners::pointer> > >::iterator it1=(*it0).second.find(issuerOut);
+		if(it1 != (*it0).second.end())
+		{
+			std::map<uint160, std::map<uint160, BookListeners::pointer> >::iterator it2=(*it1).second.find(currencyIn);
+			if(it2 != (*it1).second.end())
+			{
+				std::map<uint160, BookListeners::pointer>::iterator it3=(*it2).second.find(currencyOut);
+				if(it3 != (*it2).second.end())
+				{
+					return( (*it3).second );
+				}
+			}
+		}
+	}
+	return(BookListeners::pointer());
+}
+
+/*
+"CreatedNode" : {
+"LedgerEntryType" : "Offer",
+"LedgerIndex" : "F353BF8A7DCE35EA2985596F4C8421E30EF3B9A21618566BFE0ED00B62A8A5AB",
+"NewFields" : {
+"Account" : "rB5TihdPbKgMrkFqrqUC3yLdE8hhv4BdeY",
+"BookDirectory" : "FF26BE244767D0EA9EFD523941439009E4185E4CBB918F714C08E1BC9BF04000",
+"Sequence" : 112,
+"TakerGets" : "400000000",
+"TakerPays" : {
+"currency" : "BTC",
+"issuer" : "r3kmLJN5D28dHuH8vZNUZpMC43pEHpaocV",
+"value" : "1"
+}
+}
+}
+
+"ModifiedNode" : {
+"FinalFields" : {
+"Account" : "rHTxKLzRbniScyQFGMb3NodmxA848W8dKM",
+"BookDirectory" : "407AF8FFDE71114B1981574FDDA9B0334572D56FC579735B4B0BD7A625405555",
+"BookNode" : "0000000000000000",
+"Flags" : 0,
+"OwnerNode" : "0000000000000000",
+"Sequence" : 32,
+"TakerGets" : "149900000000",
+"TakerPays" : {
+"currency" : "USD",
+"issuer" : "r9vbV3EHvXWjSkeQ6CAcYVPGeq7TuiXY2X",
+"value" : "49.96666666666667"
+}
+},
+"LedgerEntryType" : "Offer",
+"LedgerIndex" : "C60F8CC514208FA5F7BD03CF1B64B38B7183CD52318FCBBD3726350D4FE693B0",
+"PreviousFields" : {
+"TakerGets" : "150000000000",
+"TakerPays" : {
+"currency" : "USD",
+"issuer" : "r9vbV3EHvXWjSkeQ6CAcYVPGeq7TuiXY2X",
+"value" : "50"
+}
+},
+"PreviousTxnID" : "1A6AAE3F1AC5A8A7554A5ABC395D17FED5BF62CD90181AA8E4315EDFED4EDEB3",
+"PreviousTxnLgrSeq" : 140734
+}
+
+*/
+// Based on the meta, send the meta to the streams that are listening 
+// We need to determine which streams a given meta effects
+void OrderBookDB::processTxn(const SerializedTransaction& stTxn, TER terResult,TransactionMetaSet::pointer& meta,Json::Value& jvObj)
+{
+	if(terResult==tesSUCCESS)
+	{
+		// check if this is an offer or an offer cancel or a payment that consumes an offer
+		//check to see what the meta looks like
+		BOOST_FOREACH(STObject& node,meta->getNodes())
+		{
+			try{
+				if(node.getFieldU16(sfLedgerEntryType)==ltOFFER)
+				{
+					SField* field=NULL;
+
+					if(node.getFName() == sfModifiedNode)
+					{
+						field=&sfPreviousFields;
+					}else if(node.getFName() == sfCreatedNode)
+					{
+						field=&sfNewFields;
+					}
+
+					const STObject* previous = dynamic_cast<const STObject*>(node.peekAtPField(*field));
+					if(previous)
+					{
+						STAmount takerGets = previous->getFieldAmount(sfTakerGets);
+						uint160 currencyOut=takerGets.getCurrency();
+						uint160 issuerOut=takerGets.getIssuer();
+
+						STAmount takerPays = previous->getFieldAmount(sfTakerPays);
+						uint160 currencyIn=takerPays.getCurrency();
+						uint160 issuerIn=takerPays.getIssuer();
+
+						// determine the OrderBook
+						BookListeners::pointer book=getBookListeners(currencyIn,currencyOut,issuerIn,issuerOut);
+						if(book) book->publish(jvObj);
+					}
+				}
+			}catch(...)
+			{
+				cLog(lsINFO) << "Fields not found in OrderBookDB::processTxn";
+			}
+		}
+	}
+}
+
+void BookListeners::addSubscriber(InfoSub* sub)
+{
+	mListeners.insert(sub);
+}
+
+void BookListeners::removeSubscriber(InfoSub* sub)
+{
+	mListeners.erase(sub);
+}
+
+void BookListeners::publish(Json::Value& jvObj)
+{
+	//Json::Value jvObj=node.getJson(0);
+
+	BOOST_FOREACH(InfoSub* sub,mListeners)
+	{
+		sub->send(jvObj);
+	}
+}
+
 // vim:ts=4

src/cpp/ripple/OrderBookDB.h

@@ -1,21 +1,39 @@
 #include "Ledger.h"
 #include "OrderBook.h"
+#include <boost/shared_ptr.hpp>
+
 
 //
 // XXX Eventually make this cached and just update it as transactions come in.
 // But, for now it is probably faster to just generate it each time.
 //
 
+class BookListeners
+{
+	boost::unordered_set<InfoSub*> mListeners;
+public:
+	typedef boost::shared_ptr<BookListeners> pointer;
+
+	void addSubscriber(InfoSub* sub);
+	void removeSubscriber(InfoSub* sub);
+	void publish(Json::Value& jvObj);
+};
+
 class OrderBookDB
 {
 	std::vector<OrderBook::pointer> mEmptyVector;
 	std::vector<OrderBook::pointer> mXRPOrders;
 	std::map<uint160, std::vector<OrderBook::pointer> > mIssuerMap;
+	//std::vector<OrderBook::pointer> mAllOrderBooks;
+
+	// issuerIn, issuerOut, currencyIn, currencyOut
+	std::map<uint160, std::map<uint160, std::map<uint160, std::map<uint160, BookListeners::pointer> > > > mListeners; 
 
 	std::map<uint256, bool >  mKnownMap;
 
 public:
-	OrderBookDB(Ledger::pointer ledger);
+	OrderBookDB();
+	void setup(Ledger::pointer ledger);
 
 	// return list of all orderbooks that want XRP
 	std::vector<OrderBook::pointer>& getXRPInBooks(){ return mXRPOrders; }
@@ -28,6 +46,14 @@ public:
 
 	// returns the best rate we can find
 	float getPrice(uint160& currencyIn,uint160& currencyOut);
+
+
+	BookListeners::pointer getBookListeners(uint160 currencyIn, uint160 currencyOut, uint160 issuerIn, uint160 issuerOut);
+	BookListeners::pointer makeBookListeners(uint160 currencyIn, uint160 currencyOut, uint160 issuerIn, uint160 issuerOut);
+
+	// see if this txn effects any orderbook
+	void processTxn(const SerializedTransaction& stTxn, TER terResult,TransactionMetaSet::pointer& meta,Json::Value& jvObj);
+
 };
 
 // vim:ts=4

src/cpp/ripple/Pathfinder.cpp

@@ -12,8 +12,6 @@
 SETUP_LOG();
 
 /*
-JED: V IIII
-
 we just need to find a succession of the highest quality paths there until we find enough width
 
 Don't do branching within each path
@@ -134,12 +132,13 @@ Pathfinder::Pathfinder(const RippleAddress& uSrcAccountID, const RippleAddress&
 		mDstAccountID(uDstAccountID.getAccountID()),
 		mDstAmount(saDstAmount),
 		mSrcCurrencyID(uSrcCurrencyID),
-		mSrcIssuerID(uSrcIssuerID),
+		mSrcIssuerID(uSrcIssuerID)
-		mOrderBook(theApp->getLedgerMaster().getCurrentLedger())
 {
 	mLedger		= theApp->getLedgerMaster().getCurrentLedger();
 	mSrcAmount	= STAmount(uSrcCurrencyID, uSrcIssuerID, 1, 0, true);	// -1/uSrcIssuerID/uSrcIssuerID
 
+	theApp->getOrderBookDB().setup( theApp->getLedgerMaster().getCurrentLedger()); // TODO: have the orderbook update itself rather than rebuild it from scratch each time
+
 	// Construct the default path for later comparison.
 
 	PathState::pointer	psDefault	= boost::make_shared<PathState>(mDstAmount, mSrcAmount, mLedger);
@@ -326,8 +325,7 @@ bool Pathfinder::findPaths(const unsigned int iMaxSteps, const unsigned int iMax
 			else if (!speEnd.mCurrencyID)
 			{
 				// Cursor is for XRP, continue with qualifying books: XRP -> non-XRP
-
+				BOOST_FOREACH(OrderBook::ref book, theApp->getOrderBookDB().getXRPInBooks())
-				BOOST_FOREACH(OrderBook::ref book, mOrderBook.getXRPInBooks())
 				{
 					// New end is an order book with the currency and issuer.
 
@@ -420,7 +418,8 @@ bool Pathfinder::findPaths(const unsigned int iMaxSteps, const unsigned int iMax
 				// Every book that wants the source currency.
 				std::vector<OrderBook::pointer> books;
 
-				mOrderBook.getBooks(speEnd.mIssuerID, speEnd.mCurrencyID, books);
+
+				theApp->getOrderBookDB().getBooks(speEnd.mIssuerID, speEnd.mCurrencyID, books);
 
 				BOOST_FOREACH(OrderBook::ref book, books)
 				{
@@ -573,7 +572,7 @@ void Pathfinder::addOptions(PathOption::pointer tail)
 {
 	if (!tail->mCurrencyID)
 	{ // source XRP
-		BOOST_FOREACH(OrderBook::ref book, mOrderBook.getXRPInBooks())
+		BOOST_FOREACH(OrderBook::ref book, theApp->getOrderBookDB().getXRPInBooks())
 		{
 			PathOption::pointer pathOption(new PathOption(tail));
 
@@ -607,7 +606,7 @@ void Pathfinder::addOptions(PathOption::pointer tail)
 
 		// every offer that wants the source currency
 		std::vector<OrderBook::pointer> books;
-		mOrderBook.getBooks(tail->mCurrentAccount, tail->mCurrencyID, books);
+		theApp->getOrderBookDB().getBooks(tail->mCurrentAccount, tail->mCurrencyID, books);
 
 		BOOST_FOREACH(OrderBook::ref book,books)
 		{

src/cpp/ripple/Pathfinder.h

@@ -3,7 +3,6 @@
 
 #include "SerializedTypes.h"
 #include "RippleAddress.h"
-#include "OrderBookDB.h"
 #include "RippleCalc.h"
 #include <boost/shared_ptr.hpp>
 
@@ -42,7 +41,7 @@ class Pathfinder
 	uint160				mSrcIssuerID;
 	STAmount			mSrcAmount;
 
-	OrderBookDB			mOrderBook;
+	//OrderBookDB			mOrderBook;
 	Ledger::pointer		mLedger;
 	PathState::pointer	mPsDefault;
 

src/cpp/ripple/RPCHandler.cpp

@@ -1787,6 +1787,21 @@ Json::Value RPCHandler::doLogin(Json::Value jvRequest)
 }
 #endif
 
+static void textTime(std::string& text, int& seconds, const char *unitName, int unitVal)
+{
+	int i = seconds / unitVal;
+	if (i == 0)
+		return;
+	seconds -= unitVal * i;
+	if (!text.empty())
+		text += ", ";
+	text += boost::lexical_cast<std::string>(i);
+	text += " ";
+	text += unitName;
+	if (i > 1)
+		text += "s";
+}
+
 // {
 //   min_count: <number>  // optional, defaults to 10
 // }
@@ -1808,6 +1823,15 @@ Json::Value RPCHandler::doGetCounts(Json::Value jvRequest)
 	if (dbKB > 0)
 		ret["dbKB"] = dbKB;
 
+	std::string uptime;
+	int s = upTime();
+	textTime(uptime, s, "year", 365*24*60*60);
+	textTime(uptime, s, "day", 24*60*60);
+	textTime(uptime, s, "hour", 24*60);
+	textTime(uptime, s, "minute", 60);
+	textTime(uptime, s, "second", 1);
+	ret["uptime"] = uptime;
+
 	return ret;
 }
 
@@ -2456,6 +2480,25 @@ Json::Value RPCHandler::doSubscribe(Json::Value jvRequest)
 		}
 	}
 
+	if (jvRequest.isMember("books"))
+	{
+		for (Json::Value::iterator it = jvRequest["books"].begin(); it != jvRequest["books"].end(); it++)
+		{
+			uint160 currencyOut;
+			STAmount::issuerFromString(currencyOut,(*it)["CurrencyOut"].asString());
+			uint160 issuerOut=RippleAddress::createNodePublic( (*it)["IssuerOut"].asString() ).getAccountID();
+			uint160 currencyIn;
+			STAmount::issuerFromString(currencyOut,(*it)["CurrencyIn"].asString());
+			uint160 issuerIn=RippleAddress::createNodePublic( (*it)["IssuerIn"].asString() ).getAccountID();
+
+			mNetOps->subBook(ispSub,currencyIn,currencyOut,issuerIn,issuerOut);
+			if((*it)["StateNow"].asBool())
+			{
+
+			}
+		}
+	}
+
 	return jvResult;
 }
 
@@ -2553,6 +2596,21 @@ Json::Value RPCHandler::doUnsubscribe(Json::Value jvRequest)
 		}
 	}
 
+	if (jvRequest.isMember("books"))
+	{
+		for (Json::Value::iterator it = jvRequest["books"].begin(); it != jvRequest["books"].end(); it++)
+		{
+			uint160 currencyOut;
+			STAmount::issuerFromString(currencyOut,(*it)["CurrencyOut"].asString());
+			uint160 issuerOut=RippleAddress::createNodePublic( (*it)["IssuerOut"].asString() ).getAccountID();
+			uint160 currencyIn;
+			STAmount::issuerFromString(currencyOut,(*it)["CurrencyIn"].asString());
+			uint160 issuerIn=RippleAddress::createNodePublic( (*it)["IssuerIn"].asString() ).getAccountID();
+
+			mNetOps->unsubBook(ispSub,currencyIn,currencyOut,issuerIn,issuerOut);
+		}
+	}
+
 	return jvResult;
 }
 

src/cpp/ripple/RippleAddress.cpp

@@ -11,11 +11,11 @@
 #include <openssl/rand.h>
 
 #include "key.h"
-#include "Config.h"
 #include "BitcoinUtil.h"
 #include "rfc1751.h"
 #include "utils.h"
 #include "Log.h"
+#include "Serializer.h"
 
 SETUP_LOG();
 

src/cpp/ripple/Suppression.cpp

@@ -4,6 +4,8 @@
 
 DECLARE_INSTANCE(Suppression);
 
+extern int upTime();
+
 Suppression& SuppressionTable::findCreateEntry(const uint256& index, bool& created)
 {
 	boost::unordered_map<uint256, Suppression>::iterator fit = mSuppressionMap.find(index);
@@ -15,11 +17,11 @@ Suppression& SuppressionTable::findCreateEntry(const uint256& index, bool& creat
 	}
 	created = true;
 
-	time_t now = time(NULL);
+	int now = upTime();
-	time_t expireTime = now - mHoldTime;
+	int expireTime = now - mHoldTime;
 
 	// See if any supressions need to be expired
-	std::map< time_t, std::list<uint256> >::iterator it = mSuppressionTimes.begin();
+	std::map< int, std::list<uint256> >::iterator it = mSuppressionTimes.begin();
 	if ((it != mSuppressionTimes.end()) && (it->first <= expireTime))
 	{
 		BOOST_FOREACH(const uint256& lit, it->second)

src/cpp/ripple/Suppression.h

@@ -50,7 +50,7 @@ protected:
 	boost::unordered_map<uint256, Suppression> mSuppressionMap;
 
 	// Stores all expiration times and the hashes indexed for them
-	std::map< time_t, std::list<uint256> > mSuppressionTimes;
+	std::map< int, std::list<uint256> > mSuppressionTimes;
 
 	int mHoldTime;
 

src/cpp/ripple/TaggedCache.h

@@ -11,6 +11,7 @@
 
 #include "Log.h"
 extern LogPartition TaggedCachePartition;
+extern int upTime();
 
 // This class implements a cache and a map. The cache keeps objects alive
 // in the map. The map allows multiple code paths that reference objects
@@ -36,15 +37,15 @@ protected:
 	class cache_entry
 	{
 	public:
-		time_t			last_use;
+		int				last_use;
 		data_ptr		ptr;
 		weak_data_ptr	weak_ptr;
 
-		cache_entry(time_t l, const data_ptr& d) : last_use(l), ptr(d), weak_ptr(d) { ; }
+		cache_entry(int l, const data_ptr& d) : last_use(l), ptr(d), weak_ptr(d) { ; }
 		bool isCached()		{ return !!ptr; }
 		bool isExpired()	{ return weak_ptr.expired(); }
 		data_ptr lock()		{ return weak_ptr.lock(); }
-		void touch()		{ last_use = time(NULL); }
+		void touch()		{ last_use = upTime(); }
 	};
 
 	typedef std::pair<key_type, cache_entry>				cache_pair;
@@ -59,11 +60,11 @@ protected:
 	int			mCacheCount;	// Number of items cached
 
 	cache_type	mCache;			// Hold strong reference to recent objects
-	time_t		mLastSweep;
+	int			mLastSweep;
 
 public:
 	TaggedCache(const char *name, int size, int age)
-		: mName(name), mTargetSize(size), mTargetAge(age), mCacheCount(0), mLastSweep(time(NULL)) { ; }
+		: mName(name), mTargetSize(size), mTargetAge(age), mCacheCount(0), mLastSweep(upTime()) { ; }
 
 	int getTargetSize() const;
 	int getTargetAge() const;
@@ -128,8 +129,8 @@ template<typename c_Key, typename c_Data> void TaggedCache<c_Key, c_Data>::sweep
 {
 	boost::recursive_mutex::scoped_lock sl(mLock);
 
-	time_t mLastSweep = time(NULL);
+	int mLastSweep = upTime();
-	time_t target = mLastSweep - mTargetAge;
+	int target = mLastSweep - mTargetAge;
 	int cacheRemovals = 0, mapRemovals = 0, cc = 0;
 
 	if ((mTargetSize != 0) && (mCache.size() > mTargetSize))
@@ -243,7 +244,7 @@ bool TaggedCache<c_Key, c_Data>::canonicalize(const key_type& key, boost::shared
 	cache_iterator cit = mCache.find(key);
 	if (cit == mCache.end())
 	{
-		mCache.insert(cache_pair(key, cache_entry(time(NULL), data)));
+		mCache.insert(cache_pair(key, cache_entry(upTime(), data)));
 		++mCacheCount;
 		return false;
 	}

src/cpp/ripple/TransactionMeta.h

@@ -56,6 +56,7 @@ public:
 	void addRaw(Serializer&, TER, uint32 index);
 
 	STObject getAsObject() const;
+	STArray& getNodes(){ return(mNodes); }
 
 	static bool thread(STObject& node, const uint256& prevTxID, uint32 prevLgrID);
 };

test/send-test.js

@@ -40,7 +40,7 @@ buster.testCase("Fee Changes", {
 
 buster.testCase("Sending", {
   'setUp'     : testutils.build_setup(),
-  // 'setUp'     : testutils.build_setup({verbose: true , no_server: true}),
+  //'setUp'     : testutils.build_setup({verbose: true , no_server: true}),
   'tearDown'  : testutils.build_teardown(),
 
   "send XRP to non-existent account with insufficent fee" :