Merge pull request #7 from kryptokrauts/bugfix/control-api

Bugfix/control api

Author: mitch-lbw

.gitignore

@@ -3,4 +3,5 @@ node_modules/
 dist/
 .lh/
 lib/
-*.tgz
+*.tgz
+.env_local*

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@kryptokrauts/event-processor-node-lib",
-  "version": "1.0.6",
+  "version": "1.0.7",
   "description": "Wrapper for ease listening on antelope blockchain based on @blockmatic/antelope-ship-reader",
   "main": "dist/index.js",
   "types": "dist/index.d.ts",

src/control-api.ts

@@ -4,7 +4,7 @@ import { control_api_port, getLogger } from './common/config';
 const api = express();
 const logger = getLogger('reset-api');
 
-export async function startControlApi(reset_handler) {
+export async function startControlApi(reset_handler, instance) {
   api.get('/reset', (request, response) => {
     const restart_at_block = request.query.blocknum;
     const reset_database = request.query.reset_db || false;
@@ -15,7 +15,7 @@ export async function startControlApi(reset_handler) {
       logger.info(
         `Parameters provided: blocknum: ${restart_at_block}, reset_db: ${reset_database}`,
       );
-      reset_handler(restart_at_block, reset_database);
+      reset_handler.call(instance, restart_at_block, reset_database);
       response
         .status(200)
         .send(

src/eosio/ship-reader-wrapper.ts

@@ -65,7 +65,7 @@ export class ShipReaderWrapper {
     logger.info(KAFKA_TOPIC_CONFIG);
 
     // start control api for sending reset message
-    startControlApi(this.handleExternalReset);
+    startControlApi(this.handleExternalReset, this);
 
     // connect to kafka and retrieve last processed message data
     this.kafka_wrapper = new KafkaWrapper({ header_prefix: this.config.message_header_prefix });
@@ -380,19 +380,25 @@ export class ShipReaderWrapper {
     }
   }
 
+  /**
+   * send external reset message
+   * @param restart_at_block
+   * @param reset_db
+   */
   private async handleExternalReset(restart_at_block: number, reset_db: boolean) {
     try {
       logger.warn(`Got external reset event to restart at blockum ${restart_at_block}`);
       const resetEvent: string = this.createResetEvent(
-        'external_restart',
+        'external_blocknum_reset',
         `caused by control api call`,
         restart_at_block,
         reset_db,
       );
       resetEvent && (await this.kafka_wrapper.sendEvent(resetEvent, 'reset_event'));
       await this.gracefulShutdown();
-    } finally {
-      process.kill(process.pid, -1);
+    } catch (e) {
+      logger.error(e);
+      this.sendEventAndEndProcess('handle external reset', e);
     }
   }
 }