Skip to content

Commit

Permalink
feat: Remove async line reader from cspell-io
Browse files Browse the repository at this point in the history
  • Loading branch information
@Jason3S
Jason3S committed Sep 11, 2021
1 parent e688e2b commit 8dd73be
Show file tree
Hide file tree
Showing 9 changed files with 41 additions and 248 deletions.
19 changes: 13 additions & 6 deletions packages/cspell-io/.vscode/launch.json
View file
Expand Up @@ -7,12 +7,19 @@
{
"type": "node",
"request": "launch",
"name": "Launch Program",
"program": "${workspaceFolder}/dist/index.js",
"preLaunchTask": "tsc: build - tsconfig.json",
"outFiles": [
"${workspaceFolder}/dist/**/*.js"
]
"name": "cspell-io: Jest current-file",
"program": "${workspaceFolder}/../../node_modules/.bin/jest",
"cwd": "${workspaceFolder}",
"args": [
"--runInBand",
"${fileBasename}"
],
"console": "integratedTerminal",
"internalConsoleOptions": "neverOpen",
"disableOptimisticBPs": true,
"windows": {
"program": "${workspaceFolder}/../../node_modules/jest/bin/jest",
}
}
]
}
11 changes: 0 additions & 11 deletions packages/cspell-io/src/async/asyncIterable.ts
View file

This file was deleted.

79 changes: 2 additions & 77 deletions packages/cspell-io/src/file/fileReader.test.ts
View file
@@ -1,90 +1,15 @@
import * as fReader from './fileReader';
import * as fs from 'fs-extra';
import * as path from 'path';
import { Readable } from 'stream';
import * as asyncIterable from '../async/asyncIterable';

describe('Validate the fileReader', () => {
const samplePath = path.join(__dirname, '..', '..', 'samples');
const fileCities = path.join(samplePath, 'cities.txt');
const sampleFiles = ['cities.txt', 'cities.CRLF.txt', 'cities.noEOL.txt'].map((f) => path.join(samplePath, f));

test('tests reading a file', async () => {
const expected = await fs.readFile(__filename, 'utf8');
const result = await fReader.readFile(__filename, 'utf8');
expect(result).toBe(expected);
});

test('tests stringsToLines', async () => {
const strings = stringToStream('a1\n2\n3\n4', '5\n6');
const a = await asyncIterable.toArray(fReader.streamLineByLineAsync(strings));
expect(a).toEqual(['a1', '2', '3', '45', '6']);
});

test('tests stringsToLines trailing new line', async () => {
const strings = stringToStream('a1\n2\n3\n4', '5\n6\n');
const a = await asyncIterable.toArray(fReader.streamLineByLineAsync(strings));
expect(a).toEqual(['a1', '2', '3', '45', '6', '']);
});

test('the file reader', async () => {
const lines = await asyncIterable.toArray(fReader.streamFileLineByLineAsync(__filename));
const actual = lines.join('\n');
const expected = fs.readFileSync(__filename, 'utf8');
expect(actual).toBe(expected);
});

test('the lineReaderAsync', async () => {
const lines = await asyncIterable.toArray(fReader.lineReaderAsync(__filename));
const expected = fs.readFileSync(__filename, 'utf8').split('\n');
expect(lines).toEqual(expected);
});

test('tests reading the cities sample', async () => {
const lines = await asyncIterable.toArray(fReader.lineReaderAsync(fileCities));
const file = await fs.readFile(fileCities, 'utf8');
expect(lines).toEqual(file.split('\n'));
});

test('tests streamFileLineByLineAsync', async () => {
await Promise.all(
sampleFiles.map(async (filename) => {
const lines = await asyncIterable.toArray(fReader.streamFileLineByLineAsync(filename));
const file = await fs.readFile(filename, 'utf8');
// compare to file: ${filename}
expect(lines).toEqual(file.split(/\r?\n/));
})
);
});

test('tests streamFileLineByLineAsync 2', async () => {
const lines = await asyncIterable.toArray(fReader.streamFileLineByLineAsync(__filename));
const file = await fs.readFile(__filename, 'utf8');
expect(lines).toEqual(file.split('\n'));
});

test('missing file', async () => {
const result = asyncIterable.toArray(fReader.lineReaderAsync(__filename + 'not.found'));
return result.then(
() => {
expect('not to be here').toBe(true);
return;
},
(e) => {
// expect(e).to.be.instanceof(Error); // Since jest currently mocks Error, this test fails.
expect(e.code).toBe('ENOENT');
}
);
const result = fReader.readFile(__filename + '.missing.txt', 'utf8');
await expect(result).rejects.toEqual(expect.objectContaining({ code: 'ENOENT' }));
});
});

function stringToStream(...strings: string[]): NodeJS.ReadableStream {
return new Readable({
read: function () {
for (const s of strings) {
this.push(s);
}
this.push(null);
},
});
}
161 changes: 16 additions & 145 deletions packages/cspell-io/src/file/fileReader.ts
View file
@@ -1,156 +1,27 @@
// cSpell:ignore curr
// cSpell:words zlib iconv
import * as fs from 'fs';
import * as zlib from 'zlib';
import * as readline from 'readline';
import { PassThrough, pipeline as pipelineCB } from 'stream';
import { promisify } from 'util';

const defaultEncoding: BufferEncoding = 'utf8';

export function readFile(filename: string, encoding: BufferEncoding = defaultEncoding): Promise<string> {
return new Promise((resolve, reject) => {
const data: string[] = [];
const stream = prepareFileStream(filename, encoding, reject);
let resolved = false;
function complete() {
resolve(data.join(''));
resolved = resolved || (resolve(data.join('')), true);
}
stream.on('error', reject);
stream.on('data', (d: string) => data.push(d));
stream.on('close', complete);
stream.on('end', complete);
});
}
const pipeline = promisify(pipelineCB);

/**
* Reads a file line by line. The last value emitted by the Observable is always an empty string.
* @param filename
* @param encoding defaults to 'utf8'
*/
export function lineReaderAsync(filename: string, encoding: BufferEncoding = defaultEncoding): AsyncIterable<string> {
return streamFileLineByLineAsync(filename, encoding);
}
const defaultEncoding: BufferEncoding = 'utf8';

function prepareFileStream(filename: string, encoding: BufferEncoding, fnError: (e: Error) => void) {
const pipes: NodeJS.ReadWriteStream[] = [];
if (filename.match(/\.gz$/i)) {
pipes.push(zlib.createGunzip());
}
export async function readFile(filename: string, encoding: BufferEncoding = defaultEncoding): Promise<string> {
const isGzip = filename.match(/\.gz$/i);
const fileStream = fs.createReadStream(filename);
fileStream.on('error', fnError);
const stream = pipes.reduce<NodeJS.ReadableStream>((s, p) => s.pipe(p).on('error', fnError), fileStream);
stream.setEncoding(encoding);
return stream;
}

/**
* Emit a file line by line
* @param filename full path to the file to read.
* @param encoding defaults to 'utf8'
*/
export function streamFileLineByLineAsync(
filename: string,
encoding: BufferEncoding = defaultEncoding
): AsyncIterableIterator<string> {
const fnError = (e: Error) => {
iter.throw && iter.throw(e);
};
const stream = prepareFileStream(filename, encoding, fnError);
const iter = streamLineByLineAsync(stream);
return iter;
}

type Resolve<T> = (value: T | Promise<T>) => void;
// eslint-disable-next-line @typescript-eslint/no-explicit-any
type Reject = (reason?: any) => void;

interface Resolvers<T = IteratorResult<string>> {
resolve: Resolve<T>;
reject: Reject;
const zip = isGzip ? zlib.createGunzip() : new PassThrough();
const t = pipeline(fileStream, zip, streamToText(encoding));
return await t;
}

/**
* Emit a file line by line
* @param filename full path to the file to read.
* @param encoding defaults to 'utf8'
*/
export function streamLineByLineAsync(
stream: NodeJS.ReadableStream,
encoding: BufferEncoding = defaultEncoding
): AsyncIterableIterator<string> {
let data = '.';
let done = false;
let error: Error | undefined;
const buffer: string[] = [];
const pending: Resolvers[] = [];
const fnError = (e: Error | undefined) => {
error = e;
};
const fnComplete = () => {
// readline will consume the last newline without emitting an empty last line.
// If the last data read contains a new line, then emit an empty string.
if (data.match(/(?:(?:\r?\n)|(?:\r))$/)) {
buffer.push('');
}
processBuffer();
done = true;
};
// We want to capture the last line.
stream.on('data', (d) => (data = dataToString(d, encoding)));
stream.on('error', fnError);
const rl = readline.createInterface({
input: stream,
terminal: false,
});
rl.on('close', fnComplete);
rl.on('line', (text: string) => {
buffer.push(text);
processBuffer();
});

function registerPromise(resolve: Resolve<IteratorResult<string>>, reject: Reject) {
pending.push({ resolve, reject });
processBuffer();
}

function processBuffer() {
if (error && pending.length && !buffer.length) {
const p = pending.shift();
p?.reject(error);
return;
}
while (pending.length && buffer.length) {
const p = pending.shift();
const b = buffer.shift();
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
p?.resolve({ done: false, value: b! });
function streamToText(encoding: BufferEncoding): (source: fs.ReadStream) => Promise<string> {
return async function (source: fs.ReadStream): Promise<string> {
const chunks: string[] = [];
source.setEncoding(encoding); // Work with strings rather than `Buffer`s.
for await (const chunk of source) {
chunks.push(chunk);
}
if (!done) {
pending.length ? rl.resume() : rl.pause();
}
if (done && pending.length && !buffer.length) {
const p = pending.shift();
p?.resolve({ done, value: undefined });
}
}

const iter: AsyncIterableIterator<string> = {
[Symbol.asyncIterator]: () => iter,
next() {
return new Promise(registerPromise);
},
throw(e) {
fnError(e);
return new Promise(registerPromise);
},
return chunks.join('');
};

return iter;
}

function dataToString(data: string | Buffer, encoding: BufferEncoding = 'utf8'): string {
if (typeof data === 'string') {
return data;
}
return data.toString(encoding);
}
12 changes: 7 additions & 5 deletions packages/cspell-io/src/file/fileWriter.ts
View file
@@ -1,6 +1,9 @@
import * as fs from 'fs';
import * as zlib from 'zlib';
import * as stream from 'stream';
import { promisify } from 'util';

const pipeline = promisify(stream.pipeline);

export function writeToFile(filename: string, data: string): NodeJS.WritableStream {
return writeToFileIterable(filename, [data]);
Expand All @@ -14,9 +17,8 @@ export function writeToFileIterable(filename: string, data: Iterable<string>): N
}

export function writeToFileIterableP(filename: string, data: Iterable<string>): Promise<void> {
const stream = writeToFileIterable(filename, data);
return new Promise<void>((resolve, reject) => {
stream.on('finish', () => resolve());
stream.on('error', (e: Error) => reject(e));
});
const sourceStream = stream.Readable.from(data);
const writeStream = fs.createWriteStream(filename);
const zip = filename.match(/\.gz$/) ? zlib.createGzip() : new stream.PassThrough();
return pipeline(sourceStream, zip, writeStream);
}
1 change: 0 additions & 1 deletion packages/cspell-io/src/index.ts
View file
@@ -1,2 +1 @@
export * from './file';
export { toArray as asyncIterableToArray } from './async/asyncIterable';
2 changes: 1 addition & 1 deletion test-packages/test-cspell-io/src/index.ts
View file
Expand Up @@ -6,7 +6,7 @@ console.log('start');
/**
* The main goal here is to make sure it compiles. The unit tests are validation that it compiled as expected.
*/
const functions = [io.asyncIterableToArray, io.lineReaderAsync, io.readFile];
const functions = [io.readFile];

functions.forEach((fn) => assert(typeof fn === 'function', "typeof %o === 'function'", fn));

Expand Down
2 changes: 1 addition & 1 deletion test-packages/test-cspell-lib-webpack/src/index.ts
View file
Expand Up @@ -7,7 +7,7 @@ console.log('start');
/**
* The main goal here is to make sure it compiles. The unit tests are validation that it compiled as expected.
*/
const functions = [lib.asyncIterableToArray, lib.calcOverrideSettings];
const functions = [lib.calcOverrideSettings];
functions.forEach((fn) => assert(typeof fn === 'function', "typeof %o === 'function'", fn));

console.log('done');
2 changes: 1 addition & 1 deletion test-packages/test-cspell-lib/src/index.ts
View file
Expand Up @@ -6,7 +6,7 @@ console.log('start');
/**
* The main goal here is to make sure it compiles. The unit tests are validation that it compiled as expected.
*/
const functions = [lib.asyncIterableToArray, lib.calcOverrideSettings];
const functions = [lib.calcOverrideSettings];
functions.forEach((fn) => assert(typeof fn === 'function', "typeof %o === 'function'", fn));

console.log('done');

0 comments on commit 8dd73be

Please sign in to comment.