Merge pull request #1234 from yemel/refactor/blockchain

Refactor/blockchain
This commit is contained in:
Matias Alejo Garcia 2014-08-29 19:51:20 -03:00
commit ccbbe58fc1
21 changed files with 656 additions and 614 deletions

View file

@ -9,8 +9,7 @@ angular.module('copayApp.controllers').controller('AddressesController',
$scope.loading = true;
w.generateAddress(null, function() {
$timeout(function() {
controllerUtils.setSocketHandlers();
controllerUtils.updateAddressList();
controllerUtils.updateGlobalAddresses();
$scope.loading = false;
}, 1);
});

View file

@ -57,9 +57,6 @@ angular.module('copayApp.controllers').controller('SidebarController', function(
return new Array(num);
}
// Init socket handlers (with no wallet yet)
controllerUtils.setSocketHandlers();
if ($rootScope.wallet) {
$scope.$on('$idleWarn', function(a,countdown) {
if (!(countdown%5))

View file

@ -109,7 +109,9 @@ angular.module('copayApp.controllers').controller('TransactionsController',
var addresses = w.getAddressesStr();
if (addresses.length > 0) {
$scope.blockchain_txs = $scope.wallet.txCache || [];
w.blockchain.getTransactions(addresses, function(txs) {
w.blockchain.getTransactions(addresses, function(err, txs) {
if (err) throw err;
$timeout(function() {
$scope.blockchain_txs = [];
for (var i = 0; i < txs.length; i++) {

View file

@ -1,72 +1,180 @@
'use strict';
var util = require('util');
var async = require('async');
var request = require('request');
var bitcore = require('bitcore');
var coinUtil = bitcore.util;
var io = require('socket.io-client');
var EventEmitter = require('events').EventEmitter;
var preconditions = require('preconditions').singleton();
var http;
if (process.version) {
http = require('http');
};
/*
This class lets interfaces with the blockchain, making general queries and
subscribing to transactions on adressess and blocks.
function Insight(opts) {
opts = opts || {};
this.host = opts.host || 'localhost';
this.port = opts.port || '3001';
this.schema = opts.schema || 'http';
this.retryDelay = opts.retryDelay || 5000;
Opts:
- host
- port
- schema
- reconnection (optional)
- reconnectionDelay (optional)
Events:
- tx: activity on subscribed address.
- block: a new block that includes a subscribed address.
- connect: the connection with the blockchain is ready.
- disconnect: the connection with the blochckain is unavailable.
*/
var Insight = function (opts) {
this.status = this.STATUS.DISCONNECTED;
this.subscribed = {};
this.listeningBlocks = false;
preconditions.checkArgument(opts).shouldBeObject(opts)
.checkArgument(opts.host)
.checkArgument(opts.port)
.checkArgument(opts.schema);
this.url = opts.schema + '://' + opts.host + ':' + opts.port;
this.opts = {
'reconnection': opts.reconnection || true,
'reconnectionDelay': opts.reconnectionDelay || 1000,
'secure': opts.schema === 'https'
};
this.socket = this.getSocket(this.url, this.opts);
// Emmit connection events
var self = this;
this.socket.on('connect', function() {
self.status = self.STATUS.CONNECTED;
self.subscribeToBlocks();
self.emit('connect', 0);
});
this.socket.on('connect_error', function() {
if (self.status != self.STATUS.CONNECTED) return;
self.status = self.STATUS.DISCONNECTED;
self.emit('disconnect');
});
this.socket.on('connect_timeout', function() {
if (self.status != self.STATUS.CONNECTED) return;
self.status = self.STATUS.DISCONNECTED;
self.emit('disconnect');
});
this.socket.on('reconnect', function(attempt) {
if (self.status != self.STATUS.DISCONNECTED) return;
self.status = self.STATUS.CONNECTED;
self.emit('connect', attempt);
});
}
function _asyncForEach(array, fn, callback) {
array = array.slice(0);
util.inherits(Insight, EventEmitter);
function processOne() {
var item = array.pop();
fn(item, function(result) {
if (array.length > 0) {
setTimeout(processOne, 0); // schedule immediately
} else {
callback(); // Done!
}
});
}
if (array.length > 0) {
setTimeout(processOne, 0); // schedule immediately
} else {
callback(); // Done!
}
Insight.prototype.STATUS = {
CONNECTED: 'connected',
DISCONNECTED: 'disconnected',
DESTROYED: 'destroyed'
}
/** @private */
Insight.prototype.subscribeToBlocks = function() {
if (this.listeningBlocks || !this.socket.connected) return;
var self = this;
this.socket.emit('subscribe', 'inv');
this.socket.on('block', function(blockHash) {
self.emit('block', blockHash);
});
this.listeningBlocks = true;
}
/** @private */
Insight.prototype.getSocket = function(url, opts) {
return io(this.url, this.opts);
}
/** @private */
Insight.prototype.request = function(path, cb) {
preconditions.checkArgument(path).shouldBeFunction(cb);
request(this.url + path, cb);
}
/** @private */
Insight.prototype.requestPost = function(path, data, cb) {
preconditions.checkArgument(path).checkArgument(data).shouldBeFunction(cb);
request({method: "POST", url: this.url + path, json: data}, cb);
}
Insight.prototype.destroy = function() {
this.socket.destroy();
this.subscribed = {};
this.status = this.STATUS.DESTROYED;
this.removeAllListeners();
};
Insight.prototype._getOptions = function(method, path, data) {
return {
host: this.host,
port: this.port,
schema: this.schema,
method: method,
path: path,
data: data,
headers: {
'Access-Control-Request-Headers': ''
Insight.prototype.subscribe = function(addresses) {
addresses = Array.isArray(addresses) ? addresses : [addresses];
var self = this;
function handlerFor(self, address) {
return function (txid) {
// verify the address is still subscribed
if (!self.subscribed[address]) return;
self.emit('tx', {address: address, txid: txid});
}
};
}
addresses.forEach(function(address) {
preconditions.checkArgument(new bitcore.Address(address).isValid());
// skip already subscibed
if (!self.subscribed[address]) {
self.subscribed[address] = true;
self.socket.emit('subscribe', address);
self.socket.on(address, handlerFor(self, address));
}
});
};
Insight.prototype.getSubscriptions = function(addresses) {
return Object.keys(this.subscribed);
}
// This is vulneable to txid maneability
// TODO: if ret = false,
// check output address from similar transactions.
//
Insight.prototype.checkSentTx = function(tx, cb) {
var hash = coinUtil.formatHashFull(tx.getHash());
var options = this._getOptions('GET', '/api/tx/' + hash);
Insight.prototype.unsubscribe = function(addresses) {
addresses = Array.isArray(addresses) ? addresses : [addresses];
var self = this;
this._request(options, function(err, res) {
if (err) return cb(err);
var ret = false;
if (res && res.txid === hash) {
ret = hash;
}
return cb(err, ret);
addresses.forEach(function(address) {
preconditions.checkArgument(new bitcore.Address(address).isValid());
self.socket.removeEventListener(address);
delete self.subscribed[address];
});
};
Insight.prototype.unsubscribeAll = function() {
this.unsubscribe(this.getSubscriptions());
};
Insight.prototype.broadcast = function(rawtx, cb) {
preconditions.checkArgument(rawtx);
preconditions.shouldBeFunction(cb);
this.requestPost('/api/tx/send', {rawtx: rawtx}, function(err, res, body) {
if (err || res.statusCode != 200) cb(err || res);
cb(null, body.txid);
});
};
Insight.prototype.getTransaction = function(txid, cb) {
preconditions.shouldBeFunction(cb);
this.request('/api/tx/' + txid, function(err, res, body) {
if (err || res.statusCode != 200 || !body) return cb(err || res);
cb(null, JSON.parse(body));
});
};
@ -75,77 +183,53 @@ Insight.prototype.getTransactions = function(addresses, cb) {
preconditions.shouldBeFunction(cb);
var self = this;
if (!addresses || !addresses.length) return cb([]);
if (!addresses.length) return cb(null, []);
var txids = [];
var txs = [];
_asyncForEach(addresses, function(addr, callback) {
var options = self._getOptions('GET', '/api/addr/' + addr);
self._request(options, function(err, res) {
if (res && res.transactions) {
var txids_tmp = res.transactions;
for (var i = 0; i < txids_tmp.length; i++) {
txids.push(txids_tmp[i]);
}
}
callback();
// Iterator: get a list of transaction ids for an address
function getTransactionIds(address, next) {
self.request('/api/addr/' + address, function(err, res, body) {
if (err || res.statusCode != 200 || !body) return next(err || res);
next(null, JSON.parse(body).transactions);
});
}, function() {
var uniqueTxids = {};
for (var k in txids) {
uniqueTxids[txids[k]] = 1;
}
_asyncForEach(Object.keys(uniqueTxids), function(txid, callback2) {
var options = self._getOptions('GET', '/api/tx/' + txid);
self._request(options, function(err, res) {
txs.push(res);
callback2();
});
}, function() {
return cb(txs);
}
async.map(addresses, getTransactionIds, function then(err, txids) {
if (err) return cb(err);
// txids it's a list of list, let's fix that:
var txidsList = txids.reduce(function(a, r) {
return r.concat(a);
});
// Remove duplicated txids
txidsList = txidsList.filter(function(elem, pos, self) {
return self.indexOf(elem) == pos;
});
// Now get the transactions for that list of txIds
async.map(txidsList, self.getTransaction.bind(self), function then(err, txs) {
if (err) return cb(err);
cb(null, txs);
});
});
};
Insight.prototype.getUnspent = function(addresses, cb) {
if (!addresses || !addresses.length) return cb(null, []);
preconditions.shouldBeArray(addresses);
preconditions.shouldBeFunction(cb);
var all = [];
var options = this._getOptions('POST', '/api/addrs/utxo', 'addrs=' + addresses.join(','));
var self = this;
this._request(options, function(err, res) {
if (err) {
return cb(err);
}
if (res && res.length > 0) {
all = all.concat(res);
}
return cb(null, all);
this.requestPost('/api/addrs/utxo', {addrs: addresses.join(',')}, function(err, res, body) {
if (err || res.statusCode != 200) return cb(err || res);
cb(null, body);
});
};
Insight.prototype.sendRawTransaction = function(rawtx, cb) {
if (!rawtx) throw new Error('rawtx must be set');
Insight.prototype.getActivity = function(addresses, cb) {
preconditions.shouldBeArray(addresses);
var options = this._getOptions('POST', '/api/tx/send', 'rawtx=' + rawtx);
this._request(options, function(err, res) {
if (err) return cb();
this.getTransactions(addresses, function then(err, txs) {
if (err) return cb(err);
return cb(res.txid);
});
};
Insight.prototype.checkActivity = function(addresses, cb) {
if (!addresses) throw new Error('address must be set');
this.getTransactions(addresses, function onResult(txs) {
var flatArray = function(xss) {
return xss.reduce(function(r, xs) {
return r.concat(xs);
@ -177,112 +261,4 @@ Insight.prototype.checkActivity = function(addresses, cb) {
});
};
Insight.prototype._requestNode = function(options, callback) {
if (options.method === 'POST') {
options.headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Content-Length': options.data.length,
};
}
var req = http.request(options, function(response) {
var ret, errTxt, e;
if (response.statusCode == 200 || response.statusCode === 304) {
response.on('data', function(chunk) {
try {
ret = JSON.parse(chunk);
} catch (e2) {
errTxt = 'CRITICAL: Wrong response from insight' + e2;
}
});
} else {
errTxt = "INSIGHT ERROR:" + response.statusCode;
console.log(errTxt);
e = new Error(errTxt);
return callback(e);
}
response.on('end', function() {
if (errTxt) {
console.log("INSIGHT ERROR:" + errTxt);
e = new Error(errTxt);
}
return callback(e, ret);
});
response.on('error', function(e) {
return callback(e, ret);
});
});
if (options.data) {
req.write(options.data);
}
req.end();
};
Insight.prototype._requestBrowser = function(options, callback) {
var self = this;
var request = new XMLHttpRequest();
var url = (options.schema || 'http') + '://' + options.host;
if (options.port !== 80) {
url = url + ':' + options.port;
}
url = url + options.path;
if (options.data && options.method === 'GET') {
url = url + '?' + options.data;
}
request.open(options.method, url, true);
request.timeout = 5000;
request.ontimeout = function() {
setTimeout(function() {
return self._request(options, callback);
}, self.retryDelay);
return callback(new Error('Insight request timeout'));
};
request.onreadystatechange = function() {
if (request.readyState !== 4) return;
var ret, errTxt, e;
if (request.status === 200 || request.status === 304) {
try {
ret = JSON.parse(request.responseText);
} catch (e2) {
errTxt = 'CRITICAL: Wrong response from insight' + e2;
}
} else if (request.status >= 400 && request.status < 499) {
errTxt = 'CRITICAL: Bad request to insight: '+request.status;
} else {
errTxt = 'Error code: ' + request.status + ' - Status: ' + request.statusText + ' - Description: ' + request.responseText;
setTimeout(function() {
return self._request(options, callback);
}, self.retryDelay);
}
if (errTxt) {
console.log("INSIGHT ERROR:", e);
e = new Error(errTxt);
}
return callback(e, ret);
};
if (options.method === 'POST') {
request.setRequestHeader('Content-Type', 'application/x-www-form-urlencoded');
}
request.send(options.data || null);
};
Insight.prototype._request = function(options, callback) {
if (typeof process === 'undefined' || !process.version) {
this._requestBrowser(options, callback);
} else {
this._requestNode(options, callback);
}
};
module.exports = Insight;

View file

@ -193,14 +193,12 @@ Wallet.prototype._getKeyMap = function(txp) {
Wallet.prototype._checkSentTx = function(ntxid, cb) {
var txp = this.txProposals.get(ntxid);
var tx = txp.builder.build();
var txid = bitcore.util.formatHashFull(tx.getHash());
this.blockchain.checkSentTx(tx, function(err, txid) {
var ret = false;
if (txid) {
txp.setSent(txid);
ret = txid;
}
return cb(ret);
this.blockchain.getTransaction(txid, function(err, tx) {
if (err) return cb(false);
txp.setSent(tx.txid);
cb(ret);
});
};
@ -806,7 +804,7 @@ Wallet.prototype.sendTx = function(ntxid, cb) {
this.log('Raw transaction: ', txHex);
var self = this;
this.blockchain.sendRawTransaction(txHex, function(txid) {
this.blockchain.broadcast(txHex, function(err, txid) {
self.log('BITCOIND txid:', txid);
if (txid) {
self.txProposals.get(ntxid).setSent(txid);
@ -1724,7 +1722,7 @@ Wallet.prototype.indexDiscovery = function(start, change, copayerIndex, gap, cb)
// Optimize window to minimize the derivations.
var scanWindow = (lastActive == -1) ? gap : gap - (scanIndex - lastActive) + 1;
var addresses = self.deriveAddresses(scanIndex, scanWindow, change, copayerIndex);
self.blockchain.checkActivity(addresses, function(err, actives) {
self.blockchain.getActivity(addresses, function(err, actives) {
if (err) throw err;
// Check for new activities in the newlly scanned addresses
@ -1752,6 +1750,7 @@ Wallet.prototype.close = function() {
this.log('## CLOSING');
this.lock.release();
this.network.cleanUp();
this.blockchain.destroy();
};
Wallet.prototype.getNetwork = function() {

View file

@ -35,6 +35,7 @@ Network.prototype.cleanUp = function() {
this.criticalErr = '';
this.removeAllListeners();
if (this.socket) {
this.socket.removeAllListeners();
this.socket.disconnect();
this.socket = null;
}

View file

@ -2,7 +2,7 @@
var bitcore = require('bitcore');
angular.module('copayApp.services')
.factory('controllerUtils', function($rootScope, $sce, $location, notification, $timeout, Socket, video, uriHandler) {
.factory('controllerUtils', function($rootScope, $sce, $location, notification, $timeout, video, uriHandler) {
var root = {};
root.getVideoMutedStatus = function(copayer) {
if (!$rootScope.videoInfo) return;
@ -24,8 +24,6 @@ angular.module('copayApp.services')
if ($rootScope.wallet)
$rootScope.wallet.close();
Socket.removeAllListeners();
$rootScope.wallet = null;
delete $rootScope['wallet'];
@ -68,7 +66,7 @@ angular.module('copayApp.services')
uriHandler.register();
$rootScope.unitName = config.unitName;
$rootScope.txAlertCount = 0;
$rootScope.insightError = 0;
$rootScope.reconnecting = false;
$rootScope.isCollapsed = true;
$rootScope.$watch('txAlertCount', function(txAlertCount) {
if (txAlertCount && txAlertCount > 0) {
@ -100,11 +98,9 @@ angular.module('copayApp.services')
root.startNetwork = function(w, $scope) {
Socket.removeAllListeners();
root.setupRootVariables();
root.installStartupHandlers(w, $scope);
root.setSocketHandlers();
root.updateGlobalAddresses();
var handlePeerVideo = function(err, peerID, url) {
if (err) {
@ -125,6 +121,8 @@ angular.module('copayApp.services')
});
w.on('ready', function(myPeerID) {
$rootScope.wallet = w;
root.setConnectionListeners($rootScope.wallet);
if ($rootScope.pendingPayment) {
$location.path('send');
} else {
@ -135,7 +133,7 @@ angular.module('copayApp.services')
});
w.on('publicKeyRingUpdated', function(dontDigest) {
root.setSocketHandlers();
root.updateGlobalAddresses();
if (!dontDigest) {
$rootScope.$digest();
}
@ -199,13 +197,7 @@ angular.module('copayApp.services')
$rootScope.updatingBalance = true;
w.getBalance(function(err, balanceSat, balanceByAddrSat, safeBalanceSat) {
if (err) {
console.error('Error: ' + err.message); //TODO
root._setCommError();
return null;
} else {
root._clearCommError();
}
if (err) throw err;
var satToUnit = 1 / config.unitToSatoshi;
var COIN = bitcore.util.COIN;
@ -298,81 +290,53 @@ angular.module('copayApp.services')
});
}
var connectionLost = false;
$rootScope.$watch('insightError', function(status) {
if (!status) return;
// Reconnected
if (status === -1) {
if (!connectionLost) return; // Skip on first reconnect
connectionLost = false;
root.setConnectionListeners = function(wallet) {
wallet.blockchain.on('connect', function(attempts) {
if (attempts == 0) return;
notification.success('Networking restored', 'Connection to Insight re-established');
return;
$rootScope.reconnecting = false;
root.updateBalance(function() {
$rootScope.$digest();
});
});
wallet.blockchain.on('disconnect', function() {
notification.error('Networking problem', 'Connection to Insight lost, trying to reconnect...');
$rootScope.reconnecting = true;
$rootScope.$digest();
});
wallet.blockchain.on('tx', function(tx) {
notification.funds('Funds received!', tx.address);
root.updateBalance(function() {
$rootScope.$digest();
});
});
if (!$rootScope.wallet.spendUnconfirmed) {
wallet.blockchain.on('block', function(block) {
root.updateBalance(function() {
$rootScope.$digest();
});
});
}
}
// Retry
if (status == 1) return; // Skip the first try
connectionLost = true;
notification.error('Networking problem', 'Connection to Insight lost, reconnecting (attempt number ' + (status - 1) + ')');
});
root._setCommError = function(e) {
if ($rootScope.insightError < 0)
$rootScope.insightError = 0;
$rootScope.insightError++;
};
root._clearCommError = function(e) {
if ($rootScope.insightError > 0)
$rootScope.insightError = -1;
else
$rootScope.insightError = 0;
};
root.setSocketHandlers = function() {
root.updateAddressList();
if (!Socket.sysEventsSet) {
Socket.sysOn('error', root._setCommError);
Socket.sysOn('reconnect_error', root._setCommError);
Socket.sysOn('reconnect_failed', root._setCommError);
Socket.sysOn('connect', root._clearCommError);
Socket.sysOn('reconnect', root._clearCommError);
Socket.sysEventsSet = true;
}
root.updateGlobalAddresses = function() {
if (!$rootScope.wallet) return;
var currentAddrs = Socket.getListeners();
root.updateAddressList();
var currentAddrs = $rootScope.wallet.blockchain.getSubscriptions();
var allAddrs = $rootScope.addrInfos;
var newAddrs = [];
for (var i in allAddrs) {
var a = allAddrs[i];
if (!currentAddrs[a.addressStr])
newAddrs.push(a);
if (!currentAddrs[a.addressStr] && !a.isChange)
newAddrs.push(a.addressStr);
}
for (var i = 0; i < newAddrs.length; i++) {
Socket.emit('subscribe', newAddrs[i].addressStr);
}
newAddrs.forEach(function(a) {
Socket.on(a.addressStr, function(txid) {
if (!a.isChange)
notification.funds('Funds received!', a.addressStr);
root.updateBalance(function() {
$rootScope.$digest();
});
});
});
if (!$rootScope.wallet.spendUnconfirmed && !Socket.isListeningBlocks()) {
Socket.emit('subscribe', 'inv');
Socket.on('block', function(block) {
root.updateBalance(function() {
$rootScope.$digest();
});
});
}
$rootScope.wallet.blockchain.subscribe(newAddrs);
};
return root;
});

View file

@ -1,80 +0,0 @@
'use strict';
angular.module('copayApp.services').factory('Socket',
function($rootScope) {
var listeners = [];
var url = (config.socket.schema || 'http') + '://' + config.socket.host + ':' + config.socket.port;
var opts = {
'reconnection': true,
'reconnectionDelay': config.socket.reconnectDelay || 500,
'secure': config.socket.schema === 'https' ? true : false,
};
var socket = io(url, opts);
return {
on: function(event, callback) {
var wrappedCallback = function() {
var args = arguments;
$rootScope.$apply(function() {
callback.apply(socket, args);
});
};
socket.on(event, wrappedCallback);
if (event !== 'connect') {
listeners.push({
event: event,
fn: wrappedCallback
});
}
},
sysOn: function(event, callback) {
var wrappedCallback = function() {
var args = arguments;
$rootScope.$apply(function() {
callback.apply(socket, args);
});
};
socket.io.on(event, wrappedCallback);
},
getListeners: function() {
var ret = {};
var addrList = listeners
.filter(function(i) {
return i.event != 'block';
})
.map(function(i) {
return i.event;
});
for (var i in addrList) {
ret[addrList[i]] = 1;
}
return ret;
},
isListeningBlocks: function() {
return listeners.filter(function(i) {
return i.event == 'block';
}).length > 0;
},
emit: function(event, data, callback) {
socket.emit(event, data, function() {
var args = arguments;
$rootScope.$apply(function() {
if (callback) {
callback.apply(socket, args);
}
});
});
},
removeAllListeners: function() {
for (var i = 0; i < listeners.length; i++) {
var details = listeners[i];
socket.removeAllListeners(details.event);
}
listeners = [];
}
};
});