flow control

package.json

@@ -1,6 +1,6 @@
 {
   "name": "hyperion-history",
-  "version": "3.3.9-7",
+  "version": "3.3.9-8",
   "description": "Scalable Full History API Solution for EOSIO based blockchains",
   "main": "launcher.js",
   "scripts": {

workers/deserializer.ts

@@ -8,6 +8,7 @@ import {Type} from "../addons/eosjs-native/eosjs-serialize";
 import {debugLog, hLog} from "../helpers/common_functions";
 import {createHash} from "crypto";
 import flatstr from 'flatstr';
+import {Options} from "amqplib";
 
 const FJS = require('fast-json-stringify');
 
@@ -124,6 +125,9 @@ export default class MainDSWorker extends HyperionWorker {
 
     allowedDynamicContracts: Set<string> = new Set<string>();
 
+    backpressureQueue: any[] = [];
+    waitToSend = false;
+
     constructor() {
 
         super();
@@ -290,6 +294,16 @@ export default class MainDSWorker extends HyperionWorker {
             this.ch.consume(process.env['worker_queue'], (data) => {
                 this.consumerQueue.push(data).catch(console.log);
             });
+            this.ch.on('drain', args => {
+                this.waitToSend = false;
+                while (this.backpressureQueue.length > 0) {
+                    const msg = this.backpressureQueue.shift();
+                    const status = this.controlledSendToQueue(msg.queue, msg.payload, msg.options);
+                    if (!status) {
+                        break;
+                    }
+                }
+            });
         }
     }
 
@@ -622,19 +636,34 @@ export default class MainDSWorker extends HyperionWorker {
         }
 
         const pool_queue = `${this.chain}:ds_pool:${selected_q}`;
-        if (this.ch_ready) {
-            const enqueueResult = this.ch.sendToQueue(pool_queue, bufferFromJson(trace, true), {headers});
-            if (!enqueueResult) {
-                hLog("Backpressure");
-                console.log("Header size: " + JSON.stringify(headers).length);
-                console.log(headers);
+        const payload = bufferFromJson(trace, true);
+
+        if (!this.waitToSend) {
+            if (this.ch_ready) {
+                this.controlledSendToQueue(pool_queue, payload, {headers});
+                return true;
+            } else {
+                return false;
             }
-            return true;
         } else {
+            this.backpressureQueue.push({
+                queue: pool_queue,
+                payload: payload,
+                options: {headers}
+            });
             return false;
         }
     }
 
+    controlledSendToQueue(pool_queue: string, payload: Buffer, options: Options.Publish): boolean {
+        const enqueueResult = this.ch.sendToQueue(pool_queue, payload, options);
+        if (!enqueueResult) {
+            this.waitToSend = true;
+            hLog("Backpressure");
+        }
+        return enqueueResult;
+    }
+
     createSerialBuffer(inputArray) {
         return new Serialize.SerialBuffer({textEncoder: this.txEnc, textDecoder: this.txDec, array: inputArray});
     }