refactor: define our Writer and Stringifier interface

package.json

@@ -80,14 +80,12 @@
   },
   "dependencies": {
     "@kintone/rest-api-client": "^3.3.10",
-    "@types/pumpify": "^1.4.1",
     "chalk": "4",
     "csv-parse": "^4.16.3",
     "csv-stringify": "5.6.5",
     "https-proxy-agent": "^5.0.1",
     "iconv-lite": "^0.6.3",
     "inquirer": "^8.2.5",
-    "pumpify": "^2.0.1",
     "yargs": "^17.7.1"
   }
 }

src/record/export/repositories/localRecordRepositoryMock.ts

@@ -1,6 +1,5 @@
-import type { LocalRecordRepository } from "../usecases/interface";
+import type { LocalRecordRepository, Writer } from "../usecases/interface";
 import type { LocalRecord } from "../types/record";
-import { Writable } from "stream";
 
 export class LocalRecordRepositoryMock implements LocalRecordRepository {
   readonly format = "csv";
@@ -16,12 +15,12 @@ export class LocalRecordRepositoryMock implements LocalRecordRepository {
   }
 }
 
-class WritableMock extends Writable {
+class WritableMock implements Writer {
   public underlyingSink: LocalRecord[] = [];
-  constructor() {
-    super({ objectMode: true });
-  }
-  _write(chunk: LocalRecord[]) {
+  async write(chunk: LocalRecord[]) {
     this.underlyingSink.push(...chunk);
   }
+  async end() {
+    /* noop */
+  }
 }

src/record/export/repositories/stringifiers/__tests__/index.test.ts

@@ -10,8 +10,8 @@ describe("stringifier", () => {
       schema,
       useLocalFilePath: false,
     });
-    stringifier.write(input);
-    stringifier.end();
+    await stringifier.write(input);
+    await stringifier.end();
 
     let actual = "";
     for await (const chunk of stringifier) {
@@ -28,9 +28,9 @@ describe("stringifier", () => {
       useLocalFilePath: false,
     });
     for (const localRecord of input) {
-      stringifier.write([localRecord]);
+      await stringifier.write([localRecord]);
     }
-    stringifier.end();
+    await stringifier.end();
     let actual = "";
     for await (const chunk of stringifier) {
       actual += chunk;

src/record/export/repositories/stringifiers/csvStringifier/__tests__/index.test.ts

@@ -36,7 +36,10 @@ describe("csvStringifier", () => {
       pattern.useLocalFilePath
     );
     const source = Readable.from(pattern.input.map((record) => [record]));
-    source.pipe(csvStringifier);
+    for await (const chunk of source) {
+      await csvStringifier.write(chunk);
+    }
+    await csvStringifier.end();
     let actual = "";
     for await (const chunk of csvStringifier) {
       actual += chunk;

src/record/export/repositories/stringifiers/csvStringifier/index.ts

@@ -3,7 +3,6 @@ import type { CsvRow } from "../../../../../kintone/types";
 import type { RecordSchema } from "../../../types/schema";
 
 import stringify from "csv-stringify";
-import Pumpify from "pumpify";
 
 import { convertRecord, recordReader } from "./record";
 import { LINE_BREAK, SEPARATOR } from "./constants";
@@ -12,9 +11,10 @@ import type { TransformCallback } from "stream";
 import { PassThrough, Transform } from "stream";
 import type { Stringifier } from "../index";
 
-export class CsvStringifier extends Pumpify.obj implements Stringifier {
+export class CsvStringifier implements Stringifier {
   private readonly recordTransformer: Transform;
   private readonly csvStringifier: stringify.Stringifier;
+  private readonly encoder: PassThrough;
 
   constructor(schema: RecordSchema, useLocalFilePath: boolean) {
     const recordTransform = new RecordTransform(schema, useLocalFilePath);
@@ -29,11 +29,35 @@ export class CsvStringifier extends Pumpify.obj implements Stringifier {
       quoted_empty: false,
     });
 
-    const encoder = new PassThrough({ encoding: "utf-8" });
-
-    super(recordTransform, csvStringifier, encoder);
     this.recordTransformer = recordTransform;
     this.csvStringifier = csvStringifier;
+    this.encoder = new PassThrough({ encoding: "utf-8" });
+    this.recordTransformer.pipe(this.csvStringifier).pipe(this.encoder);
+  }
+
+  async write(records: LocalRecord[]) {
+    return new Promise<void>((resolve, reject) => {
+      this.recordTransformer.write(records, (e) => {
+        if (e) {
+          reject(e);
+        }
+        resolve();
+      });
+    });
+  }
+  read(size?: number) {
+    return this.encoder.read(size);
+  }
+  async end() {
+    return new Promise<void>((resolve) => {
+      this.recordTransformer.end(resolve);
+    });
+  }
+  pipe<T extends NodeJS.WritableStream>(destination: T) {
+    this.encoder.pipe(destination);
+  }
+  async *[Symbol.asyncIterator](): AsyncGenerator<string, void, undefined> {
+    yield* this.encoder;
   }
 }
 

src/record/export/repositories/stringifiers/index.ts

@@ -1,9 +1,15 @@
 import type { RecordSchema } from "../../types/schema";
 
 import { CsvStringifier } from "./csvStringifier";
-import type { Duplex } from "stream";
+import type { LocalRecord } from "../../types/record";
 
-export type Stringifier = Duplex;
+export type Stringifier = {
+  write(records: LocalRecord[]): Promise<void>;
+  read(size?: number): string;
+  end(): Promise<void>;
+  pipe<T extends NodeJS.WritableStream>(destination: T): void;
+  [Symbol.asyncIterator](): AsyncIterableIterator<string>;
+};
 
 type FactoryOptions = {
   format: "csv";

src/record/export/usecases/get.ts

@@ -43,9 +43,9 @@ export const getRecords = async (
           attachmentsDir,
         })
     );
-    writer.write(localRecords);
+    await writer.write(localRecords);
   }
-  writer.end();
+  await writer.end();
 };
 
 const recordsReducer: (

src/record/export/usecases/interface.ts

@@ -1,8 +1,11 @@
-import type { Writable } from "stream";
+import type { LocalRecord } from "../types/record";
 
 export type LocalRecordRepository = {
   readonly format: string;
   readonly writer: () => Writer;
 };
 
-type Writer = Writable;
+export type Writer = {
+  write(records: LocalRecord[]): Promise<void>;
+  end(): Promise<void>;
+};

yarn.lock

@@ -1986,13 +1986,6 @@
   dependencies:
     "@babel/types" "^7.3.0"
 
-"@types/duplexify@*":
-  version "3.6.1"
-  resolved "https://registry.yarnpkg.com/@types/duplexify/-/duplexify-3.6.1.tgz#5685721cf7dc4a21b6f0e8a8efbec6b4d2fbafad"
-  integrity sha512-n0zoEj/fMdMOvqbHxmqnza/kXyoGgJmEpsXjpP+gEqE1Ye4yNqc7xWipKnUoMpWhMuzJQSfK2gMrwlElly7OGQ==
-  dependencies:
-    "@types/node" "*"
-
 "@types/fs-extra@^11.0.1":
   version "11.0.1"
   resolved "https://registry.yarnpkg.com/@types/fs-extra/-/fs-extra-11.0.1.tgz#f542ec47810532a8a252127e6e105f487e0a6ea5"
@@ -2090,14 +2083,6 @@
   resolved "https://registry.yarnpkg.com/@types/ps-tree/-/ps-tree-1.1.2.tgz#5c60773a38ffb1402e049902a7b7a8d3c67cd59a"
   integrity sha512-ZREFYlpUmPQJ0esjxoG1fMvB2HNaD3z+mjqdSosZvd3RalncI9NEur73P8ZJz4YQdL64CmV1w0RuqoRUlhQRBw==
 
-"@types/pumpify@^1.4.1":
-  version "1.4.1"
-  resolved "https://registry.yarnpkg.com/@types/pumpify/-/pumpify-1.4.1.tgz#5a0650f39a3f8f077c7e544d0c5ae2899b28394c"
-  integrity sha512-l7u/Dnh1OG9T7VH6TvulR0g8oE8hgIW5409mSUKi8Vxw2+JV18aTa06Sv5bvNjrD0zbsB/cuZ/iTFQgFNfzIuw==
-  dependencies:
-    "@types/duplexify" "*"
-    "@types/node" "*"
-
 "@types/rollup-plugin-auto-external@^2.0.2":
   version "2.0.2"
   resolved "https://registry.yarnpkg.com/@types/rollup-plugin-auto-external/-/rollup-plugin-auto-external-2.0.2.tgz#d1c9697c2445db956df5215d68c35b5fe794b99c"
@@ -2995,16 +2980,6 @@ duplexer@~0.1.1:
   resolved "https://registry.yarnpkg.com/duplexer/-/duplexer-0.1.2.tgz#3abe43aef3835f8ae077d136ddce0f276b0400e6"
   integrity sha512-jtD6YG370ZCIi/9GTaJKQxWTZD045+4R4hTk/x1UyoqadyJ9x9CgSi1RlVDQF8U2sxLLSnFkCaMihqljHIWgMg==
 
-duplexify@^4.1.1:
-  version "4.1.2"
-  resolved "https://registry.yarnpkg.com/duplexify/-/duplexify-4.1.2.tgz#18b4f8d28289132fa0b9573c898d9f903f81c7b0"
-  integrity sha512-fz3OjcNCHmRP12MJoZMPglx8m4rrFP8rovnk4vT8Fs+aonZoCwGg10dSsQsfP/E62eZcPTMSMP6686fu9Qlqtw==
-  dependencies:
-    end-of-stream "^1.4.1"
-    inherits "^2.0.3"
-    readable-stream "^3.1.1"
-    stream-shift "^1.0.0"
-
 electron-to-chromium@^1.4.172:
   version "1.4.177"
   resolved "https://registry.yarnpkg.com/electron-to-chromium/-/electron-to-chromium-1.4.177.tgz#b6a4436eb788ca732556cd69f384b8a3c82118c5"
@@ -5612,15 +5587,6 @@ pump@^3.0.0:
     end-of-stream "^1.1.0"
     once "^1.3.1"
 
-pumpify@^2.0.1:
-  version "2.0.1"
-  resolved "https://registry.yarnpkg.com/pumpify/-/pumpify-2.0.1.tgz#abfc7b5a621307c728b551decbbefb51f0e4aa1e"
-  integrity sha512-m7KOje7jZxrmutanlkS1daj1dS6z6BgslzOXmcSEpIlCxM3VJH7lG5QLeck/6hgF6F4crFf01UtQmNsJfweTAw==
-  dependencies:
-    duplexify "^4.1.1"
-    inherits "^2.0.3"
-    pump "^3.0.0"
-
 punycode@^2.1.0:
   version "2.1.1"
   resolved "https://registry.yarnpkg.com/punycode/-/punycode-2.1.1.tgz#b58b010ac40c22c5657616c8d2c2c02c7bf479ec"
@@ -6133,11 +6099,6 @@ stream-meter@^1.0.4:
   dependencies:
     readable-stream "^2.1.4"
 
-stream-shift@^1.0.0:
-  version "1.0.1"
-  resolved "https://registry.yarnpkg.com/stream-shift/-/stream-shift-1.0.1.tgz#d7088281559ab2778424279b0877da3c392d5a3d"
-  integrity sha512-AiisoFqQ0vbGcZgQPY1cdP2I76glaVA/RauYR4G4thNFgkTqr90yXTo4LYX60Jl+sIlPNHHdGSwo01AvbKUSVQ==
-
 string-length@^4.0.1:
   version "4.0.2"
   resolved "https://registry.yarnpkg.com/string-length/-/string-length-4.0.2.tgz#a8a8dc7bd5c1a82b9b3c8b87e125f66871b6e57a"