status-im
/
embark-area-51
mirror of https://github.com/status-im/embark-area-51.git
2
0
Fork 0
Code Issues Projects Releases Wiki Activity

move blockchain connection code to its own module

This commit is contained in:
Iuri Matias 2018-07-27 16:54:33 -04:00 committed by Pascal Precht
parent a0f209745f
commit b8cbee460b
No known key found for this signature in database
GPG Key ID: 0EE28D8D6FD85D7D
5 changed files with 186 additions and 455 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

280
lib/contracts/blockchain.js
View File

@ -1,280 +0,0 @@
const Web3 = require('web3');
const async = require('async');
const Provider = require('./provider.js');
const utils = require('../utils/utils');
const WEB3_READY = 'web3Ready';
class Blockchain {
constructor(options) {
const self = this;
this.plugins = options.plugins;
this.logger = options.logger;
this.events = options.events;
this.contractsConfig = options.contractsConfig;
this.blockchainConfig = options.blockchainConfig;
this.web3 = options.web3;
this.isDev = options.isDev;
this.web3Endpoint = '';
this.isWeb3Ready = false;
this.web3StartedInProcess = false;
self.events.setCommandHandler("blockchain:web3:isReady", (cb) => {
cb(self.isWeb3Ready);
});
if (!this.web3) {
this.initWeb3();
} else {
this.isWeb3Ready = true;
}
this.registerServiceCheck();
this.registerRequests();
this.registerWeb3Object();
this.registerEvents();
}
initWeb3() {
const self = this;
this.web3 = new Web3();
if (this.contractsConfig.deployment.type !== "rpc" && this.contractsConfig.deployment.type !== "ws") {
const message = __("contracts config error: unknown deployment type %s", this.contractsConfig.deployment.type);
this.logger.error(message);
}
const protocol = (this.contractsConfig.deployment.type === "rpc") ? this.contractsConfig.deployment.protocol : 'ws';
this.web3Endpoint = utils.buildUrl(protocol, this.contractsConfig.deployment.host, this.contractsConfig.deployment.port);//`${protocol}://${this.contractsConfig.deployment.host}:${this.contractsConfig.deployment.port}`;
const providerOptions = {
web3: this.web3,
accountsConfig: this.contractsConfig.deployment.accounts,
blockchainConfig: this.blockchainConfig,
logger: this.logger,
isDev: this.isDev,
type: this.contractsConfig.deployment.type,
web3Endpoint: self.web3Endpoint
};
this.provider = new Provider(providerOptions);
self.events.request("processes:launch", "blockchain", () => {
self.provider.startWeb3Provider(() => {
self.provider.fundAccounts(() => {
self.isWeb3Ready = true;
self.events.emit(WEB3_READY);
self.registerWeb3Object();
});
});
});
}
registerEvents() {
const self = this;
self.events.on('check:wentOffline:Ethereum', () => {
self.logger.trace('Ethereum went offline: stopping web3 provider...');
self.provider.stop();
// once the node goes back online, we can restart the provider
self.events.once('check:backOnline:Ethereum', () => {
self.logger.trace('Ethereum back online: starting web3 provider...');
self.provider.startWeb3Provider(() => {
self.logger.trace('web3 provider restarted after ethereum node came back online');
});
});
});
}
onReady(callback) {
if (this.isWeb3Ready) {
return callback();
}
this.events.once(WEB3_READY, () => {
callback();
});
}
registerServiceCheck() {
const self = this;
const NO_NODE = 'noNode';
this.events.request("services:register", 'Ethereum', function (cb) {
async.waterfall([
function checkNodeConnection(next) {
if (!self.web3.currentProvider) {
return next(NO_NODE, {name: "No Blockchain node found", status: 'off'});
}
next();
},
function checkVersion(next) {
// TODO: web3_clientVersion method is currently not implemented in web3.js 1.0
self.web3._requestManager.send({method: 'web3_clientVersion', params: []}, (err, version) => {
if (err) {
self.isWeb3Ready = false;
return next(null, {name: "Ethereum node (version unknown)", status: 'on'});
}
if (version.indexOf("/") < 0) {
self.events.emit(WEB3_READY);
self.isWeb3Ready = true;
return next(null, {name: version, status: 'on'});
}
let nodeName = version.split("/")[0];
let versionNumber = version.split("/")[1].split("-")[0];
let name = nodeName + " " + versionNumber + " (Ethereum)";
self.events.emit(WEB3_READY);
self.isWeb3Ready = true;
return next(null, {name: name, status: 'on'});
});
}
], (err, statusObj) => {
if (err && err !== NO_NODE) {
return cb(err);
}
cb(statusObj);
});
}, 5000, 'off');
}
registerRequests() {
const self = this;
this.events.setCommandHandler("blockchain:defaultAccount:get", function(cb) {
cb(self.defaultAccount());
});
this.events.setCommandHandler("blockchain:defaultAccount:set", function(account, cb) {
self.setDefaultAccount(account);
cb();
});
this.events.setCommandHandler("blockchain:block:byNumber", function(blockNumber, cb) {
self.getBlock(blockNumber, cb);
});
this.events.setCommandHandler("blockchain:gasPrice", function(cb) {
self.getGasPrice(cb);
});
this.events.setCommandHandler("blockchain:contract:create", function(params, cb) {
cb(self.ContractObject(params));
});
}
defaultAccount() {
return this.web3.eth.defaultAccount;
}
setDefaultAccount(account) {
this.web3.eth.defaultAccount = account;
}
getAccounts(cb) {
this.web3.eth.getAccounts(cb);
}
getCode(address, cb) {
this.web3.eth.getCode(address, cb);
}
getBlock(blockNumber, cb) {
this.web3.eth.getBlock(blockNumber, cb);
}
getGasPrice(cb) {
const self = this;
this.onReady(() => {
self.web3.eth.getGasPrice(cb);
});
}
ContractObject(params) {
return new this.web3.eth.Contract(params.abi, params.address);
}
deployContractObject(contractObject, params) {
return contractObject.deploy({arguments: params.arguments, data: params.data});
}
estimateDeployContractGas(deployObject, cb) {
return deployObject.estimateGas().then((gasValue) => {
cb(null, gasValue);
}).catch(cb);
}
deployContractFromObject(deployContractObject, params, cb) {
const self = this;
let hash;
let calledBacked = false;
function callback(err, receipt) {
if (calledBacked) {
return;
}
if (!err && !receipt.contractAddress) {
return; // Not deployed yet. Need to wait
}
if (interval) {
clearInterval(interval);
}
calledBacked = true;
cb(err, receipt);
}
// This interval is there to compensate for the event that sometimes doesn't get triggered when using WebSocket
// FIXME The issue somehow only happens when the blockchain node is started in the same terminal
const interval = setInterval(() => {
if (!hash) {
return; // Wait until we receive the hash
}
self.web3.eth.getTransactionReceipt(hash, (err, receipt) => {
if (!err && !receipt) {
return; // Transaction is not yet complete
}
callback(err, receipt);
});
}, 500);
deployContractObject.send({
from: params.from, gas: params.gas, gasPrice: params.gasPrice
}, function (err, transactionHash) {
if (err) {
return callback(err);
}
hash = transactionHash;
}).on('receipt', function (receipt) {
if (receipt.contractAddress !== undefined) {
callback(null, receipt);
}
}).then(function (_contract) {
if (!hash) {
return; // Somehow we didn't get the receipt yet... Interval will catch it
}
self.web3.eth.getTransactionReceipt(hash, callback);
}).catch(callback);
}
determineDefaultAccount(cb) {
const self = this;
self.getAccounts(function (err, accounts) {
if (err) {
self.logger.error(err);
return cb(new Error(err));
}
let accountConfig = self.blockchainConfig.account;
let selectedAccount = accountConfig && accountConfig.address;
self.setDefaultAccount(selectedAccount || accounts[0]);
cb();
});
}
registerWeb3Object() {
// doesn't feel quite right, should be a cmd or plugin method
// can just be a command without a callback
this.events.emit("runcode:register", "web3", this.web3);
}
}
module.exports = Blockchain;

144
lib/contracts/provider.js
View File

@ -1,144 +0,0 @@
const async = require('async');
const AccountParser = require('../utils/accountParser');
const fundAccount = require('./fundAccount');
class Provider {
constructor(options) {
this.web3 = options.web3;
this.accountsConfig = options.accountsConfig;
this.blockchainConfig = options.blockchainConfig;
this.type = options.type;
this.web3Endpoint = options.web3Endpoint;
this.logger = options.logger;
this.isDev = options.isDev;
this.engine = new ProviderEngine();
this.asyncMethods = {};
}
startWeb3Provider(callback) {
const self = this;
if (this.type === 'rpc') {
self.engine.addProvider(new RpcSubprovider({
rpcUrl: self.web3Endpoint
}));
} else if (this.type === 'ws') {
//self.engine.addProvider(new WsSubprovider({
console.log('USing ws');
self.addProvider(new SubscriptionSubprovider());
self.addProvider(new WsSubprovider({
rpcUrl: self.web3Endpoint,
origin: this.blockchainConfig.wsOrigins.split(',')[0]
}));
//self.provider = new this.web3.providers.WebsocketProvider(self.web3Endpoint, {headers: {Origin: "embark"}});
} else {
return callback(__("contracts config error: unknown deployment type %s", this.type));
}
// network connectivity error
self.engine.on('error', (err) => {
// report connectivity errors
self.logger.error(err);
});
self.engine.start();
//self.on('error', (err) => {
// console.log('ERR', JSON.stringify(err));
// // report connectivity errors as trace due to polling
// self.logger.trace('web3 provider error: ', err);
// self.logger.trace('stopping web3 provider due to error');
// // prevent continuous polling errors
// self.stop();
//});
//self.web3.setProvider(self);
//self.start();
self.accounts = AccountParser.parseAccountsConfig(self.accountsConfig, self.web3, self.logger);
self.addresses = [];
if (!self.accounts.length) {
return callback();
}
self.accounts.forEach(account => {
self.addresses.push(account.address);
self.web3.eth.accounts.wallet.add(account);
});
self.realAccountFunction = self.web3.eth.getAccounts;
self.web3.eth.getAccounts = function (cb) {
if (!cb) {
cb = function () {
};
}
return new Promise((resolve, reject) => {
self.realAccountFunction((err, accounts) => {
if (err) {
return cb(err);
}
accounts = accounts.concat(self.addresses);
// accounts = self.addresses.concat(accounts);
cb(null, accounts);
resolve(accounts);
});
}
realSend(payload, cb);
};
callback();
}
stop() {
if (this.provider && this.provider.removeAllListeners) {
this.provider.removeAllListeners('connect');
this.provider.removeAllListeners('error');
this.provider.removeAllListeners('end');
this.provider.removeAllListeners('data');
this.provider.responseCallbacks = {};
this.provider = null;
}
}
fundAccounts(callback) {
const self = this;
if (!self.accounts.length) {
return callback();
}
if (!self.isDev) {
return callback();
}
async.each(self.accounts, (account, eachCb) => {
fundAccount(self.web3, account.address, account.hexBalance, eachCb);
}, callback);
}
stop() {
this.engine.stop();
}
eth_accounts(payload, cb) {
return cb(null, this.addresses);
}
sendAsync(payload, callback) {
let method = this.asyncMethods[payload.method];
if (method) {
return method.call(method, payload, (err, result) => {
if (err) {
return callback(err);
}
let response = {'id': payload.id, 'jsonrpc': '2.0', 'result': result};
callback(null, response);
});
}
this.engine.sendAsync.apply(this.engine, arguments);
}
send() {
return this.engine.send.apply(this.engine, arguments);
}
}
module.exports = Provider;

80
lib/modules/blockchain_connector/index.js
View File

@ -1,15 +1,12 @@
const Web3 = require('web3'); const Web3 = require('web3');
const async = require('async'); const async = require('async');
const Provider = require('./provider.js'); const Provider = require('./provider.js');
const utils = require('../../utils/utils'); const utils = require('../utils/utils');
const constants = require('../../constants');
const embarkJsUtils = require('embarkjs').Utils;
const WEB3_READY = 'blockchain:ready'; const WEB3_READY = 'blockchain:ready';
// TODO: consider another name, this is the blockchain connector class Blockchain {
class BlockchainConnector { constructor(options) {
constructor(embark, options) {
const self = this; const self = this;
this.plugins = options.plugins; this.plugins = options.plugins;
this.logger = embark.logger; this.logger = embark.logger;
@ -25,10 +22,6 @@ class BlockchainConnector {
cb(self.isWeb3Ready); cb(self.isWeb3Ready);
}); });
self.events.setCommandHandler("blockchain:object", (cb) => {
cb(self);
});
if (!this.web3) { if (!this.web3) {
this.initWeb3(); this.initWeb3();
} else { } else {
@ -39,7 +32,6 @@ class BlockchainConnector {
this.registerRequests(); this.registerRequests();
this.registerWeb3Object(); this.registerWeb3Object();
this.registerEvents(); this.registerEvents();
this.subscribeToPendingTransactions();
} }
initWeb3(cb) { initWeb3(cb) {
@ -336,14 +328,60 @@ class BlockchainConnector {
} }
deployContractFromObject(deployContractObject, params, cb) { deployContractFromObject(deployContractObject, params, cb) {
embarkJsUtils.secureSend(this.web3, deployContractObject, { const self = this;
let hash;
let calledBacked = false;
function callback(err, receipt) {
if (calledBacked) {
return;
}
if (!err && !receipt.contractAddress) {
return; // Not deployed yet. Need to wait
}
if (interval) {
clearInterval(interval);
}
calledBacked = true;
cb(err, receipt);
}
// This interval is there to compensate for the event that sometimes doesn't get triggered when using WebSocket
// FIXME The issue somehow only happens when the blockchain node is started in the same terminal
const interval = setInterval(() => {
if (!hash) {
return; // Wait until we receive the hash
}
self.web3.eth.getTransactionReceipt(hash, (err, receipt) => {
if (!err && !receipt) {
return; // Transaction is not yet complete
}
callback(err, receipt);
});
}, 500);
deployContractObject.send({
from: params.from, gas: params.gas, gasPrice: params.gasPrice from: params.from, gas: params.gas, gasPrice: params.gasPrice
}, true, cb); }, function (err, transactionHash) {
if (err) {
return callback(err);
}
hash = transactionHash;
}).on('receipt', function (receipt) {
if (receipt.contractAddress !== undefined) {
callback(null, receipt);
}
}).then(function (_contract) {
if (!hash) {
return; // Somehow we didn't get the receipt yet... Interval will catch it
}
self.web3.eth.getTransactionReceipt(hash, callback);
}).catch(callback);
} }
determineDefaultAccount(cb) { determineDefaultAccount(cb) {
const self = this; const self = this;
self.getAccounts(function(err, accounts) { self.getAccounts(function (err, accounts) {
if (err) { if (err) {
self.logger.error(err); self.logger.error(err);
return cb(new Error(err)); return cb(new Error(err));
@ -361,20 +399,6 @@ class BlockchainConnector {
// can just be a command without a callback // can just be a command without a callback
this.events.emit("runcode:register", "web3", this.web3, false); this.events.emit("runcode:register", "web3", this.web3, false);
} }
subscribeToPendingTransactions() {
const self = this;
this.onReady(() => {
if (self.logsSubscription) {
self.logsSubscription.unsubscribe();
}
self.logsSubscription = self.web3.eth
.subscribe('newBlockHeaders', () => {})
.on("data", function (blockHeader) {
self.events.emit('block:header', blockHeader);
});
});
}
} }
BlockchainConnector.ACCEPTED_TYPES = ['rpc', 'ws', 'vm']; BlockchainConnector.ACCEPTED_TYPES = ['rpc', 'ws', 'vm'];

55
lib/modules/blockchain_connector/provider.js
View File

@ -11,13 +11,17 @@ class Provider {
this.web3Endpoint = options.web3Endpoint; this.web3Endpoint = options.web3Endpoint;
this.logger = options.logger; this.logger = options.logger;
this.isDev = options.isDev; this.isDev = options.isDev;
this.engine = new ProviderEngine();
this.asyncMethods = {};
} }
startWeb3Provider(callback) { startWeb3Provider(callback) {
const self = this; const self = this;
if (this.type === 'rpc') { if (this.type === 'rpc') {
self.provider = new this.web3.providers.HttpProvider(self.web3Endpoint); self.engine.addProvider(new RpcSubprovider({
rpcUrl: self.web3Endpoint
}));
} else if (this.type === 'ws') { } else if (this.type === 'ws') {
// Note: don't pass to the provider things like {headers: {Origin: "embark"}}. Origin header is for browser to fill // Note: don't pass to the provider things like {headers: {Origin: "embark"}}. Origin header is for browser to fill
// to protect user, it has no meaning if it is used server-side. See here for more details: https://github.com/ethereum/go-ethereum/issues/16608 // to protect user, it has no meaning if it is used server-side. See here for more details: https://github.com/ethereum/go-ethereum/issues/16608
@ -32,7 +36,26 @@ class Provider {
return callback(__("contracts config error: unknown deployment type %s", this.type)); return callback(__("contracts config error: unknown deployment type %s", this.type));
} }
self.web3.setProvider(self.provider);
// network connectivity error
self.engine.on('error', (err) => {
// report connectivity errors
self.logger.error(err);
});
self.engine.start();
//self.on('error', (err) => {
// console.log('ERR', JSON.stringify(err));
// // report connectivity errors as trace due to polling
// self.logger.trace('web3 provider error: ', err);
// self.logger.trace('stopping web3 provider due to error');
// // prevent continuous polling errors
// self.stop();
//});
//self.web3.setProvider(self);
//self.start();
self.web3.eth.getAccounts((err, accounts) => { self.web3.eth.getAccounts((err, accounts) => {
if (err) { if (err) {
@ -102,10 +125,36 @@ class Provider {
if (!self.isDev) { if (!self.isDev) {
return callback(); return callback();
} }
async.eachLimit(self.accounts, 1, (account, eachCb) => { async.each(self.accounts, (account, eachCb) => {
fundAccount(self.web3, account.address, account.hexBalance, eachCb); fundAccount(self.web3, account.address, account.hexBalance, eachCb);
}, callback); }, callback);
} }
stop() {
this.engine.stop();
}
eth_accounts(payload, cb) {
return cb(null, this.addresses);
}
sendAsync(payload, callback) {
let method = this.asyncMethods[payload.method];
if (method) {
return method.call(method, payload, (err, result) => {
if (err) {
return callback(err);
}
let response = {'id': payload.id, 'jsonrpc': '2.0', 'result': result};
callback(null, response);
});
}
this.engine.sendAsync.apply(this.engine, arguments);
}
send() {
return this.engine.send.apply(this.engine, arguments);
}
} }
module.exports = Provider; module.exports = Provider;

82
lib/modules/blockchain_connector/provider.js~HEAD Normal file
View File

@ -0,0 +1,82 @@
const async = require('async');
const AccountParser = require('../../utils/accountParser');
const fundAccount = require('./fundAccount');
class Provider {
constructor(options) {
this.web3 = options.web3;
this.accountsConfig = options.accountsConfig;
this.blockchainConfig = options.blockchainConfig;
this.type = options.type;
this.web3Endpoint = options.web3Endpoint;
this.logger = options.logger;
this.isDev = options.isDev;
}
startWeb3Provider(callback) {
const self = this;
if (this.type === 'rpc') {
self.provider = new this.web3.providers.HttpProvider(self.web3Endpoint);
} else if (this.type === 'ws') {
self.provider = new this.web3.providers.WebsocketProvider(self.web3Endpoint, {headers: {Origin: "embark"}});
} else {
return callback(__("contracts config error: unknown deployment type %s", this.type));
}
self.web3.setProvider(self.provider);
self.accounts = AccountParser.parseAccountsConfig(self.accountsConfig, self.web3, self.logger);
self.addresses = [];
if (!self.accounts.length) {
return callback();
}
self.accounts.forEach(account => {
self.addresses.push(account.address);
self.web3.eth.accounts.wallet.add(account);
});
self.web3.eth.defaultAccount = self.addresses[0];
const realSend = self.provider.send.bind(self.provider);
self.provider.send = function (payload, cb) {
if (payload.method === 'eth_accounts') {
return realSend(payload, function (err, result) {
if (err) {
return cb(err);
}
result.result = result.result.concat(self.addresses);
cb(null, result);
});
}
realSend(payload, cb);
};
callback();
}
stop() {
if (this.provider && this.provider.removeAllListeners) {
this.provider.removeAllListeners('connect');
this.provider.removeAllListeners('error');
this.provider.removeAllListeners('end');
this.provider.removeAllListeners('data');
this.provider.responseCallbacks = {};
this.provider = null;
}
}
fundAccounts(callback) {
const self = this;
if (!self.accounts.length) {
return callback();
}
if (!self.isDev) {
return callback();
}
async.eachLimit(self.accounts, 1, (account, eachCb) => {
fundAccount(self.web3, account.address, account.hexBalance, eachCb);
}, callback);
}
}
module.exports = Provider;