-
Notifications
You must be signed in to change notification settings - Fork 576
/
stream.ts
112 lines (96 loc) · 2.81 KB
/
stream.ts
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
import { defaultReadOptions, parseRecord } from "./_io.ts";
import type { LineReader } from "./_io.ts";
import { TextDelimiterStream } from "../../streams/delimiter.ts";
export interface CsvStreamOptions {
separator?: string;
comment?: string;
}
/** @deprecated Use CsvStreamOptions instead. */
export type CSVStreamOptions = CsvStreamOptions;
class StreamLineReader implements LineReader {
#reader: ReadableStreamDefaultReader<string>;
#done = false;
constructor(reader: ReadableStreamDefaultReader<string>) {
this.#reader = reader;
}
async readLine(): Promise<string | null> {
const { value, done } = await this.#reader.read();
if (done) {
this.#done = true;
return null;
} else {
// NOTE: Remove trailing CR for compatibility with golang's `encoding/csv`
return stripLastCR(value!);
}
}
isEOF(): Promise<boolean> {
return Promise.resolve(this.#done);
}
cancel() {
this.#reader.cancel();
}
}
function stripLastCR(s: string): string {
return s.endsWith("\r") ? s.slice(0, -1) : s;
}
export class CsvStream implements TransformStream<string, Array<string>> {
readonly #readable: ReadableStream<Array<string>>;
readonly #options: CsvStreamOptions;
readonly #lineReader: StreamLineReader;
readonly #lines: TextDelimiterStream;
#lineIndex = 0;
constructor(options: CSVStreamOptions = defaultReadOptions) {
this.#options = {
...defaultReadOptions,
...options,
};
this.#lines = new TextDelimiterStream("\n");
this.#lineReader = new StreamLineReader(this.#lines.readable.getReader());
this.#readable = new ReadableStream<Array<string>>({
pull: (controller) => this.#pull(controller),
cancel: () => this.#lineReader.cancel(),
});
}
async #pull(
controller: ReadableStreamDefaultController<Array<string>>,
): Promise<void> {
const line = await this.#lineReader.readLine();
if (line === "") {
// Found an empty line
this.#lineIndex++;
return this.#pull(controller);
}
if (line === null) {
// Reached to EOF
controller.close();
this.#lineReader.cancel();
return;
}
const record = await parseRecord(
line,
this.#lineReader,
this.#options,
this.#lineIndex,
);
if (record === null) {
controller.close();
this.#lineReader.cancel();
return;
}
this.#lineIndex++;
if (record.length > 0) {
controller.enqueue(record);
} else {
return this.#pull(controller);
}
}
get readable(): ReadableStream<Array<string>> {
return this.#readable;
}
get writable(): WritableStream<string> {
return this.#lines.writable;
}
}
/** @deprecated Use CsvStream instead. */
export const CSVStream = CsvStream;