Updates, working to get the network code going.

This commit is contained in:
JoelKatz
2011-11-30 21:29:38 -08:00
parent 284445a7fa
commit f3ed8b7ec9
9 changed files with 295 additions and 172 deletions

View File

@@ -2,6 +2,7 @@
#include "Config.h"
#include "PeerDoor.h"
#include "RPCDoor.h"
#include "BitcoinUtil.h"
#include "database/SqliteDatabase.h"
//#include <boost/log/trivial.hpp>
#include <iostream>
@@ -26,7 +27,6 @@ Application::Application()
mKnownNodes.load();
//mUNL.load();
mWallet.load();
mLedgerMaster.load();
mPeerDoor=NULL;
mRPCDoor=NULL;
}

View File

@@ -4,9 +4,7 @@
#include "UniqueNodeList.h"
#include "ConnectionPool.h"
#include "KnownNodeList.h"
#include "LedgerMaster.h"
#include "TimingService.h"
#include "ValidationCollection.h"
#include "Wallet.h"
#include "database/database.h"
@@ -22,10 +20,8 @@ class Application
UniqueNodeList mUNL;
KnownNodeList mKnownNodes;
Wallet mWallet;
ValidationCollection mValidations;
Database* mDatabase;
LedgerMaster mLedgerMaster;
ConnectionPool mConnectionPool;
PeerDoor* mPeerDoor;
@@ -40,9 +36,7 @@ public:
Application();
ConnectionPool& getConnectionPool(){ return(mConnectionPool); }
LedgerMaster& getLedgerMaster(){ return(mLedgerMaster); }
UniqueNodeList& getUNL(){ return(mUNL); }
ValidationCollection& getValidationCollection(){ return(mValidations); }
Wallet& getWallet(){ return(mWallet); }
Database* getDB(){ return(mDatabase); }

304
Peer.cpp
View File

@@ -82,6 +82,199 @@ void Peer::sendPacket(PackedMessage::pointer packet)
}
}
void Peer::start_read_header()
{
mReadbuf.resize(HEADER_SIZE);
asio::async_read(mSocket, asio::buffer(mReadbuf),
boost::bind(&Peer::handle_read_header, shared_from_this(),
asio::placeholders::error));
}
void Peer::start_read_body(unsigned msg_len)
{
// m_readbuf already contains the header in its first HEADER_SIZE
// bytes. Expand it to fit in the body as well, and start async
// read into the body.
//
mReadbuf.resize(HEADER_SIZE + msg_len);
asio::mutable_buffers_1 buf = asio::buffer(&mReadbuf[HEADER_SIZE], msg_len);
asio::async_read(mSocket, buf,
boost::bind(&Peer::handle_read_body, shared_from_this(),
asio::placeholders::error));
}
void Peer::handle_read_header(const boost::system::error_code& error)
{
if(!error)
{
unsigned msg_len = PackedMessage::getLength(mReadbuf);
start_read_body(msg_len);
}else cout << "Peer::connected Error: " << error << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
void Peer::handle_read_body(const boost::system::error_code& error)
{
if(!error)
{
processReadBuffer();
start_read_header();
}else cout << "Peer::connected Error: " << error << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
void Peer::processReadBuffer()
{
int type=PackedMessage::getType(mReadbuf);
switch(type)
{
case newcoin::HELLO:
{
newcoin::TMHello msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvHello(msg);
else cout << "parse error: " << type << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
break;
case newcoin::ERROR_MSG:
{
newcoin::TMErrorMsg msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvErrorMessage(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::PING:
{
newcoin::TMPing msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvPing(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::GET_CONTACTS:
{
newcoin::TMGetContacts msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvGetContacts(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::CONTACT:
{
newcoin::TMContact msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvContact(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::SEARCH_TRANSACTION:
{
newcoin::TMSearchTransaction msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvSearchTransaction(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::GET_ACCOUNT:
{
newcoin::TMGetAccount msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvGetAccount(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::ACCOUNT:
{
newcoin::TMAccount msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvAccount(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::TRANSACTION:
{
newcoin::TMTransaction msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvTransaction(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::GET_LEDGER:
{
newcoin::TMGetLedger msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvGetLedger(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::LEDGER:
{
newcoin::TMLedger msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvLedger(msg);
else cout << "pars error: " << type << endl;
}
#if 0
case newcoin::PROPOSE_LEDGER:
{
newcoin::TM msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recv(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::CLOSE_LEDGER:
{
newcoin::TM msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recv(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::GET_VALIDATION:
{
newcoin::TM msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recv(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::VALIDATION:
{
newcoin::TM msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recv(msg);
else cout << "pars error: " << type << endl;
}
#endif
case newcoin::GET_OBJECT:
{
newcoin::TMGetObjectByHash msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvGetObjectByHash(msg);
else cout << "pars error: " << type << endl;
}
case newcoin::OBJECT:
{
newcoin::TMObjectByHash msg;
if(msg.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
recvObjectByHash(msg);
else cout << "pars error: " << type << endl;
}
default:
cout << "Unknown Msg: " << type << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
}
#if 0
void Peer::sendHello()
{
newcoin::Hello* hello=new newcoin::Hello();
@@ -165,114 +358,6 @@ void Peer::sendGetFullLedger(uint256& hash)
sendPacket(packet);
}
void Peer::start_read_header()
{
mReadbuf.resize(HEADER_SIZE);
asio::async_read(mSocket, asio::buffer(mReadbuf),
boost::bind(&Peer::handle_read_header, shared_from_this(),
asio::placeholders::error));
}
void Peer::start_read_body(unsigned msg_len)
{
// m_readbuf already contains the header in its first HEADER_SIZE
// bytes. Expand it to fit in the body as well, and start async
// read into the body.
//
mReadbuf.resize(HEADER_SIZE + msg_len);
asio::mutable_buffers_1 buf = asio::buffer(&mReadbuf[HEADER_SIZE], msg_len);
asio::async_read(mSocket, buf,
boost::bind(&Peer::handle_read_body, shared_from_this(),
asio::placeholders::error));
}
void Peer::handle_read_header(const boost::system::error_code& error)
{
if(!error)
{
unsigned msg_len = PackedMessage::getLength(mReadbuf);
start_read_body(msg_len);
}else cout << "Peer::connected Error: " << error << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
void Peer::handle_read_body(const boost::system::error_code& error)
{
if(!error)
{
processReadBuffer();
start_read_header();
}else cout << "Peer::connected Error: " << error << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
void Peer::processReadBuffer()
{
int type=PackedMessage::getType(mReadbuf);
switch(type)
{
case newcoin::HELLO:
{
newcoin::Hello hello;
if(hello.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
receiveHello(hello);
else cout << "parse error: " << type << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
break;
case newcoin::TRANSACTION:
{
TransactionPtr trans(new newcoin::Transaction());
if(trans->ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
receiveTransaction(trans);
else cout << "parse error: " << type << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
break;
case newcoin::FULL_LEDGER:
{
newcoin::FullLedger ledger;
if(ledger.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
receiveFullLedger(ledger);
else cout << "parse error: " << type << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
break;
case newcoin::VALIDATION:
{
newcoin::Validation validation;
if(validation.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
receiveValidation(validation);
else cout << "parse error: " << type << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
break;
case newcoin::PROPOSE_LEDGER:
{
newcoin::ProposeLedger prop;
if(prop.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
receiveProposeLedger(prop);
else cout << "parse error: " << type << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
break;
case newcoin::GET_FULL_LEDGER:
{
newcoin::GetFullLedger getFullLedger;
if(getFullLedger.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
receiveGetFullLedger(getFullLedger);
else cout << "parse error: " << type << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
case newcoin::GET_VALIDATIONS:
{
newcoin::GetValidations getValid;
if(getValid.ParseFromArray(&mReadbuf[HEADER_SIZE], mReadbuf.size() - HEADER_SIZE))
receiveGetValidations(getValid);
else cout << "parse error: " << type << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
default:
cout << "Unknown Msg: " << type << endl; //else BOOST_LOG_TRIVIAL(info) << "Error: " << error;
}
}
void Peer::receiveHello(newcoin::Hello& packet)
{
// TODO:6 add this guy to your KNL
@@ -328,6 +413,8 @@ void Peer::receiveFullLedger(newcoin::FullLedger& packet)
theApp->getLedgerMaster().addFullLedger(packet);
}
#endif
void Peer::connectTo(KnownNode& node)
{
tcp::endpoint endpoint( address::from_string(node.mIP), node.mPort);
@@ -335,3 +422,4 @@ void Peer::connectTo(KnownNode& node)
boost::bind(&Peer::connected, this, asio::placeholders::error) );
}

28
Peer.h
View File

@@ -30,7 +30,6 @@ class Peer : public boost::enable_shared_from_this<Peer>
std::list<PackedMessage::pointer> mSendQ;
PackedMessage::pointer mSendingPacket;
Peer(boost::asio::io_service& io_service);
void handle_write(const boost::system::error_code& error, size_t bytes_transferred);
@@ -44,17 +43,26 @@ class Peer : public boost::enable_shared_from_this<Peer>
void sendPacketForce(PackedMessage::pointer packet);
void sendHello();
void sendTransaction();
void sendTransaction(newcoin::TMTransaction& packet);
void sendValidation();
void receiveHello(newcoin::Hello& packet);
void receiveTransaction(TransactionPtr trans);
void receiveValidation(newcoin::Validation& packet);
void receiveFullLedger(newcoin::FullLedger& packet);
void receiveProposeLedger(newcoin::ProposeLedger& packet);
void receiveGetFullLedger(newcoin::GetFullLedger& packet);
void receiveGetValidations(newcoin::GetValidations& packet);
void recvHello(newcoin::TMHello& packet);
void recvTransaction(newcoin::TMTransaction& packet);
void recvValidation(newcoin::TMValidation& packet);
void recvGetValidation(newcoin::TMGetValidations& packet);
void recvContact(newcoin::TMContact& packet);
void recvGetContacts(newcoin::TMGetContacts& packet);
void recvIndexedObject(newcoin::TMIndexedObject& packet);
void recvGetObjectByHash(newcoin::TMGetObjectByHash& packet);
void recvObjectByHash(newcoin::TMObjectByHash& packet);
void recvPing(newcoin::TMPing& packet);
void recvErrorMessage(newcoin::TMErrorMsg& packet);
void recvSearchTransaction(newcoin::TMSearchTransaction& packet);
void recvGetAccount(newcoin::TMGetAccount& packet);
void recvAccount(newcoin::TMAccount& packet);
void recvGetLedger(newcoin::TMGetLedger& packet);
void recvLedger(newcoin::TMLedger& packet);
public:
typedef boost::shared_ptr<Peer> pointer;

View File

@@ -32,22 +32,24 @@ void TimingService::start(boost::asio::io_service& ioService)
void TimingService::handleLedger()
{
cout << "publish ledger" << endl;
#if 0
theApp->getLedgerMaster().startFinalization();
mLedgerTimer->expires_at(mLedgerTimer->expires_at() + boost::posix_time::seconds(theConfig.LEDGER_SECONDS));
mLedgerTimer->async_wait(boost::bind(&TimingService::handleLedger, this));
mPropTimer->expires_at(mLedgerTimer->expires_at() + boost::posix_time::seconds(theConfig.LEDGER_PROPOSAL_DELAY_SECONDS));
mPropTimer->async_wait(boost::bind(&TimingService::handleProp, this));
#endif
}
void TimingService::handleProp()
{
theApp->getLedgerMaster().sendProposal();
// theApp->getLedgerMaster().sendProposal();
}
void TimingService::handleValid()
{
theApp->getLedgerMaster().endFinalization();
// theApp->getLedgerMaster().endFinalization();
}
int TimingService::getCurrentLedgerIndex()

View File

@@ -10,18 +10,19 @@ LocalAccount::LocalAccount(bool) : mAmount(0), mSeqNum(0)
mAddress.SetPubKey(mPublicKey.GetPubKey());
}
#if 0
Wallet::Wallet()
{
}
void Wallet::load()
{
// WRITEME
}
#if 0
int64 Wallet::getBalance()
{
int64 total = 0;

View File

@@ -4,6 +4,7 @@
// file license.txt or http://www.opensource.org/licenses/mit-license.php.
#include "keystore.h"
#include "BitcoinUtil.h"
#include <vector>

View File

@@ -7,35 +7,27 @@ enum MessageType {
PING= 2;
// network presence detection
GET_CONTACTS= 10;
GET_CONTACTS= 10;
CONTACT= 11;
// operations for 'small' nodes
SEARCH_TRANSACTION= 20;
GET_ACCOUNT= 21;
ACCOUNT= 22;
ACCOUNT= 22;
// transaction and ledger processing
TRANSACTION= 30;
GET_LEDGER= 31;
LEDGER= 32;
GET_LEDGER= 31;
LEDGER= 32;
PROPOSE_LEDGER= 33;
CLOSE_LEDGER= 34;
// data replication and synchronization
GET_VALIDATIONS= 40;
VALIDATION= 41;
GET_OBJECT= 42;
OBJECT= 43;
}
message TMBaseMessage {
required MessageType type = 1;
optional uint32 seq = 2;
optional uint32 querySeq = 3; // if this is a reply
required bytes innerMessage = 4;
VALIDATION= 41;
GET_OBJECT= 42;
OBJECT= 43;
}
@@ -55,25 +47,25 @@ you must first combine coins from one address to another.
*/
enum TransactionStatus {
NEW = 1; // origin node did/could not validate
CURRENT = 2; // scheduled to go in this ledger
COMMITED = 3; // in a closed ledger
NEW = 1; // origin node did/could not validate
CURRENT = 2; // scheduled to go in this ledger
COMMITED = 3; // in a closed ledger
REJECT_CONFLICT = 4;
REJECT_INVALID = 5;
REJECT_FUNDS = 6;
HELD_SEQ = 7;
HELD_LEDGER = 8; // held for future ledger
HELD_SEQ = 7;
HELD_LEDGER = 8; // held for future ledger
}
message TMTransaction {
required bytes from = 1;
required bytes dest = 2;
required uint64 amount = 3;
required bytes from = 1;
required bytes dest = 2;
required uint64 amount = 3;
required uint32 sourceLedgerIndex = 4;
required uint32 seqNum = 5;
required uint32 ident = 6;
required bytes pubKey = 7;
required bytes sig = 8;
required uint32 seqNum = 5;
required uint32 ident = 6;
required bytes pubKey = 7;
required bytes sig = 8;
required TransactionStatus status = 9;
optional uint64 receiveTimestamp = 10;
optional uint32 ledgerIndexPossible = 11; // the node may not know
@@ -89,7 +81,7 @@ message TMValidation {
optional uint64 timestamp = 3; // only in proposed ledgers
optional uint32 confidence = 4; // only in proposed ledgers
required bytes hanko = 5;
required bytes sig = 6;
required bytes sig = 6;
}
@@ -103,13 +95,13 @@ message TMGetValidations {
message TMContact {
required bytes pubKey = 1;
required bytes pubKey = 1;
required uint32 softwareVersion = 2;
required uint32 protoVersion = 3;
required uint64 nodeFlags = 4;
required uint64 timestamp = 5;
repeated bytes nodeInfo = 6;
required bytes signature = 7;
required uint64 nodeFlags = 4;
required uint64 timestamp = 5;
repeated bytes nodeInfo = 6;
required bytes signature = 7;
}
// request node information
@@ -119,20 +111,43 @@ message TMGetContacts {
}
message TMSearchTransaction {
required uint32 maxTrans =1;
optional bytes toAccount =2;
optional bytes fromAccount =3;
optional uint32 minLedger =4;
optional bytes fromAcctSeq =5;
repeated bytes transID =6;
}
message TMGetAccount {
repeated bytes acctID =1;
}
message Account {
required bytes accountID =1;
required uint64 balance =2;
required uint32 accountSeq =3;
required uint32 ledgerSeq =4;
}
message TMAccount{
repeated Account accounts =1;
}
message TMIndexedObject
{
enum ObjectType {
TRANSACTION = 1;
TRANSACTION = 1;
TRANSACTION_NODE = 2; // a node in a transaction tree
TRANSACTION_LEAF = 3; // a leaf in a transaction tree
ACCOUNT = 4; // a single account state (with balance/sequence)
ACCOUNT = 4; // a single account state (with balance/sequence)
ACCOUNT_NODE = 5; // a node in an account state tree
ACCOUNT_LEAF = 6; // a leaf in an account state tree
LEDGER = 7;
LEDGER = 7;
}
required bytes hash = 1;
required bytes hash = 1;
required ObjectType type = 2;
}
@@ -148,10 +163,24 @@ message TMGetObjectByHash
message TMObjectByHash
{
required TMIndexedObject object = 1;
required bytes data = 2;
required bytes data = 2;
}
message LedgerNodes {
required bytes nodeid = 1;
required bytes nodedata = 2;
}
message TMGetLedger {
optional bytes hash = 1;
optional uint32 seq = 2;
repeated LedgerNodes nodes = 3;
}
message TMLedger {
required bytes hash = 1;
repeated LedgerNodes nodes = 2;
}
message TMPing {
@@ -167,6 +196,6 @@ message TMPing {
message TMErrorMsg {
optional int32 errorCode = 1;
optional string message = 2;
optional int32 errorCode = 1;
optional string message = 2;
}