machankura/mempool
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved websocket handling to its own file and cleaned up index.ts

Browse Source
This commit is contained in:
softsimon 2020-02-26 17:49:53 +07:00
parent e0726f9493
commit e4bd08cf31
7 changed files with 231 additions and 188 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/package.json
View File

@ -21,6 +21,6 @@
"@types/request": "^2.48.2",
"@types/ws": "^6.0.4",
"tslint": "^5.11.0",
"typescript": "~3.6.4"
"typescript": "^3.6.4"
}
}

22
backend/src/api/disk-cache.ts
View File

@ -1,14 +1,30 @@
import * as fs from 'fs';
import memPool from './mempool';
class DiskCache {
static FILE_NAME = './cache.json';
constructor() { }
saveData(dataBlob: string) {
constructor() {
process.on('SIGINT', () => {
this.saveData(JSON.stringify(memPool.getMempool()));
console.log('Mempool data saved to disk cache');
process.exit(2);
});
}
loadMempoolCache() {
const cacheData = this.loadData();
if (cacheData) {
console.log('Restoring mempool data from disk cache');
memPool.setMempool(JSON.parse(cacheData));
}
}
private saveData(dataBlob: string) {
fs.writeFileSync(DiskCache.FILE_NAME, dataBlob, 'utf8');
}
loadData(): string {
private loadData(): string {
return fs.readFileSync(DiskCache.FILE_NAME, 'utf8');
}
}

1
backend/src/api/fiat-conversion.ts
View File

@ -10,6 +10,7 @@ class FiatConversion {
constructor() { }
public startService() {
console.log('Starting currency rates service');
setInterval(this.updateCurrency.bind(this), 1000 * 60 * 60);
this.updateCurrency();
}

3
backend/src/api/mempool.ts
View File

@ -113,8 +113,7 @@ class Mempool {
}
});
console.log(`New mempool size: ${Object.keys(newMempool).length} ` +
` Change: ${transactions.length - Object.keys(newMempool).length}`);
console.log(`New mempool size: ${Object.keys(newMempool).length} Change: ${diff}`);
this.mempoolCache = newMempool;

5
backend/src/api/statistics.ts
View File

@ -11,10 +11,11 @@ class Statistics {
this.newStatisticsEntryCallback = fn;
}
constructor() {
}
constructor() { }
public startStatistics(): void {
console.log('Starting statistics service');
const now = new Date();
const nextInterval = new Date(now.getFullYear(), now.getMonth(), now.getDate(), now.getHours(),
Math.floor(now.getMinutes() / 1) * 1 + 1, 0, 0);

190
backend/src/api/websocket-handler.ts Normal file
View File

@ -0,0 +1,190 @@
import * as WebSocket from 'ws';
import { Block, TransactionExtended, Statistic } from '../interfaces';
import blocks from './blocks';
import memPool from './mempool';
import mempoolBlocks from './mempool-blocks';
import fiatConversion from './fiat-conversion';
class WebsocketHandler {
private wss: WebSocket.Server | undefined;
constructor() { }
setWebsocketServer(wss: WebSocket.Server) {
this.wss = wss;
}
setupConnectionHandling() {
if (!this.wss) {
throw new Error('WebSocket.Server is not set');
}
this.wss.on('connection', (client: WebSocket) => {
client.on('message', (message: any) => {
try {
const parsedMessage = JSON.parse(message);
if (parsedMessage.action === 'want') {
client['want-blocks'] = parsedMessage.data.indexOf('blocks') > -1;
client['want-mempool-blocks'] = parsedMessage.data.indexOf('mempool-blocks') > -1;
client['want-live-2h-chart'] = parsedMessage.data.indexOf('live-2h-chart') > -1;
client['want-stats'] = parsedMessage.data.indexOf('stats') > -1;
}
if (parsedMessage && parsedMessage['track-tx']) {
if (/^[a-fA-F0-9]{64}$/.test(parsedMessage['track-tx'])) {
client['track-tx'] = parsedMessage['track-tx'];
} else {
client['track-tx'] = null;
}
}
if (parsedMessage && parsedMessage['track-address']) {
if (/^([a-km-zA-HJ-NP-Z1-9]{26,35}|[a-km-zA-HJ-NP-Z1-9]{80}|[a-z]{2,5}1[ac-hj-np-z02-9]{8,87})$/
.test(parsedMessage['track-address'])) {
client['track-address'] = parsedMessage['track-address'];
} else {
client['track-address'] = null;
}
}
if (parsedMessage.action === 'init') {
const _blocks = blocks.getBlocks();
if (!_blocks) {
return;
}
client.send(JSON.stringify({
'mempoolInfo': memPool.getMempoolInfo(),
'vBytesPerSecond': memPool.getVBytesPerSecond(),
'blocks': _blocks,
'conversions': fiatConversion.getTickers()['BTCUSD'],
'mempool-blocks': mempoolBlocks.getMempoolBlocks(),
}));
}
} catch (e) {
console.log(e);
}
});
});
}
handleNewStatistic(stats: Statistic) {
if (!this.wss) {
throw new Error('WebSocket.Server is not set');
}
this.wss.clients.forEach((client: WebSocket) => {
if (client.readyState !== WebSocket.OPEN) {
return;
}
if (!client['want-live-2h-chart']) {
return;
}
client.send(JSON.stringify({
'live-2h-chart': stats
}));
});
}
handleMempoolChange(newMempool: { [txid: string]: TransactionExtended }, newTransactions: TransactionExtended[]) {
if (!this.wss) {
throw new Error('WebSocket.Server is not set');
}
mempoolBlocks.updateMempoolBlocks(newMempool);
const mBlocks = mempoolBlocks.getMempoolBlocks();
const mempoolInfo = memPool.getMempoolInfo();
const vBytesPerSecond = memPool.getVBytesPerSecond();
this.wss.clients.forEach((client: WebSocket) => {
if (client.readyState !== WebSocket.OPEN) {
return;
}
const response = {};
if (client['want-stats']) {
response['mempoolInfo'] = mempoolInfo;
response['vBytesPerSecond'] = vBytesPerSecond;
}
if (client['want-mempool-blocks']) {
response['mempool-blocks'] = mBlocks;
}
// Send all new incoming transactions related to tracked address
if (client['track-address']) {
const foundTransactions: TransactionExtended[] = [];
newTransactions.forEach((tx) => {
const someVin = tx.vin.some((vin) => vin.prevout.scriptpubkey_address === client['track-address']);
if (someVin) {
foundTransactions.push(tx);
return;
}
const someVout = tx.vout.some((vout) => vout.scriptpubkey_address === client['track-address']);
if (someVout) {
foundTransactions.push(tx);
}
});
if (foundTransactions.length) {
response['address-transactions'] = foundTransactions;
}
}
if (Object.keys(response).length) {
client.send(JSON.stringify(response));
}
});
}
handleNewBlock(block: Block, txIds: string[], transactions: TransactionExtended[]) {
if (!this.wss) {
throw new Error('WebSocket.Server is not set');
}
this.wss.clients.forEach((client) => {
if (client.readyState !== WebSocket.OPEN) {
return;
}
if (!client['want-blocks']) {
return;
}
const response = {
'block': block
};
if (client['track-tx'] && txIds.indexOf(client['track-tx']) > -1) {
client['track-tx'] = null;
response['txConfirmed'] = true;
}
if (client['track-address']) {
const foundTransactions: TransactionExtended[] = [];
transactions.forEach((tx) => {
if (tx.vin && tx.vin.some((vin) => vin.prevout.scriptpubkey_address === client['track-address'])) {
foundTransactions.push(tx);
return;
}
if (tx.vout && tx.vout.some((vout) => vout.scriptpubkey_address === client['track-address'])) {
foundTransactions.push(tx);
}
});
if (foundTransactions.length) {
response['address-block-transactions'] = foundTransactions;
}
}
client.send(JSON.stringify(response));
});
}
}
export default new WebsocketHandler();

196
backend/src/index.ts
View File

@ -9,18 +9,15 @@ import * as WebSocket from 'ws';
import routes from './routes';
import blocks from './api/blocks';
import memPool from './api/mempool';
import mempoolBlocks from './api/mempool-blocks';
import diskCache from './api/disk-cache';
import statistics from './api/statistics';
import { Block, TransactionExtended, Statistic } from './interfaces';
import websocketHandler from './api/websocket-handler';
import fiatConversion from './api/fiat-conversion';
class Server {
private wss: WebSocket.Server;
private server: https.Server | http.Server;
private app: any;
wss: WebSocket.Server;
server: https.Server | http.Server;
app: any;
constructor() {
this.app = express();
@ -44,196 +41,35 @@ class Server {
this.wss = new WebSocket.Server({ server: this.server });
}
this.setUpRoutes();
this.setUpHttpApiRoutes();
this.setUpWebsocketHandling();
this.setUpMempoolCache();
this.runMempoolIntervalFunctions();
statistics.startStatistics();
fiatConversion.startService();
diskCache.loadMempoolCache();
this.server.listen(config.HTTP_PORT, () => {
console.log(`Server started on port ${config.HTTP_PORT}`);
});
}
private async runMempoolIntervalFunctions() {
async runMempoolIntervalFunctions() {
await memPool.updateMemPoolInfo();
await blocks.updateBlocks();
await memPool.updateMempool();
setTimeout(this.runMempoolIntervalFunctions.bind(this), config.ELECTRS_POLL_RATE_MS);
}
private setUpMempoolCache() {
const cacheData = diskCache.loadData();
if (cacheData) {
memPool.setMempool(JSON.parse(cacheData));
}
process.on('SIGINT', (options) => {
console.log('SIGINT');
diskCache.saveData(JSON.stringify(memPool.getMempool()));
process.exit(2);
});
setUpWebsocketHandling() {
websocketHandler.setWebsocketServer(this.wss);
websocketHandler.setupConnectionHandling();
statistics.setNewStatisticsEntryCallback(websocketHandler.handleNewStatistic.bind(websocketHandler));
blocks.setNewBlockCallback(websocketHandler.handleNewBlock.bind(websocketHandler));
memPool.setMempoolChangedCallback(websocketHandler.handleMempoolChange.bind(websocketHandler));
}
private setUpWebsocketHandling() {
this.wss.on('connection', (client: WebSocket) => {
client.on('message', (message: any) => {
try {
const parsedMessage = JSON.parse(message);
if (parsedMessage.action === 'want') {
client['want-blocks'] = parsedMessage.data.indexOf('blocks') > -1;
client['want-mempool-blocks'] = parsedMessage.data.indexOf('mempool-blocks') > -1;
client['want-live-2h-chart'] = parsedMessage.data.indexOf('live-2h-chart') > -1;
client['want-stats'] = parsedMessage.data.indexOf('stats') > -1;
}
if (parsedMessage && parsedMessage['track-tx']) {
if (/^[a-fA-F0-9]{64}$/.test(parsedMessage['track-tx'])) {
client['track-tx'] = parsedMessage['track-tx'];
} else {
client['track-tx'] = null;
}
}
if (parsedMessage && parsedMessage['track-address']) {
if (/^([a-km-zA-HJ-NP-Z1-9]{26,35}|[a-km-zA-HJ-NP-Z1-9]{80}|[a-z]{2,5}1[ac-hj-np-z02-9]{8,87})$/
.test(parsedMessage['track-address'])) {
client['track-address'] = parsedMessage['track-address'];
} else {
client['track-address'] = null;
}
}
if (parsedMessage.action === 'init') {
const _blocks = blocks.getBlocks();
if (!_blocks) {
return;
}
client.send(JSON.stringify({
'mempoolInfo': memPool.getMempoolInfo(),
'vBytesPerSecond': memPool.getVBytesPerSecond(),
'blocks': _blocks,
'conversions': fiatConversion.getTickers()['BTCUSD'],
'mempool-blocks': mempoolBlocks.getMempoolBlocks(),
}));
}
} catch (e) {
console.log(e);
}
});
});
statistics.setNewStatisticsEntryCallback((stats: Statistic) => {
this.wss.clients.forEach((client: WebSocket) => {
if (client.readyState !== WebSocket.OPEN) {
return;
}
if (!client['want-live-2h-chart']) {
return;
}
client.send(JSON.stringify({
'live-2h-chart': stats
}));
});
});
blocks.setNewBlockCallback((block: Block, txIds: string[], transactions: TransactionExtended[]) => {
this.wss.clients.forEach((client) => {
if (client.readyState !== WebSocket.OPEN) {
return;
}
if (!client['want-blocks']) {
return;
}
const response = {
'block': block
};
if (client['track-tx'] && txIds.indexOf(client['track-tx']) > -1) {
client['track-tx'] = null;
response['txConfirmed'] = true;
}
if (client['track-address']) {
const foundTransactions: TransactionExtended[] = [];
transactions.forEach((tx) => {
if (tx.vin.some((vin) => vin.prevout.scriptpubkey_address === client['track-address'])) {
foundTransactions.push(tx);
return;
}
if (tx.vout.some((vout) => vout.scriptpubkey_address === client['track-address'])) {
foundTransactions.push(tx);
}
});
if (foundTransactions.length) {
response['address-block-transactions'] = foundTransactions;
}
}
client.send(JSON.stringify(response));
});
});
memPool.setMempoolChangedCallback((newMempool: { [txid: string]: TransactionExtended }, newTransactions: TransactionExtended[]) => {
mempoolBlocks.updateMempoolBlocks(newMempool);
const mBlocks = mempoolBlocks.getMempoolBlocks();
const mempoolInfo = memPool.getMempoolInfo();
const vBytesPerSecond = memPool.getVBytesPerSecond();
this.wss.clients.forEach((client: WebSocket) => {
if (client.readyState !== WebSocket.OPEN) {
return;
}
const response = {};
if (client['want-stats']) {
response['mempoolInfo'] = mempoolInfo;
response['vBytesPerSecond'] = vBytesPerSecond;
}
if (client['want-mempool-blocks']) {
response['mempool-blocks'] = mBlocks;
}
// Send all new incoming transactions related to tracked address
if (client['track-address']) {
const foundTransactions: TransactionExtended[] = [];
newTransactions.forEach((tx) => {
const someVin = tx.vin.some((vin) => vin.prevout.scriptpubkey_address === client['track-address']);
if (someVin) {
foundTransactions.push(tx);
return;
}
const someVout = tx.vout.some((vout) => vout.scriptpubkey_address === client['track-address']);
if (someVout) {
foundTransactions.push(tx);
}
});
if (foundTransactions.length) {
response['address-transactions'] = foundTransactions;
}
}
if (Object.keys(response).length) {
client.send(JSON.stringify(response));
}
});
});
}
private setUpRoutes() {
setUpHttpApiRoutes() {
this.app
.get(config.API_ENDPOINT + 'fees/recommended', routes.getRecommendedFees)
.get(config.API_ENDPOINT + 'fees/mempool-blocks', routes.getMempoolBlocks)
@ -244,8 +80,8 @@ class Server {
.get(config.API_ENDPOINT + 'statistics/3m', routes.get3MStatistics.bind(routes))
.get(config.API_ENDPOINT + 'statistics/6m', routes.get6MStatistics.bind(routes))
.get(config.API_ENDPOINT + 'statistics/1y', routes.get1YStatistics.bind(routes))
;
}
;
}
}
const server = new Server();