include amqp-connection-manager
This commit is contained in:
Igor Lins e Silva
+85
-77
@@ -1,91 +1,99 @@
|
||||
import {debugLog, hLog} from "../helpers/common_functions";
|
||||
import got, {HTTPError} from "got";
|
||||
import {connect, Connection} from 'amqplib';
|
||||
import amqp, {ChannelWrapper} from "amqp-connection-manager";
|
||||
import {IAmqpConnectionManager} from "amqp-connection-manager/dist/types/AmqpConnectionManager";
|
||||
|
||||
export async function createConnection(config): Promise<Connection> {
|
||||
try {
|
||||
const amqp_url = getAmpqUrl(config);
|
||||
const conn: Connection = await connect(amqp_url);
|
||||
debugLog("[AMQP] connection established");
|
||||
return conn;
|
||||
} catch (e) {
|
||||
hLog("[AMQP] failed to connect!");
|
||||
hLog(e.message);
|
||||
await new Promise(resolve => setTimeout(resolve, 5000));
|
||||
return await createConnection(config);
|
||||
}
|
||||
export async function createConnection(config): Promise<IAmqpConnectionManager> {
|
||||
try {
|
||||
const amqp_url = getAmpqUrl(config);
|
||||
// const conn: Connection = await connect(amqp_url);
|
||||
const conn: IAmqpConnectionManager = amqp.connect(amqp_url);
|
||||
debugLog("[AMQP] connection established");
|
||||
return conn;
|
||||
} catch (e) {
|
||||
hLog("[AMQP] failed to connect!");
|
||||
hLog(e.message);
|
||||
await new Promise(resolve => setTimeout(resolve, 5000));
|
||||
return await createConnection(config);
|
||||
}
|
||||
}
|
||||
|
||||
export function getAmpqUrl(config): string {
|
||||
let frameMaxValue = '0x10000';
|
||||
if (config.frameMax) {
|
||||
frameMaxValue = config.frameMax;
|
||||
}
|
||||
const u = encodeURIComponent(config.user);
|
||||
const p = encodeURIComponent(config.pass);
|
||||
const v = encodeURIComponent(config.vhost);
|
||||
console.log(`max frame: ${frameMaxValue}`);
|
||||
return `amqp://${u}:${p}@${config.host}/${v}?frameMax=${frameMaxValue}`;
|
||||
let frameMaxValue = '0x10000';
|
||||
if (config.frameMax) {
|
||||
frameMaxValue = config.frameMax;
|
||||
}
|
||||
const u = encodeURIComponent(config.user);
|
||||
const p = encodeURIComponent(config.pass);
|
||||
const v = encodeURIComponent(config.vhost);
|
||||
console.log(`max frame: ${frameMaxValue}`);
|
||||
return `amqp://${u}:${p}@${config.host}/${v}?frameMax=${frameMaxValue}`;
|
||||
}
|
||||
|
||||
async function createChannels(connection) {
|
||||
try {
|
||||
const channel = await connection.createChannel();
|
||||
const confirmChannel = await connection.createConfirmChannel();
|
||||
return [channel, confirmChannel];
|
||||
} catch (e) {
|
||||
hLog("[AMQP] failed to create channels");
|
||||
hLog(e);
|
||||
return null;
|
||||
}
|
||||
async function createChannels(connection: IAmqpConnectionManager) {
|
||||
try {
|
||||
// const channel = await connection.createChannel();
|
||||
const channel = connection.createChannel({
|
||||
confirm: false
|
||||
});
|
||||
// const confirmChannel = await connection.createConfirmChannel();
|
||||
const confirmChannel = connection.createChannel({
|
||||
confirm: true
|
||||
})
|
||||
return [channel, confirmChannel];
|
||||
} catch (e) {
|
||||
hLog("[AMQP] failed to create channels");
|
||||
hLog(e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export async function amqpConnect(onReconnect, config, onClose) {
|
||||
let connection = await createConnection(config);
|
||||
if (connection) {
|
||||
const channels = await createChannels(connection);
|
||||
if (channels) {
|
||||
connection.on('error', (err) => {
|
||||
hLog(err.message);
|
||||
});
|
||||
connection.on('close', () => {
|
||||
hLog('Connection closed!');
|
||||
onClose();
|
||||
setTimeout(async () => {
|
||||
hLog('Retrying in 5 seconds...');
|
||||
const _channels = await amqpConnect(onReconnect, config, onClose);
|
||||
onReconnect(_channels);
|
||||
return _channels;
|
||||
}, 5000);
|
||||
});
|
||||
return channels;
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
export async function amqpConnect(onReconnect, config, onClose): Promise<ChannelWrapper[] | null> {
|
||||
let connection = await createConnection(config);
|
||||
if (connection) {
|
||||
const channels = await createChannels(connection);
|
||||
if (channels) {
|
||||
connection.on('error', (err) => {
|
||||
hLog(err.message);
|
||||
});
|
||||
connection.on('close', () => {
|
||||
hLog('Connection closed!');
|
||||
onClose();
|
||||
setTimeout(async () => {
|
||||
hLog('Retrying in 5 seconds...');
|
||||
const _channels = await amqpConnect(onReconnect, config, onClose);
|
||||
onReconnect(_channels);
|
||||
return _channels;
|
||||
}, 5000);
|
||||
});
|
||||
return channels;
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
} else {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export async function checkQueueSize(q_name, config) {
|
||||
try {
|
||||
const v = encodeURIComponent(config.vhost);
|
||||
const apiUrl = `${config.protocol}://${config.api}/api/queues/${v}/${encodeURIComponent(q_name)}`;
|
||||
const opts = {
|
||||
username: config.user,
|
||||
password: config.pass
|
||||
};
|
||||
const data = await got.get(apiUrl, opts).json() as any;
|
||||
return data.messages;
|
||||
} catch (e) {
|
||||
hLog(`[WARNING] Checking queue size failed! - ${e.message}`);
|
||||
if (e.response && e.response.body) {
|
||||
if (e instanceof HTTPError) {
|
||||
hLog(e.response.body);
|
||||
} else {
|
||||
hLog(JSON.stringify(e.response.body, null, 2));
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
try {
|
||||
const v = encodeURIComponent(config.vhost);
|
||||
const apiUrl = `${config.protocol}://${config.api}/api/queues/${v}/${encodeURIComponent(q_name)}`;
|
||||
const opts = {
|
||||
username: config.user,
|
||||
password: config.pass
|
||||
};
|
||||
const data = await got.get(apiUrl, opts).json() as any;
|
||||
return data.messages;
|
||||
} catch (e) {
|
||||
hLog(`[WARNING] Checking queue size failed! - ${e.message}`);
|
||||
if (e.response && e.response.body) {
|
||||
if (e instanceof HTTPError) {
|
||||
hLog(e.response.body);
|
||||
} else {
|
||||
hLog(JSON.stringify(e.response.body, null, 2));
|
||||
}
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import {StateHistorySocket} from "./state-history";
|
||||
import fetch from 'cross-fetch';
|
||||
import {exec} from "child_process";
|
||||
import {hLog} from "../helpers/common_functions";
|
||||
import {ChannelWrapper} from "amqp-connection-manager";
|
||||
|
||||
export class ConnectionManager {
|
||||
|
||||
@@ -131,7 +132,7 @@ export class ConnectionManager {
|
||||
}
|
||||
}
|
||||
|
||||
async createAMQPChannels(onReconnect, onClose) {
|
||||
async createAMQPChannels(onReconnect, onClose): Promise<ChannelWrapper[]> {
|
||||
return await amqpConnect(onReconnect, this.conn.amqp, onClose);
|
||||
}
|
||||
|
||||
|
||||
Generated
+21
@@ -19,6 +19,7 @@
|
||||
"@fastify/redis": "^5.0.0",
|
||||
"@fastify/swagger": "6.1.0",
|
||||
"@pm2/io": "^5.0.0",
|
||||
"amqp-connection-manager": "^4.1.14",
|
||||
"amqplib": "^0.10.3",
|
||||
"async": "^3.2.4",
|
||||
"base-x": "^4.0.0",
|
||||
@@ -443,6 +444,21 @@
|
||||
"url": "https://github.com/sponsors/epoberezkin"
|
||||
}
|
||||
},
|
||||
"node_modules/amqp-connection-manager": {
|
||||
"version": "4.1.14",
|
||||
"resolved": "https://registry.npmjs.org/amqp-connection-manager/-/amqp-connection-manager-4.1.14.tgz",
|
||||
"integrity": "sha512-1km47dIvEr0HhMUazqovSvNwIlSvDX2APdUpULaINtHpiki1O+cLRaTeXb/jav4OLtH+k6GBXx5gsKOT9kcGKQ==",
|
||||
"dependencies": {
|
||||
"promise-breaker": "^6.0.0"
|
||||
},
|
||||
"engines": {
|
||||
"node": ">=10.0.0",
|
||||
"npm": ">5.0.0"
|
||||
},
|
||||
"peerDependencies": {
|
||||
"amqplib": "*"
|
||||
}
|
||||
},
|
||||
"node_modules/amqplib": {
|
||||
"version": "0.10.3",
|
||||
"resolved": "https://registry.npmjs.org/amqplib/-/amqplib-0.10.3.tgz",
|
||||
@@ -2247,6 +2263,11 @@
|
||||
"resolved": "https://registry.npmjs.org/process-warning/-/process-warning-1.0.0.tgz",
|
||||
"integrity": "sha512-du4wfLyj4yCZq1VupnVSZmRsPJsNuxoDQFdCFHLaYiEbFBD7QE0a+I4D7hOxrVnh78QE/YipFAj9lXHiXocV+Q=="
|
||||
},
|
||||
"node_modules/promise-breaker": {
|
||||
"version": "6.0.0",
|
||||
"resolved": "https://registry.npmjs.org/promise-breaker/-/promise-breaker-6.0.0.tgz",
|
||||
"integrity": "sha512-BthzO9yTPswGf7etOBiHCVuugs2N01/Q/94dIPls48z2zCmrnDptUUZzfIb+41xq0MnYZ/BzmOd6ikDR4ibNZA=="
|
||||
},
|
||||
"node_modules/proxy-addr": {
|
||||
"version": "2.0.7",
|
||||
"resolved": "https://registry.npmjs.org/proxy-addr/-/proxy-addr-2.0.7.tgz",
|
||||
|
||||
@@ -35,6 +35,7 @@
|
||||
"@fastify/redis": "^5.0.0",
|
||||
"@fastify/swagger": "6.1.0",
|
||||
"@pm2/io": "^5.0.0",
|
||||
"amqp-connection-manager": "^4.1.14",
|
||||
"amqplib": "^0.10.3",
|
||||
"async": "^3.2.4",
|
||||
"base-x": "^4.0.0",
|
||||
|
||||
@@ -16,20 +16,21 @@ class DspEventConsumer {
|
||||
console.log('Starting DSP Consumer...');
|
||||
[this.ch] = await this.manager.createAMQPChannels((channels) => {
|
||||
[this.ch] = channels;
|
||||
this.onConnect();
|
||||
this.onConnect().catch(console.log);
|
||||
}, () => {
|
||||
this.ch_ready = false;
|
||||
});
|
||||
this.onConnect();
|
||||
this.onConnect().catch(console.log);
|
||||
}
|
||||
onConnect() {
|
||||
async onConnect() {
|
||||
if (this.conf.settings.dsp_parser) {
|
||||
const q = `${this.manager.chain}:dsp`;
|
||||
console.log(q);
|
||||
this.ch.prefetch(100);
|
||||
this.ch.assertQueue(q, { durable: true });
|
||||
this.ch.consume(q, (data) => {
|
||||
await this.ch.assertQueue(q, { durable: true });
|
||||
await this.ch.consume(q, (data) => {
|
||||
this.onMessage(data);
|
||||
}, {
|
||||
prefetch: 100
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1 +1 @@
|
||||
{"version":3,"file":"dsp-consumer.js","sourceRoot":"","sources":["dsp-consumer.ts"],"names":[],"mappings":";;AAAA,8CAAsD;AACtD,gEAA+D;AAI/D,MAAM,gBAAgB;IAUlB;QALQ,aAAQ,GAAY,KAAK,CAAC;QAC1B,cAAS,GAAG,CAAC,CAAC;QACd,WAAM,GAAG,CAAC,CAAC;QACX,iBAAY,GAAuB,IAAI,GAAG,EAAE,CAAC;QAGjD,MAAM,EAAE,GAAG,IAAI,4BAAmB,EAAE,CAAC;QACrC,IAAI,CAAC,IAAI,GAAG,EAAE,CAAC,MAAM,CAAC;QACtB,IAAI,CAAC,OAAO,GAAG,IAAI,iCAAiB,CAAC,EAAE,CAAC,CAAC;IAC7C,CAAC;IAED,KAAK,CAAC,GAAG;QACL,OAAO,CAAC,GAAG,CAAC,0BAA0B,CAAC,CAAC;QACxC,CAAC,IAAI,CAAC,EAAE,CAAC,GAAG,MAAM,IAAI,CAAC,OAAO,CAAC,kBAAkB,CAAC,CAAC,QAAQ,EAAE,EAAE;YAC3D,CAAC,IAAI,CAAC,EAAE,CAAC,GAAG,QAAQ,CAAC;YACrB,IAAI,CAAC,SAAS,EAAE,CAAC;QACrB,CAAC,EAAE,GAAG,EAAE;YACJ,IAAI,CAAC,QAAQ,GAAG,KAAK,CAAC;QAC1B,CAAC,CAAC,CAAC;QACH,IAAI,CAAC,SAAS,EAAE,CAAC;IACrB,CAAC;IAEO,SAAS;QACb,IAAI,IAAI,CAAC,IAAI,CAAC,QAAQ,CAAC,UAAU,EAAE;YAC/B,MAAM,CAAC,GAAG,GAAG,IAAI,CAAC,OAAO,CAAC,KAAK,MAAM,CAAC;YACtC,OAAO,CAAC,GAAG,CAAC,CAAC,CAAC,CAAC;YACf,IAAI,CAAC,EAAE,CAAC,QAAQ,CAAC,GAAG,CAAC,CAAC;YACtB,IAAI,CAAC,EAAE,CAAC,WAAW,CAAC,CAAC,EAAE,EAAC,OAAO,EAAE,IAAI,EAAC,CAAC,CAAC;YACxC,IAAI,CAAC,EAAE,CAAC,OAAO,CAAC,CAAC,EAAE,CAAC,IAAI,EAAE,EAAE;gBACxB,IAAI,CAAC,SAAS,CAAC,IAAI,CAAC,CAAC;YACzB,CAAC,CAAC,CAAC;SACN;IACL,CAAC;IAED,aAAa,CAAC,SAAiB;QAC3B,UAAU,CAAC,GAAG,EAAE;YACZ,IAAI,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,SAAS,CAAC,EAAE;gBAClC,MAAM,aAAa,GAAG,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,SAAS,CAAC,CAAC,IAAI,CAAC,CAAC,CAAC,EAAE,CAAC,EAAE,EAAE,CAAC,CAAC,CAAC,eAAe,GAAG,CAAC,CAAC,eAAe,CAAC,CAAC;gBAC7G,KAAK,MAAM,MAAM,IAAI,aAAa,EAAE;oBAChC,OAAO,CAAC,GAAG,CAAC,MAAM,CAAC,YAAY,CAAC,EAAE,MAAM,CAAC,SAAS,EAAE,MAAM,CAAC,eAAe,CAAC,CAAC;iBAC/E;gBACD,IAAI,CAAC,YAAY,CAAC,MAAM,CAAC,SAAS,CAAC,CAAC;aACvC;QACL,CAAC,EAAE,GAAG,CAAC,CAAC;IACZ,CAAC;IAED,SAAS,CAAC,GAAY;QAClB,IAAI;YACA,MAAM,OAAO,GAAQ,IAAI,CAAC,KAAK,CAAC,GAAG,CAAC,OAAO,CAAC,QAAQ,EAAE,CAAC,CAAC;YACxD,OAAO,CAAC,GAAG,CAAC,OAAO,CAAC,CAAC;YACrB,IAAI,OAAO,CAAC,SAAS,GAAG,IAAI,CAAC,SAAS,EAAE;gBACpC,IAAI,CAAC,SAAS,GAAG,OAAO,CAAC,SAAS,CAAC;gBACnC,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,OAAO,CAAC,SAAS,EAAE,CAAC,OAAO,CAAC,CAAC,CAAC;aACvD;iBAAM,IAAI,OAAO,CAAC,SAAS,KAAK,IAAI,CAAC,SAAS,EAAE;gBAC7C,IAAI,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,IAAI,CAAC,SAAS,CAAC,EAAE;oBACvC,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,IAAI,CAAC,SAAS,CAAC,CAAC,IAAI,CAAC,OAAO,CAAC,CAAC;iBACvD;aACJ;iBAAM;gBACH,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,OAAO,CAAC,SAAS,CAAC,CAAC,IAAI,CAAC,OAAO,CAAC,CAAC;aAC1D;YACD,IAAI,CAAC,aAAa,CAAC,IAAI,CAAC,SAAS,CAAC,CAAC;YACnC,IAAI,CAAC,EAAE,CAAC,GAAG,CAAC,GAAG,CAAC,CAAC;SACpB;QAAC,OAAO,CAAC,EAAE;YACR,IAAI,CAAC,EAAE,CAAC,IAAI,CAAC,GAAG,CAAC,CAAC;SACrB;IACL,CAAC;CAEJ;AAED,IAAI,gBAAgB,EAAE,CAAC,GAAG,EAAE,CAAC,KAAK,CAAC,OAAO,CAAC,KAAK,CAAC,CAAC"}
|
||||
{"version":3,"file":"dsp-consumer.js","sourceRoot":"","sources":["dsp-consumer.ts"],"names":[],"mappings":";;AAAA,8CAAsD;AACtD,gEAA+D;AAK/D,MAAM,gBAAgB;IAUlB;QALQ,aAAQ,GAAY,KAAK,CAAC;QAC1B,cAAS,GAAG,CAAC,CAAC;QACd,WAAM,GAAG,CAAC,CAAC;QACX,iBAAY,GAAuB,IAAI,GAAG,EAAE,CAAC;QAGjD,MAAM,EAAE,GAAG,IAAI,4BAAmB,EAAE,CAAC;QACrC,IAAI,CAAC,IAAI,GAAG,EAAE,CAAC,MAAM,CAAC;QACtB,IAAI,CAAC,OAAO,GAAG,IAAI,iCAAiB,CAAC,EAAE,CAAC,CAAC;IAC7C,CAAC;IAED,KAAK,CAAC,GAAG;QACL,OAAO,CAAC,GAAG,CAAC,0BAA0B,CAAC,CAAC;QACxC,CAAC,IAAI,CAAC,EAAE,CAAC,GAAG,MAAM,IAAI,CAAC,OAAO,CAAC,kBAAkB,CAAC,CAAC,QAAQ,EAAE,EAAE;YAC3D,CAAC,IAAI,CAAC,EAAE,CAAC,GAAG,QAAQ,CAAC;YACrB,IAAI,CAAC,SAAS,EAAE,CAAC,KAAK,CAAC,OAAO,CAAC,GAAG,CAAC,CAAC;QACxC,CAAC,EAAE,GAAG,EAAE;YACJ,IAAI,CAAC,QAAQ,GAAG,KAAK,CAAC;QAC1B,CAAC,CAAC,CAAC;QACH,IAAI,CAAC,SAAS,EAAE,CAAC,KAAK,CAAC,OAAO,CAAC,GAAG,CAAC,CAAC;IACxC,CAAC;IAEO,KAAK,CAAC,SAAS;QACnB,IAAI,IAAI,CAAC,IAAI,CAAC,QAAQ,CAAC,UAAU,EAAE;YAC/B,MAAM,CAAC,GAAG,GAAG,IAAI,CAAC,OAAO,CAAC,KAAK,MAAM,CAAC;YACtC,OAAO,CAAC,GAAG,CAAC,CAAC,CAAC,CAAC;YACf,MAAM,IAAI,CAAC,EAAE,CAAC,WAAW,CAAC,CAAC,EAAE,EAAC,OAAO,EAAE,IAAI,EAAC,CAAC,CAAC;YAC9C,MAAM,IAAI,CAAC,EAAE,CAAC,OAAO,CAAC,CAAC,EAAE,CAAC,IAAI,EAAE,EAAE;gBAC9B,IAAI,CAAC,SAAS,CAAC,IAAI,CAAC,CAAC;YACzB,CAAC,EAAE;gBACC,QAAQ,EAAE,GAAG;aAChB,CAAC,CAAC;SACN;IACL,CAAC;IAED,aAAa,CAAC,SAAiB;QAC3B,UAAU,CAAC,GAAG,EAAE;YACZ,IAAI,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,SAAS,CAAC,EAAE;gBAClC,MAAM,aAAa,GAAG,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,SAAS,CAAC,CAAC,IAAI,CAAC,CAAC,CAAC,EAAE,CAAC,EAAE,EAAE,CAAC,CAAC,CAAC,eAAe,GAAG,CAAC,CAAC,eAAe,CAAC,CAAC;gBAC7G,KAAK,MAAM,MAAM,IAAI,aAAa,EAAE;oBAChC,OAAO,CAAC,GAAG,CAAC,MAAM,CAAC,YAAY,CAAC,EAAE,MAAM,CAAC,SAAS,EAAE,MAAM,CAAC,eAAe,CAAC,CAAC;iBAC/E;gBACD,IAAI,CAAC,YAAY,CAAC,MAAM,CAAC,SAAS,CAAC,CAAC;aACvC;QACL,CAAC,EAAE,GAAG,CAAC,CAAC;IACZ,CAAC;IAED,SAAS,CAAC,GAAY;QAClB,IAAI;YACA,MAAM,OAAO,GAAQ,IAAI,CAAC,KAAK,CAAC,GAAG,CAAC,OAAO,CAAC,QAAQ,EAAE,CAAC,CAAC;YACxD,OAAO,CAAC,GAAG,CAAC,OAAO,CAAC,CAAC;YACrB,IAAI,OAAO,CAAC,SAAS,GAAG,IAAI,CAAC,SAAS,EAAE;gBACpC,IAAI,CAAC,SAAS,GAAG,OAAO,CAAC,SAAS,CAAC;gBACnC,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,OAAO,CAAC,SAAS,EAAE,CAAC,OAAO,CAAC,CAAC,CAAC;aACvD;iBAAM,IAAI,OAAO,CAAC,SAAS,KAAK,IAAI,CAAC,SAAS,EAAE;gBAC7C,IAAI,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,IAAI,CAAC,SAAS,CAAC,EAAE;oBACvC,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,IAAI,CAAC,SAAS,CAAC,CAAC,IAAI,CAAC,OAAO,CAAC,CAAC;iBACvD;aACJ;iBAAM;gBACH,IAAI,CAAC,YAAY,CAAC,GAAG,CAAC,OAAO,CAAC,SAAS,CAAC,CAAC,IAAI,CAAC,OAAO,CAAC,CAAC;aAC1D;YACD,IAAI,CAAC,aAAa,CAAC,IAAI,CAAC,SAAS,CAAC,CAAC;YACnC,IAAI,CAAC,EAAE,CAAC,GAAG,CAAC,GAAG,CAAC,CAAC;SACpB;QAAC,OAAO,CAAC,EAAE;YACR,IAAI,CAAC,EAAE,CAAC,IAAI,CAAC,GAAG,CAAC,CAAC;SACrB;IACL,CAAC;CAEJ;AAED,IAAI,gBAAgB,EAAE,CAAC,GAAG,EAAE,CAAC,KAAK,CAAC,OAAO,CAAC,KAAK,CAAC,CAAC"}
|
||||
+10
-8
@@ -1,13 +1,14 @@
|
||||
import {ConfigurationModule} from "../modules/config";
|
||||
import {ConnectionManager} from "../connections/manager.class";
|
||||
import {HyperionConfig} from "../interfaces/hyperionConfig";
|
||||
import {Channel, Message} from "amqplib/callback_api";
|
||||
import {Message} from "amqplib/callback_api";
|
||||
import {ChannelWrapper} from "amqp-connection-manager";
|
||||
|
||||
class DspEventConsumer {
|
||||
|
||||
private conf: HyperionConfig;
|
||||
private manager: ConnectionManager;
|
||||
private ch: Channel;
|
||||
private ch: ChannelWrapper;
|
||||
private ch_ready: boolean = false;
|
||||
private lastBlock = 0;
|
||||
private lastGS = 0;
|
||||
@@ -23,21 +24,22 @@ class DspEventConsumer {
|
||||
console.log('Starting DSP Consumer...');
|
||||
[this.ch] = await this.manager.createAMQPChannels((channels) => {
|
||||
[this.ch] = channels;
|
||||
this.onConnect();
|
||||
this.onConnect().catch(console.log);
|
||||
}, () => {
|
||||
this.ch_ready = false;
|
||||
});
|
||||
this.onConnect();
|
||||
this.onConnect().catch(console.log);
|
||||
}
|
||||
|
||||
private onConnect() {
|
||||
private async onConnect(): Promise<void> {
|
||||
if (this.conf.settings.dsp_parser) {
|
||||
const q = `${this.manager.chain}:dsp`;
|
||||
console.log(q);
|
||||
this.ch.prefetch(100);
|
||||
this.ch.assertQueue(q, {durable: true});
|
||||
this.ch.consume(q, (data) => {
|
||||
await this.ch.assertQueue(q, {durable: true});
|
||||
await this.ch.consume(q, (data) => {
|
||||
this.onMessage(data);
|
||||
}, {
|
||||
prefetch: 100
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -29,9 +29,11 @@ export default class MainDSWorker extends HyperionWorker {
|
||||
if (this.ch) {
|
||||
this.queueName = this.chain + ":delta_rm";
|
||||
hLog(`Launched delta updater, consuming from ${this.queueName}`);
|
||||
this.ch.assertQueue(this.queueName, {durable: true});
|
||||
this.ch.prefetch(1);
|
||||
this.ch.consume(this.queueName, this.onConsume.bind(this));
|
||||
this.ch.assertQueue(this.queueName, {durable: true}).then(() => {
|
||||
this.ch.consume(this.queueName, this.onConsume.bind(this),{
|
||||
prefetch: 1
|
||||
}).catch(console.log);
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -286,10 +286,11 @@ export default class MainDSWorker extends HyperionWorker {
|
||||
|
||||
private initConsumer() {
|
||||
if (this.ch_ready) {
|
||||
this.ch.prefetch(this.conf.prefetch.block);
|
||||
this.ch.consume(process.env['worker_queue'], (data) => {
|
||||
this.consumerQueue.push(data).catch(console.log);
|
||||
});
|
||||
}, {
|
||||
prefetch: this.conf.prefetch.block
|
||||
}).catch(console.log);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -625,7 +626,7 @@ export default class MainDSWorker extends HyperionWorker {
|
||||
if (this.ch_ready) {
|
||||
const enqueueResult = this.ch.sendToQueue(pool_queue, bufferFromJson(trace, true), {headers});
|
||||
if (!enqueueResult) {
|
||||
hLog("Failed to send trace!");
|
||||
hLog("Backpressure");
|
||||
console.log("Header size: " + JSON.stringify(headers).length);
|
||||
console.log(headers);
|
||||
}
|
||||
|
||||
+7
-7
@@ -625,12 +625,11 @@ export default class DSPoolWorker extends HyperionWorker {
|
||||
|
||||
initConsumer() {
|
||||
if (this.ch_ready) {
|
||||
this.ch.prefetch(this.conf.prefetch.block);
|
||||
this.ch.consume(this.local_queue, (data) => {
|
||||
this.consumerQueue.push(data);
|
||||
}, {}, (err, ok) => {
|
||||
hLog(err, ok);
|
||||
});
|
||||
}, {
|
||||
prefetch: this.conf.prefetch.block
|
||||
}).catch(console.log);
|
||||
debugLog(`started consuming from ${this.local_queue}`);
|
||||
}
|
||||
}
|
||||
@@ -694,13 +693,14 @@ export default class DSPoolWorker extends HyperionWorker {
|
||||
this.ch_ready = true;
|
||||
this.ch.assertQueue(this.local_queue, {
|
||||
durable: true
|
||||
});
|
||||
this.initConsumer();
|
||||
}).then(() => {
|
||||
this.initConsumer();
|
||||
}).catch(console.log);
|
||||
}
|
||||
if (this.conf.settings.dsp_parser) {
|
||||
this.ch.assertQueue(`${queue_prefix}:dsp`, {
|
||||
durable: true
|
||||
});
|
||||
}).catch(console.log);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+229
-226
@@ -11,277 +11,280 @@ import {HeapInfo} from "v8";
|
||||
import {debugLog, hLog} from "../helpers/common_functions";
|
||||
import {StateHistorySocket} from "../connections/state-history";
|
||||
import * as AbiEOS from "@eosrio/node-abieos";
|
||||
import {ChannelWrapper} from "amqp-connection-manager";
|
||||
|
||||
export abstract class HyperionWorker {
|
||||
|
||||
conf: HyperionConfig;
|
||||
manager: ConnectionManager;
|
||||
mLoader: HyperionModuleLoader;
|
||||
chain: string;
|
||||
chainId: string;
|
||||
conf: HyperionConfig;
|
||||
manager: ConnectionManager;
|
||||
mLoader: HyperionModuleLoader;
|
||||
chain: string;
|
||||
chainId: string;
|
||||
|
||||
// AMQP Channels
|
||||
ch: Channel;
|
||||
cch: ConfirmChannel;
|
||||
// AMQP Channels
|
||||
ch: ChannelWrapper;
|
||||
cch: ChannelWrapper;
|
||||
|
||||
rpc: JsonRpc;
|
||||
client: Client;
|
||||
ship: StateHistorySocket;
|
||||
rpc: JsonRpc;
|
||||
client: Client;
|
||||
ship: StateHistorySocket;
|
||||
|
||||
txEnc = new TextEncoder();
|
||||
txDec = new TextDecoder();
|
||||
cch_ready = false;
|
||||
ch_ready = false;
|
||||
txEnc = new TextEncoder();
|
||||
txDec = new TextDecoder();
|
||||
cch_ready = false;
|
||||
ch_ready = false;
|
||||
|
||||
events: EventEmitter;
|
||||
events: EventEmitter;
|
||||
|
||||
filters: Filters;
|
||||
filters: Filters;
|
||||
|
||||
failedAbiMap: Map<string, Set<number>> = new Map();
|
||||
failedAbiMap: Map<string, Set<number>> = new Map();
|
||||
|
||||
protected constructor() {
|
||||
this.checkDebugger();
|
||||
const cm = new ConfigurationModule();
|
||||
this.conf = cm.config;
|
||||
this.filters = cm.filters;
|
||||
this.manager = new ConnectionManager(cm);
|
||||
this.mLoader = new HyperionModuleLoader(cm);
|
||||
this.chain = this.conf.settings.chain;
|
||||
this.chainId = this.manager.conn.chains[this.chain].chain_id;
|
||||
this.rpc = this.manager.nodeosJsonRPC;
|
||||
this.client = this.manager.elasticsearchClient;
|
||||
this.ship = this.manager.shipClient;
|
||||
this.events = new EventEmitter();
|
||||
protected constructor() {
|
||||
this.checkDebugger();
|
||||
const cm = new ConfigurationModule();
|
||||
this.conf = cm.config;
|
||||
this.filters = cm.filters;
|
||||
this.manager = new ConnectionManager(cm);
|
||||
this.mLoader = new HyperionModuleLoader(cm);
|
||||
this.chain = this.conf.settings.chain;
|
||||
this.chainId = this.manager.conn.chains[this.chain].chain_id;
|
||||
this.rpc = this.manager.nodeosJsonRPC;
|
||||
this.client = this.manager.elasticsearchClient;
|
||||
this.ship = this.manager.shipClient;
|
||||
this.events = new EventEmitter();
|
||||
|
||||
this.mLoader.init().then(() => {
|
||||
// Connect to RabbitMQ (amqplib)
|
||||
this.events.emit('loader_ready');
|
||||
this.connectAMQP().then(() => {
|
||||
this.onConnect();
|
||||
}).catch(console.log);
|
||||
});
|
||||
this.mLoader.init().then(() => {
|
||||
// Connect to RabbitMQ (amqplib)
|
||||
this.events.emit('loader_ready');
|
||||
this.connectAMQP().then(() => {
|
||||
this.onConnect();
|
||||
}).catch(console.log);
|
||||
});
|
||||
|
||||
const bytesToString = (bytes: number) => {
|
||||
const e = Math.log(bytes) / Math.log(1024) | 0;
|
||||
const n = (bytes / Math.pow(1024, e)).toFixed(2);
|
||||
return n + ' ' + (e == 0 ? 'bytes' : (['KB', 'MB', 'GB', 'TB'])[e - 1]);
|
||||
};
|
||||
const bytesToString = (bytes: number) => {
|
||||
const e = Math.log(bytes) / Math.log(1024) | 0;
|
||||
const n = (bytes / Math.pow(1024, e)).toFixed(2);
|
||||
return n + ' ' + (e == 0 ? 'bytes' : (['KB', 'MB', 'GB', 'TB'])[e - 1]);
|
||||
};
|
||||
|
||||
|
||||
// handle ipc messages
|
||||
process.on('message', (msg: any) => {
|
||||
switch (msg.event) {
|
||||
case 'request_v8_heap_stats': {
|
||||
const report: HeapInfo = v8.getHeapStatistics();
|
||||
const used_pct = report.used_heap_size / report.heap_size_limit;
|
||||
process.send({
|
||||
event: 'v8_heap_report',
|
||||
id: process.env.worker_role + ':' + process.env.worker_id,
|
||||
data: {
|
||||
heap_usage: (used_pct * 100).toFixed(2) + "%",
|
||||
...report
|
||||
}
|
||||
});
|
||||
break;
|
||||
}
|
||||
case 'request_memory_usage': {
|
||||
const report = process.memoryUsage();
|
||||
process.send({
|
||||
event: 'memory_report',
|
||||
id: process.env.worker_role + ':' + process.env.worker_id,
|
||||
data: {
|
||||
resident: bytesToString(report.rss),
|
||||
}
|
||||
});
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
this.onIpcMessage(msg);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
// handle ipc messages
|
||||
process.on('message', (msg: any) => {
|
||||
switch (msg.event) {
|
||||
case 'request_v8_heap_stats': {
|
||||
const report: HeapInfo = v8.getHeapStatistics();
|
||||
const used_pct = report.used_heap_size / report.heap_size_limit;
|
||||
process.send({
|
||||
event: 'v8_heap_report',
|
||||
id: process.env.worker_role + ':' + process.env.worker_id,
|
||||
data: {
|
||||
heap_usage: (used_pct * 100).toFixed(2) + "%",
|
||||
...report
|
||||
}
|
||||
});
|
||||
break;
|
||||
}
|
||||
case 'request_memory_usage': {
|
||||
const report = process.memoryUsage();
|
||||
process.send({
|
||||
event: 'memory_report',
|
||||
id: process.env.worker_role + ':' + process.env.worker_id,
|
||||
data: {
|
||||
resident: bytesToString(report.rss),
|
||||
}
|
||||
});
|
||||
break;
|
||||
}
|
||||
default: {
|
||||
this.onIpcMessage(msg);
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
async connectAMQP() {
|
||||
[this.ch, this.cch] = await this.manager.createAMQPChannels((channels) => {
|
||||
[this.ch, this.cch] = channels;
|
||||
hLog('AMQP Reconnecting...');
|
||||
this.onConnect();
|
||||
}, () => {
|
||||
this.ch_ready = false;
|
||||
this.cch_ready = false;
|
||||
});
|
||||
}
|
||||
async connectAMQP() {
|
||||
[this.ch, this.cch] = await this.manager.createAMQPChannels((channels: ChannelWrapper[]) => {
|
||||
if (channels) {
|
||||
[this.ch, this.cch] = channels;
|
||||
hLog('AMQP Reconnecting...');
|
||||
this.onConnect();
|
||||
}
|
||||
}, () => {
|
||||
this.ch_ready = false;
|
||||
this.cch_ready = false;
|
||||
});
|
||||
}
|
||||
|
||||
onConnect() {
|
||||
this.ch_ready = true;
|
||||
this.cch_ready = true;
|
||||
this.assertQueues();
|
||||
this.ch.on('close', () => {
|
||||
this.ch_ready = false;
|
||||
});
|
||||
this.cch.on('close', () => {
|
||||
this.cch_ready = false;
|
||||
});
|
||||
this.events.emit('ready');
|
||||
}
|
||||
onConnect() {
|
||||
this.ch_ready = true;
|
||||
this.cch_ready = true;
|
||||
this.assertQueues();
|
||||
this.ch.on('close', () => {
|
||||
this.ch_ready = false;
|
||||
});
|
||||
this.cch.on('close', () => {
|
||||
this.cch_ready = false;
|
||||
});
|
||||
this.events.emit('ready');
|
||||
}
|
||||
|
||||
checkDebugger() {
|
||||
if (/--inspect/.test(process.execArgv.join(' '))) {
|
||||
const inspector = require('inspector');
|
||||
hLog('DEBUGGER ATTACHED', inspector.url());
|
||||
}
|
||||
}
|
||||
checkDebugger() {
|
||||
if (/--inspect/.test(process.execArgv.join(' '))) {
|
||||
const inspector = require('inspector');
|
||||
hLog('DEBUGGER ATTACHED', inspector.url());
|
||||
}
|
||||
}
|
||||
|
||||
private anyFromCode(act: any) {
|
||||
return this.chain + '::' + act.account + '::*'
|
||||
}
|
||||
private anyFromCode(act: any) {
|
||||
return this.chain + '::' + act.account + '::*'
|
||||
}
|
||||
|
||||
private anyFromName(act: any) {
|
||||
return this.chain + '::*::' + act.name;
|
||||
}
|
||||
private anyFromName(act: any) {
|
||||
return this.chain + '::*::' + act.name;
|
||||
}
|
||||
|
||||
private codeActionPair(act: any) {
|
||||
return this.chain + '::' + act.account + '::' + act.name;
|
||||
}
|
||||
private codeActionPair(act: any) {
|
||||
return this.chain + '::' + act.account + '::' + act.name;
|
||||
}
|
||||
|
||||
private anyFromDeltaCode(delta: any) {
|
||||
return this.chain + '::' + delta.code + '::*'
|
||||
}
|
||||
private anyFromDeltaCode(delta: any) {
|
||||
return this.chain + '::' + delta.code + '::*'
|
||||
}
|
||||
|
||||
private anyFromDeltaTable(delta: any) {
|
||||
return this.chain + '::*::' + delta.table;
|
||||
}
|
||||
private anyFromDeltaTable(delta: any) {
|
||||
return this.chain + '::*::' + delta.table;
|
||||
}
|
||||
|
||||
private codeDeltaPair(delta: any) {
|
||||
return this.chain + '::' + delta.code + '::' + delta.table;
|
||||
}
|
||||
private codeDeltaPair(delta: any) {
|
||||
return this.chain + '::' + delta.code + '::' + delta.table;
|
||||
}
|
||||
|
||||
protected checkBlacklist(act) {
|
||||
protected checkBlacklist(act) {
|
||||
|
||||
// test for chain::code::*
|
||||
if (this.filters.action_blacklist.has(this.anyFromCode(act))) {
|
||||
return true;
|
||||
}
|
||||
// test for chain::code::*
|
||||
if (this.filters.action_blacklist.has(this.anyFromCode(act))) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// test for chain::*::name
|
||||
if (this.filters.action_blacklist.has(this.anyFromName(act))) {
|
||||
return true;
|
||||
}
|
||||
// test for chain::*::name
|
||||
if (this.filters.action_blacklist.has(this.anyFromName(act))) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// test for chain::code::name
|
||||
return this.filters.action_blacklist.has(this.codeActionPair(act));
|
||||
}
|
||||
// test for chain::code::name
|
||||
return this.filters.action_blacklist.has(this.codeActionPair(act));
|
||||
}
|
||||
|
||||
protected checkWhitelist(act) {
|
||||
protected checkWhitelist(act) {
|
||||
|
||||
// test for chain::code::*
|
||||
if (this.filters.action_whitelist.has(this.anyFromCode(act))) {
|
||||
return true;
|
||||
}
|
||||
// test for chain::code::*
|
||||
if (this.filters.action_whitelist.has(this.anyFromCode(act))) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// test for chain::*::name
|
||||
if (this.filters.action_whitelist.has(this.anyFromName(act))) {
|
||||
return true;
|
||||
}
|
||||
// test for chain::*::name
|
||||
if (this.filters.action_whitelist.has(this.anyFromName(act))) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// test for chain::code::name
|
||||
return this.filters.action_whitelist.has(this.codeActionPair(act));
|
||||
}
|
||||
// test for chain::code::name
|
||||
return this.filters.action_whitelist.has(this.codeActionPair(act));
|
||||
}
|
||||
|
||||
protected checkDeltaBlacklist(delta) {
|
||||
protected checkDeltaBlacklist(delta) {
|
||||
|
||||
// test blacklist for chain::code::*
|
||||
if (this.filters.delta_blacklist.has(this.anyFromDeltaCode(delta))) {
|
||||
return true;
|
||||
}
|
||||
// test blacklist for chain::code::*
|
||||
if (this.filters.delta_blacklist.has(this.anyFromDeltaCode(delta))) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// test blacklist for chain::*::table
|
||||
if (this.filters.delta_blacklist.has(this.anyFromDeltaTable(delta))) {
|
||||
return true;
|
||||
}
|
||||
// test blacklist for chain::*::table
|
||||
if (this.filters.delta_blacklist.has(this.anyFromDeltaTable(delta))) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// test blacklist for chain::code::table
|
||||
return this.filters.delta_blacklist.has(this.codeDeltaPair(delta));
|
||||
}
|
||||
// test blacklist for chain::code::table
|
||||
return this.filters.delta_blacklist.has(this.codeDeltaPair(delta));
|
||||
}
|
||||
|
||||
protected checkDeltaWhitelist(delta) {
|
||||
protected checkDeltaWhitelist(delta) {
|
||||
|
||||
// test whitelist for chain::code::*
|
||||
if (this.filters.delta_whitelist.has(this.anyFromDeltaCode(delta))) {
|
||||
return true;
|
||||
}
|
||||
// test whitelist for chain::code::*
|
||||
if (this.filters.delta_whitelist.has(this.anyFromDeltaCode(delta))) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// test whitelist for chain::*::table
|
||||
if (this.filters.delta_whitelist.has(this.anyFromDeltaTable(delta))) {
|
||||
return true;
|
||||
}
|
||||
// test whitelist for chain::*::table
|
||||
if (this.filters.delta_whitelist.has(this.anyFromDeltaTable(delta))) {
|
||||
return true;
|
||||
}
|
||||
|
||||
// test whitelist for chain::code::table
|
||||
return this.filters.delta_whitelist.has(this.codeDeltaPair(delta));
|
||||
}
|
||||
// test whitelist for chain::code::table
|
||||
return this.filters.delta_whitelist.has(this.codeDeltaPair(delta));
|
||||
}
|
||||
|
||||
loadAbiHex(contract, block_num, abi_hex) {
|
||||
// check local blacklist for corrupted abis that failed to load before
|
||||
let _status;
|
||||
if (this.failedAbiMap.has(contract) && this.failedAbiMap.get(contract).has(block_num)) {
|
||||
_status = false;
|
||||
debugLog('ignore saved abi for', contract, block_num);
|
||||
} else {
|
||||
_status = AbiEOS.load_abi_hex(contract, abi_hex);
|
||||
if (!_status) {
|
||||
hLog(`AbiEOS.load_abi_hex error for ${contract} at ${block_num}`);
|
||||
if (this.failedAbiMap.has(contract)) {
|
||||
this.failedAbiMap.get(contract).add(block_num);
|
||||
} else {
|
||||
this.failedAbiMap.set(contract, new Set([block_num]));
|
||||
}
|
||||
} else {
|
||||
this.removeFromFailed(contract);
|
||||
}
|
||||
}
|
||||
return _status;
|
||||
}
|
||||
loadAbiHex(contract, block_num, abi_hex) {
|
||||
// check local blacklist for corrupted abis that failed to load before
|
||||
let _status;
|
||||
if (this.failedAbiMap.has(contract) && this.failedAbiMap.get(contract).has(block_num)) {
|
||||
_status = false;
|
||||
debugLog('ignore saved abi for', contract, block_num);
|
||||
} else {
|
||||
_status = AbiEOS.load_abi_hex(contract, abi_hex);
|
||||
if (!_status) {
|
||||
hLog(`AbiEOS.load_abi_hex error for ${contract} at ${block_num}`);
|
||||
if (this.failedAbiMap.has(contract)) {
|
||||
this.failedAbiMap.get(contract).add(block_num);
|
||||
} else {
|
||||
this.failedAbiMap.set(contract, new Set([block_num]));
|
||||
}
|
||||
} else {
|
||||
this.removeFromFailed(contract);
|
||||
}
|
||||
}
|
||||
return _status;
|
||||
}
|
||||
|
||||
removeFromFailed(contract) {
|
||||
if (this.failedAbiMap.has(contract)) {
|
||||
this.failedAbiMap.delete(contract);
|
||||
hLog(`${contract} was removed from the failed map!`);
|
||||
}
|
||||
}
|
||||
removeFromFailed(contract) {
|
||||
if (this.failedAbiMap.has(contract)) {
|
||||
this.failedAbiMap.delete(contract);
|
||||
hLog(`${contract} was removed from the failed map!`);
|
||||
}
|
||||
}
|
||||
|
||||
async loadCurrentAbiHex(contract) {
|
||||
let _status;
|
||||
if (this.failedAbiMap.has(contract) && this.failedAbiMap.get(contract).has(-1)) {
|
||||
_status = false;
|
||||
debugLog('ignore current abi for', contract);
|
||||
} else {
|
||||
const currentAbi = await this.rpc.getRawAbi(contract);
|
||||
if (currentAbi.abi.byteLength > 0) {
|
||||
const abi_hex = Buffer.from(currentAbi.abi).toString('hex');
|
||||
_status = AbiEOS.load_abi_hex(contract, abi_hex);
|
||||
if (!_status) {
|
||||
hLog(`AbiEOS.load_abi_hex error for ${contract} at head`);
|
||||
if (this.failedAbiMap.has(contract)) {
|
||||
this.failedAbiMap.get(contract).add(-1);
|
||||
} else {
|
||||
this.failedAbiMap.set(contract, new Set([-1]));
|
||||
}
|
||||
} else {
|
||||
this.removeFromFailed(contract);
|
||||
}
|
||||
} else {
|
||||
_status = false;
|
||||
}
|
||||
}
|
||||
return _status;
|
||||
}
|
||||
async loadCurrentAbiHex(contract) {
|
||||
let _status;
|
||||
if (this.failedAbiMap.has(contract) && this.failedAbiMap.get(contract).has(-1)) {
|
||||
_status = false;
|
||||
debugLog('ignore current abi for', contract);
|
||||
} else {
|
||||
const currentAbi = await this.rpc.getRawAbi(contract);
|
||||
if (currentAbi.abi.byteLength > 0) {
|
||||
const abi_hex = Buffer.from(currentAbi.abi).toString('hex');
|
||||
_status = AbiEOS.load_abi_hex(contract, abi_hex);
|
||||
if (!_status) {
|
||||
hLog(`AbiEOS.load_abi_hex error for ${contract} at head`);
|
||||
if (this.failedAbiMap.has(contract)) {
|
||||
this.failedAbiMap.get(contract).add(-1);
|
||||
} else {
|
||||
this.failedAbiMap.set(contract, new Set([-1]));
|
||||
}
|
||||
} else {
|
||||
this.removeFromFailed(contract);
|
||||
}
|
||||
} else {
|
||||
_status = false;
|
||||
}
|
||||
}
|
||||
return _status;
|
||||
}
|
||||
|
||||
abstract run(): Promise<void>
|
||||
abstract run(): Promise<void>
|
||||
|
||||
abstract assertQueues(): void
|
||||
abstract assertQueues(): void
|
||||
|
||||
abstract onIpcMessage(msg: any): void
|
||||
abstract onIpcMessage(msg: any): void
|
||||
|
||||
}
|
||||
|
||||
|
||||
+5
-3
@@ -48,9 +48,11 @@ export default class IndexerWorker extends HyperionWorker {
|
||||
this.indexQueue.pause();
|
||||
this.ch_ready = false;
|
||||
});
|
||||
this.ch.assertQueue(process.env.queue, {durable: true});
|
||||
this.ch.prefetch(this.conf.prefetch.index);
|
||||
this.ch.consume(process.env.queue, this.indexQueue.push);
|
||||
this.ch.assertQueue(process.env.queue, {durable: true}).then(r => {
|
||||
this.ch.consume(process.env.queue, this.indexQueue.push, {
|
||||
prefetch: this.conf.prefetch.index
|
||||
}).catch(console.log);
|
||||
});
|
||||
}
|
||||
} catch (e) {
|
||||
console.error('rabbitmq error!');
|
||||
|
||||
Reference in New Issue
Block a user