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
// 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;}
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;  }}