Adding stream-json tools

package-lock.json

@@ -8500,6 +8500,19 @@
       "integrity": "sha1-NbCYdbT/SfJqd35QmzCQoyJr8ks=",
       "dev": true
     },
+    "stream-chain": {
+      "version": "2.2.4",
+      "resolved": "https://registry.npmjs.org/stream-chain/-/stream-chain-2.2.4.tgz",
+      "integrity": "sha512-9lsl3YM53V5N/I1C2uJtc3Kavyi3kNYN83VkKb/bMWRk7D9imiFyUPYa0PoZbLohSVOX1mYE9YsmwObZUsth6Q=="
+    },
+    "stream-json": {
+      "version": "1.7.1",
+      "resolved": "https://registry.npmjs.org/stream-json/-/stream-json-1.7.1.tgz",
+      "integrity": "sha512-I7g0IDqvdJXbJ279/D3ZoTx0VMhmKnEF7u38CffeWdF8bfpMPsLo+5fWnkNjO2GU/JjWaRjdH+zmH03q+XGXFw==",
+      "requires": {
+        "stream-chain": "^2.2.3"
+      }
+    },
     "streamsearch": {
       "version": "0.1.2",
       "resolved": "https://registry.npmjs.org/streamsearch/-/streamsearch-0.1.2.tgz",

package.json

@@ -57,6 +57,8 @@
     "pino-pretty": "^4.7.1",
     "sharp": "^0.28.1",
     "socket.io": "^4.1.1",
+    "stream-chain": "^2.2.4",
+    "stream-json": "^1.7.1",
     "through2": "^4.0.2",
     "typescript": "^3.8.3",
     "unzipper": "^0.10.11"

services/api.service.ts

@@ -97,13 +97,7 @@ export default class ApiService extends Service {
 						this.broker
 							.call("import." + action, params, opts)
 							.then((res) => {
-								client.emit("hello", "world");
-								console.log(client);
-
-								console.log("DONE");
-
 								client.emit("comicBookCoverMetadata", res);
-								done(res);
 							})
 							.catch((err) => this.logger.error(err));
 					});

services/import.service.ts

@@ -3,7 +3,7 @@ import { Context, Service, ServiceBroker, ServiceSchema } from "moleculer";
 import fs from "fs";
 import { DbMixin } from "../mixins/db.mixin";
 import Comic from "../models/comic.model";
-import { map, flatten } from "lodash";
+import { map, flatten, isUndefined } from "lodash";
 import {
 	extractArchive,
 	getCovers,
@@ -21,6 +21,11 @@ import H from "highland";
 import { stringify } from "highland-json";
 const IO = require("socket.io")();
 
+const { parser } = require("stream-json");
+const { pick } = require("stream-json/filters/Pick");
+const { ignore } = require("stream-json/filters/Ignore");
+const { streamValues } = require("stream-json/streamers/StreamValues");
+
 export default class ProductsService extends Service {
 	// @ts-ignore
 	public constructor(
@@ -74,18 +79,24 @@ export default class ProductsService extends Service {
 									ctx.params.extractionOptions.extractionMode
 								) {
 									case "bulk":
+										let rs = new Readable();
 										const extractedDataPromises = map(
 											ctx.params.walkedFolders,
-											async (folder) =>
-												await extractArchive(
-													ctx.params
-														.extractionOptions,
-													folder
-												)
+											async (folder) => {
+												while (!isUndefined(folder)) {
+													let foo =
+														await extractArchive(
+															ctx.params
+																.extractionOptions,
+															folder
+														);
+													console.log("levar", foo);
+													rs.push(foo);
+												}
+												rs.push(null);
+											}
 										);
-										return Promise.all(
-											extractedDataPromises
-										).then((data) => flatten(data));
+
 									case "single":
 										return await extractArchive(
 											ctx.params.extractionOptions,