update onRestartInactivitySeconds

example.js

@@ -1,16 +1,17 @@
 import pkg from "./package.json" assert { type: "json" };
-import { commander, setup, prometheus, http } from "./dist/index.js";
+import { commander, setup, prometheus, http, logger } from "./dist/index.js";
 
 // Setup CLI using Commander
 const program = commander.program(pkg);
 const command = commander.run(program, pkg);
+logger.setName(pkg.name);
 
 // Custom Prometheus Counters
 const customCounter = prometheus.registerCounter("custom_counter");
 
 command.action(async options => {
   // Setup sink for Block Emitter
-  const { emitter } = await setup(options, pkg);
+  const {emitter} = await setup(options);
   console.log("setup")
 
   // Stream Blocks
@@ -26,6 +27,7 @@ command.action(async options => {
 
   // Start streaming
   await emitter.start();
-  console.log("finished");
+  http.server.close()
+  console.log("✅ finished");
 })
 program.parse();

package.json

@@ -1,6 +1,6 @@
 {
     "name": "substreams-sink",
-    "version": "0.8.2",
+    "version": "0.9.0",
     "description": "Substreams Sink",
     "type": "module",
     "exports": "./dist/index.js",

src/restartInactivitySeconds.ts

@@ -6,27 +6,21 @@ const CHECK_INACTIVITY_INTERVAL = 1000;
 
 export function onRestartInactivitySeconds(emitter: BlockEmitter, restartInactivitySeconds: number) {
     let lastUpdate = now();
-    let isStarted = false;
     let isFinished = false;
 
     async function checkInactivity() {
         if (now() - lastUpdate > restartInactivitySeconds) {
-            if (!isStarted) return;
             logger.error(`Restarting due to inactivity for ${restartInactivitySeconds} seconds`);
-            process.exit(1);
+            process.exit(1); // force quit
         }
-        if (isFinished) return;
+        if (isFinished) return; // exit out of the loop
         await setTimeout(CHECK_INACTIVITY_INTERVAL);
         checkInactivity();
     }
-
-    emitter.on("cursor", () => {
-        isStarted = true;
+    emitter.on("cursor", (cursor, clock) => {
         lastUpdate = now();
-    });
-
-    emitter.on("block", (block) => {
-        if (block.clock?.number === emitter.request.stopBlockNum - 1n) {
+        console.log(clock.number, emitter.request.stopBlockNum)
+        if (clock.number >= emitter.request.stopBlockNum - 1n) {
             isFinished = true;
         };
     });

src/setup.ts

@@ -11,10 +11,9 @@ import { logger } from "./logger.js";
 import { onRestartInactivitySeconds } from "./restartInactivitySeconds.js";
 import { applyParams } from "./applyParams.js";
 
-export async function setup(options: RunOptions, pkg: { name: string }) {
+export async function setup(options: RunOptions) {
     // Configure logging with TSLog
     if (options.verbose) logger.enable();
-    logger.setName(pkg.name);
 
     // Download Substream package
     const manifest = options.manifest;
@@ -75,7 +74,13 @@ export async function setup(options: RunOptions, pkg: { name: string }) {
     await setTimeout(options.delayBeforeStart);
 
     // Restart on inactivity
-    onRestartInactivitySeconds(emitter, options.restartInactivitySeconds);
+    // only activate once first cursor is received
+    let isStarted = false;
+    emitter.on("cursor", () => {
+        if ( isStarted ) return;
+        onRestartInactivitySeconds(emitter, options.restartInactivitySeconds);
+        isStarted = true;
+    });
 
     return { emitter, substreamPackage, moduleHash, startCursor };
 }