Persistent transactions interface (Gateway REST API)

This commit is contained in:
wltsmrz
2014-02-28 16:27:09 -08:00
parent 37065123e5
commit 5ed148af75
3 changed files with 76 additions and 71 deletions

View File

@@ -190,17 +190,32 @@ function Remote(opts, trace) {
this.on('removeListener', listenerRemoved);
function addPendingAccounts() {
opts.storage.loadAccounts(function(err, accounts) {
if (!err && Array.isArray(accounts)) {
accounts.forEach(self.account.bind(self));
function getPendingTransactions() {
self.storage.getPendingTransactions(function(err, transactions) {
if (err || !Array.isArray(transactions)) return;
var properties = [
'submittedIDs',
'clientID',
'submitIndex'
];
transactions.forEach(function(tx) {
var transaction = self.transaction();
transaction.parseJson(tx.tx_json);
properties.forEach(function(prop) {
if (typeof tx[prop] !== 'undefined') {
transaction[prop] = tx[prop];
}
});
transaction.submit();
});
});
};
if (opts.storage) {
this.storage = opts.storage;
this.once('connect', addPendingAccounts);
this.once('connect', getPendingTransactions);
}
function pingServers() {

View File

@@ -54,7 +54,6 @@ var RippleError = require('./rippleerror').RippleError;
var hashprefixes = require('./hashprefixes');
var config = require('./config');
function Transaction(remote) {
EventEmitter.call(this);
@@ -203,6 +202,7 @@ Transaction.prototype.setState = function(state) {
if (this.state !== state) {
this.state = state;
this.emit('state', state);
this.emit('save');
}
};
@@ -335,7 +335,7 @@ Transaction.prototype.addId = function(hash) {
if (this.submittedIDs.indexOf(hash) === -1) {
this.submittedIDs.unshift(hash);
this.emit('signed', hash);
this.transactionManager()._pending.save();
this.emit('save');
}
};
@@ -724,7 +724,7 @@ Transaction.prototype.walletAdd = function(src, amount, authorized_key, public_k
Transaction.prototype.submit = function(callback) {
var self = this;
this.callback = typeof callback === 'function' ? callback : function(){};
this.callback = (typeof callback === 'function') ? callback : function(){};
function transactionError(error, message) {
if (!(error instanceof RippleError)) {
@@ -761,7 +761,7 @@ Transaction.prototype.transactionManager = function() {
Transaction.prototype.abort = function(callback) {
if (!this.finalized) {
var callback = typeof callback === 'function' ? callback : function(){};
var callback = (typeof callback === 'function') ? callback : function(){};
this.once('final', callback);
this.emit('abort');
}
@@ -772,9 +772,13 @@ Transaction.prototype.iff = function(fn) {
};
Transaction.prototype.summary = function() {
return Transaction.summary.call(this);
};
Transaction.summary = function() {
return {
tx_json: this.tx_json,
clientID: this.clientID,
sourceID: this.sourceID,
submittedIDs: this.submittedIDs,
submissionAttempts: this.attempts,
state: this.state,

View File

@@ -92,7 +92,7 @@ function TransactionManager(account) {
this._remote.on('ledger_closed', updatePendingStatus);
function remoteReconnected(callback) {
var callback = typeof callback === 'function' ? callback : function(){};
var callback = (typeof callback === 'function') ? callback : function(){};
if (!self._pending.length) {
return callback();
@@ -134,34 +134,12 @@ function TransactionManager(account) {
this._remote.on('disconnect', remoteDisconnected);
function resendPending(callback) {
var callback = (typeof callback === 'function') ? callback : function(){};
function accountLoaded(err, data) {
if (err || !(Array.isArray(data))) return;
data.forEach(function(tx) {
var transaction = self._remote.transaction();
transaction.parseJson(tx.tx_json);
transaction.submittedIDs = tx.submittedIDs;
self.submit(transaction);
});
callback();
};
self._remote.storage.loadAccount(self._accountID, accountLoaded);
};
function savePending(pending) {
// Save the current state of the pending transaction list
self._remote.storage.saveAccount(self._accountID, pending);
function saveTransaction(transaction) {
self._remote.storage.saveTransaction(transaction.summary());
};
if (this._remote.storage) {
resendPending(remoteReconnected);
this._pending._save = savePending;
this._saveTx = this._remote.storage.saveTx || function(){};
this.on('save', saveTransaction);
}
};
@@ -407,6 +385,7 @@ TransactionManager.prototype._request = function(tx) {
}
message.result = message.engine_result || '';
tx.result = message;
remote._trace('transactionmanager: submit_response:', message);
@@ -442,16 +421,15 @@ TransactionManager.prototype._request = function(tx) {
if (typeof tx._iff !== 'function') {
submitTransaction();
} else {
return tx._iff(tx.summary(), function(proceed) {
if (proceed) {
submitTransaction();
} else {
return tx._iff(tx.summary(), function(err, proceed) {
if (err || !proceed) {
tx.emit('abort');
} else {
submitTransaction();
}
});
}
function submitTransaction() {
function requestTimeout() {
// ND: What if the response is just slow and we get a response that
// `submitted` above will cause to have concurrent resubmit logic streams?
@@ -473,10 +451,12 @@ TransactionManager.prototype._request = function(tx) {
}
};
tx.attempts++;
function submitTransaction() {
submitRequest.timeout(self._submissionTimeout, requestTimeout);
submitRequest.request();
tx.attempts++;
tx.emit('save', tx.summary());
tx.emit('postsubmit');
};
return submitRequest;
@@ -534,11 +514,16 @@ TransactionManager.prototype.submit = function(tx) {
// ND: We can just remove this `tx` by identity
self._pending.remove(tx);
tx.emit('final', message);
tx.emit('save');
remote._trace('transactionmanager: finalize_transaction:', tx.tx_json);
};
tx.once('cleanup', cleanup);
tx.on('save', function() {
self.emit('save', tx);
});
tx.once('error', function(message) {
tx._errorHandler(message);
});
@@ -552,7 +537,7 @@ TransactionManager.prototype.submit = function(tx) {
});
if (typeof tx.clientID === 'string') {
tx.clientID = [ this._accountID, tx.clientID ].join(':');
tx.sourceID = [ this._accountID, tx.clientID ].join(':');
}
tx.attempts = 0;
@@ -576,6 +561,7 @@ TransactionManager.prototype.submit = function(tx) {
// validated transaction clearing) to fail.
this._pending.push(tx);
this._request(tx);
tx.emit('save');
}
};