1
// Copyright (c) 2025 Cloudflare, Inc.
2
// Licensed under the Apache 2.0 license found in the LICENSE file or at:
3
//     https://opensource.org/licenses/Apache-2.0
4
import {
5
  ReadStream,
6
  createReadStream,
7
  statSync,
8
  writeFileSync,
9
  openSync,
10
  open as openAsync,
11
  close as closeAsync,
12
  read as readAsync,
13
  promises,
14
} from 'node:fs';
15
 
16
import { ok, strictEqual, throws } from 'node:assert';
17
import { mock } from 'node:test';
18
 
19
strictEqual(typeof ReadStream, 'function');
20
strictEqual(typeof createReadStream, 'function');
21
 
22
export const simpleReadStreamTest = {
23
  async test() {
24
    const largeData = 'abc'.repeat(100_000);
25
    writeFileSync('/tmp/foo', largeData);
26
 
27
    const stream = createReadStream('/tmp/foo', {
28
      encoding: 'utf8',
29
      highWaterMark: 10_000,
30
    });
31
 
32
    let data = '';
33
    for await (const chunk of stream) {
34
      data += chunk;
35
    }
36
    strictEqual(data, largeData);
37
  },
38
};
39
 
40
function prepareFile() {
41
  const path = '/tmp/elipses.txt';
42
  writeFileSync(path, '…'.repeat(10000));
43
  return path;
44
}
45
 
46
async function runTest(options) {
47
  let paused = false;
48
  let bytesRead = 0;
49
 
50
  const path = prepareFile();
51
 
52
  const file = createReadStream(path, options);
53
  const fileSize = statSync(path).size;
54
 
55
  strictEqual(file.bytesRead, 0);
56
 
57
  const promises = [];
58
 
59
  const { promise: openPromise, resolve: openResolve } =
60
    Promise.withResolvers();
61
  const { promise: endPromise, resolve: endResolve } = Promise.withResolvers();
62
  const { promise: closePromise, resolve: closeResolve } =
63
    Promise.withResolvers();
64
  promises.push(openPromise);
65
  promises.push(endPromise);
66
  promises.push(closePromise);
67
 
68
  const onOpen = mock.fn((fd) => {
69
    file.length = 0;
70
    strictEqual(typeof fd, 'number');
71
    strictEqual(file.bytesRead, 0);
72
    ok(file.readable);
73
    file.pause();
74
    file.resume();
75
    file.pause();
76
    file.resume();
77
    openResolve();
78
  });
79
 
80
  const onData = mock.fn((data) => {
81
    ok(data instanceof Buffer);
82
    ok(data.byteOffset % 8 === 0);
83
    ok(!paused);
84
    file.length += data.length;
85
 
86
    bytesRead += data.length;
87
    strictEqual(file.bytesRead, bytesRead);
88
 
89
    paused = true;
90
    file.pause();
91
 
92
    setTimeout(function () {
93
      paused = false;
94
      file.resume();
95
    }, 10);
96
  });
97
 
98
  const onEnd = mock.fn(() => {
99
    strictEqual(bytesRead, fileSize);
100
    strictEqual(file.bytesRead, fileSize);
101
    endResolve();
102
  });
103
 
104
  const onClose = mock.fn(() => {
105
    strictEqual(bytesRead, fileSize);
106
    strictEqual(file.bytesRead, fileSize);
107
    closeResolve();
108
  });
109
 
110
  file.once('open', onOpen);
111
  file.once('end', onEnd);
112
  file.once('close', onClose);
113
  file.on('data', onData);
114
 
115
  await Promise.all(promises);
116
 
117
  strictEqual(file.length, 30000);
118
  strictEqual(onOpen.mock.callCount(), 1);
119
  strictEqual(onEnd.mock.callCount(), 1);
120
  strictEqual(onClose.mock.callCount(), 1);
121
  strictEqual(onData.mock.callCount(), 1);
122
}
123
 
124
export const readStreamTest1 = {
125
  async test() {
126
    await runTest({});
127
  },
128
};
129
 
130
export const readStreamTest2 = {
131
  async test() {
132
    const customFs = {
133
      open: mock.fn((...args) => openAsync(...args)),
134
      read: mock.fn((...args) => readAsync(...args)),
135
      close: mock.fn((...args) => closeAsync(...args)),
136
    };
137
    await runTest({
138
      fs: customFs,
139
    });
140
    strictEqual(customFs.open.mock.callCount(), 1);
141
    strictEqual(customFs.read.mock.callCount(), 2);
142
    strictEqual(customFs.close.mock.callCount(), 1);
143
  },
144
};
145
 
146
export const readStreamTest3 = {
147
  async test() {
148
    const path = prepareFile();
149
    const file = createReadStream(path, { encoding: 'utf8' });
150
    file.length = 0;
151
    file.on('data', function (data) {
152
      strictEqual(typeof data, 'string');
153
      file.length += data.length;
154
 
155
      for (let i = 0; i < data.length; i++) {
156
        // http://www.fileformat.info/info/unicode/char/2026/index.htm
157
        strictEqual(data[i], '\u2026');
158
      }
159
    });
160
 
161
    const { promise, resolve } = Promise.withResolvers();
162
    file.on('close', resolve);
163
 
164
    await promise;
165
 
166
    strictEqual(file.length, 10000);
167
  },
168
};
169
 
170
export const readStreamTest4 = {
171
  async test() {
172
    const path = '/tmp/x.txt';
173
    writeFileSync(path, 'xyz');
174
    const file = createReadStream(path, { bufferSize: 1, start: 1, end: 2 });
175
    let contentRead = '';
176
    file.on('data', function (data) {
177
      contentRead += data.toString('utf-8');
178
    });
179
    const { promise, resolve } = Promise.withResolvers();
180
    file.on('end', function (data) {
181
      strictEqual(contentRead, 'yz');
182
      resolve();
183
    });
184
    await promise;
185
  },
186
};
187
 
188
export const readStreamTest5 = {
189
  async test() {
190
    const path = '/tmp/x.txt';
191
    writeFileSync(path, 'xyz\n');
192
    const file = createReadStream(path, { bufferSize: 1, start: 1 });
193
    file.data = '';
194
    file.on('data', function (data) {
195
      file.data += data.toString('utf-8');
196
    });
197
    const { promise, resolve } = Promise.withResolvers();
198
    file.on('end', function () {
199
      strictEqual(file.data, 'yz\n');
200
      resolve();
201
    });
202
    await promise;
203
  },
204
};
205
 
206
export const readStreamTest6 = {
207
  async test() {
208
    // Ref: https://github.com/nodejs/node-v0.x-archive/issues/2320
209
    const path = '/tmp/x.txt';
210
    writeFileSync(path, 'xyz\n');
211
    const file = createReadStream(path, { bufferSize: 1.23, start: 1 });
212
    file.data = '';
213
    file.on('data', function (data) {
214
      file.data += data.toString('utf-8');
215
    });
216
    const { promise, resolve } = Promise.withResolvers();
217
    file.on('end', function () {
218
      strictEqual(file.data, 'yz\n');
219
      resolve();
220
    });
221
    await promise;
222
  },
223
};
224
 
225
export const readStreamTest7 = {
226
  test() {
227
    const path = '/tmp/x.txt';
228
    writeFileSync(path, 'xyz\n');
229
    throws(() => createReadStream(path, { start: 10, end: 2 }), {
230
      code: 'ERR_OUT_OF_RANGE',
231
      name: 'RangeError',
232
    });
233
  },
234
};
235
 
236
export const readStreamTest8 = {
237
  async test() {
238
    const path = '/tmp/x.txt';
239
    writeFileSync(path, 'xyz\n');
240
    const stream = createReadStream(path, { start: 0, end: 0 });
241
    stream.data = '';
242
 
243
    stream.on('data', function (chunk) {
244
      stream.data += chunk;
245
    });
246
    const { promise, resolve } = Promise.withResolvers();
247
    stream.on('end', function () {
248
      strictEqual(stream.data, 'x');
249
      resolve();
250
    });
251
    await promise;
252
  },
253
};
254
 
255
export const readStreamTest9 = {
256
  async test() {
257
    // Verify that end works when start is not specified.
258
    const path = '/tmp/x.txt';
259
    writeFileSync(path, 'xyz\n');
260
    const stream = new createReadStream(path, { end: 1 });
261
    stream.data = '';
262
 
263
    stream.on('data', function (chunk) {
264
      stream.data += chunk;
265
    });
266
 
267
    const { promise, resolve } = Promise.withResolvers();
268
    stream.on('end', function () {
269
      strictEqual(stream.data, 'xy');
270
      resolve();
271
    });
272
    await promise;
273
  },
274
};
275
 
276
export const readStreamTest10 = {
277
  async test() {
278
    const path = '/tmp/x.txt';
279
    writeFileSync(path, 'xyz\n');
280
    let file = createReadStream(path, { autoClose: false });
281
    let data = '';
282
    file.on('data', function (chunk) {
283
      data += chunk;
284
    });
285
    const { promise, resolve, reject } = Promise.withResolvers();
286
    file.on('end', function () {
287
      strictEqual(data, 'xyz\n');
288
      process.nextTick(function () {
289
        ok(!file.closed);
290
        ok(!file.destroyed);
291
        fileNext().then(resolve, reject);
292
      });
293
    });
294
 
295
    await promise;
296
 
297
    async function fileNext() {
298
      // This will tell us if the fd is usable again or not.
299
      file = createReadStream(null, { fd: file.fd, start: 0 });
300
      file.data = '';
301
      file.on('data', function (data) {
302
        file.data += data;
303
      });
304
      const { promise, resolve } = Promise.withResolvers();
305
      const { promise: endPromise, resolve: endResolve } =
306
        Promise.withResolvers();
307
      file.on('end', function (err) {
308
        strictEqual(file.data, 'xyz\n');
309
        endResolve();
310
      });
311
      file.on('close', resolve);
312
      await Promise.all([promise, endPromise]);
313
      ok(file.closed);
314
      ok(file.destroyed);
315
    }
316
  },
317
};
318
 
319
export const readStreamTest11 = {
320
  async test() {
321
    // Just to make sure autoClose won't close the stream because of error.
322
    const { promise, resolve, reject } = Promise.withResolvers();
323
    const file = createReadStream(null, { fd: 13337, autoClose: false });
324
    file.on('data', () => reject(new Error('should not be called')));
325
    file.on('error', resolve);
326
    await promise;
327
    ok(!file.closed);
328
    ok(!file.destroyed);
329
    ok(file.fd);
330
  },
331
};
332
 
333
export const readStreamTest12 = {
334
  async test() {
335
    // Make sure stream is destroyed when file does not exist.
336
    const file = createReadStream('/path/to/file/that/does/not/exist');
337
    const { promise, resolve, reject } = Promise.withResolvers();
338
    file.on('data', () => reject(new Error('should not be called')));
339
    file.on('error', resolve);
340
    await promise;
341
    ok(file.closed);
342
    ok(file.destroyed);
343
  },
344
};
345
 
346
export const readStreamTest13 = {
347
  async test() {
348
    const example = '/tmp/x.txt';
349
    writeFileSync(example, 'xyz\n');
350
    createReadStream(example, undefined);
351
    createReadStream(example, null);
352
    createReadStream(example, 'utf8');
353
    createReadStream(example, { encoding: 'utf8' });
354
 
355
    const createReadStreamErr = (path, opt, error) => {
356
      throws(() => createReadStream(path, opt), error);
357
    };
358
 
359
    const typeError = {
360
      code: 'ERR_INVALID_ARG_TYPE',
361
      name: 'TypeError',
362
    };
363
 
364
    const rangeError = {
365
      code: 'ERR_OUT_OF_RANGE',
366
      name: 'RangeError',
367
    };
368
 
369
    [123, 0, true, false].forEach((opts) =>
370
      createReadStreamErr(example, opts, typeError)
371
    );
372
 
373
    // Case 0: Should not throw if either start or end is undefined
374
    [{}, { start: 0 }, { end: Infinity }].forEach((opts) =>
375
      createReadStream(example, opts)
376
    );
377
 
378
    // Case 1: Should throw TypeError if either start or end is not of type 'number'
379
    [
380
      { start: 'invalid' },
381
      { end: 'invalid' },
382
      { start: 'invalid', end: 'invalid' },
383
    ].forEach((opts) => createReadStreamErr(example, opts, typeError));
384
 
385
    // Case 2: Should throw RangeError if either start or end is NaN
386
    [{ start: NaN }, { end: NaN }, { start: NaN, end: NaN }].forEach((opts) =>
387
      createReadStreamErr(example, opts, rangeError)
388
    );
389
 
390
    // Case 3: Should throw RangeError if either start or end is negative
391
    [{ start: -1 }, { end: -1 }, { start: -1, end: -1 }].forEach((opts) =>
392
      createReadStreamErr(example, opts, rangeError)
393
    );
394
 
395
    // Case 4: Should throw RangeError if either start or end is fractional
396
    [{ start: 0.1 }, { end: 0.1 }, { start: 0.1, end: 0.1 }].forEach((opts) =>
397
      createReadStreamErr(example, opts, rangeError)
398
    );
399
 
400
    // Case 5: Should not throw if both start and end are whole numbers
401
    createReadStream(example, { start: 1, end: 5 });
402
 
403
    // Case 6: Should throw RangeError if start is greater than end
404
    createReadStreamErr(example, { start: 5, end: 1 }, rangeError);
405
 
406
    // Case 7: Should throw RangeError if start or end is not safe integer
407
    const NOT_SAFE_INTEGER = 2 ** 53;
408
    [
409
      { start: NOT_SAFE_INTEGER, end: Infinity },
410
      { start: 0, end: NOT_SAFE_INTEGER },
411
    ].forEach((opts) => createReadStreamErr(example, opts, rangeError));
412
  },
413
};
414
 
415
export const readStreamTest14 = {
416
  async test() {
417
    const path = '/tmp/x.txt';
418
    writeFileSync(path, 'xyz\n');
419
    let data = '';
420
    let first = true;
421
 
422
    const stream = createReadStream(path);
423
    stream.setEncoding('utf8');
424
    stream.on('data', function (chunk) {
425
      data += chunk;
426
      if (first) {
427
        first = false;
428
        stream.resume();
429
      }
430
    });
431
 
432
    const { promise, resolve } = Promise.withResolvers();
433
 
434
    process.nextTick(function () {
435
      stream.pause();
436
      setTimeout(function () {
437
        stream.resume();
438
        resolve();
439
      }, 100);
440
    });
441
 
442
    await promise;
443
    strictEqual(data, 'xyz\n');
444
  },
445
};
446
 
447
export const readStreamTest15 = {
448
  async test() {
449
    const { promise, resolve } = Promise.withResolvers();
450
    ReadStream.prototype.open = resolve;
451
    createReadStream('asd');
452
    await promise;
453
    delete ReadStream.prototype.open;
454
  },
455
};
456
 
457
// const fn = fixtures.path('elipses.txt');
458
// const rangeFile = fixtures.path('x.txt');
459
 
460
export const readStreamTest16 = {
461
  async test() {
462
    let paused = false;
463
    const path = prepareFile();
464
 
465
    const file = ReadStream(path);
466
 
467
    const promises = [];
468
    const { promise: openPromise, resolve: openResolve } =
469
      Promise.withResolvers();
470
    const { promise: endPromise, resolve: endResolve } =
471
      Promise.withResolvers();
472
    const { promise: closePromise, resolve: closeResolve } =
473
      Promise.withResolvers();
474
    promises.push(openPromise);
475
    promises.push(endPromise);
476
    promises.push(closePromise);
477
 
478
    file.on('open', function (fd) {
479
      file.length = 0;
480
      strictEqual(typeof fd, 'number');
481
      ok(file.readable);
482
 
483
      // GH-535
484
      file.pause();
485
      file.resume();
486
      file.pause();
487
      file.resume();
488
      openResolve();
489
    });
490
 
491
    file.on('data', function (data) {
492
      ok(data instanceof Buffer);
493
      ok(!paused);
494
      file.length += data.length;
495
 
496
      paused = true;
497
      file.pause();
498
 
499
      setTimeout(function () {
500
        paused = false;
501
        file.resume();
502
      }, 10);
503
    });
504
 
505
    file.on('end', endResolve);
506
 
507
    file.on('close', function () {
508
      strictEqual(file.length, 30000);
509
      closeResolve();
510
    });
511
 
512
    await Promise.all(promises);
513
  },
514
};
515
 
516
export const readStreamTest17 = {
517
  async test() {
518
    const path = prepareFile();
519
    const file = createReadStream(path, { __proto__: { encoding: 'utf8' } });
520
    file.length = 0;
521
    file.on('data', function (data) {
522
      strictEqual(typeof data, 'string');
523
      file.length += data.length;
524
 
525
      for (let i = 0; i < data.length; i++) {
526
        // http://www.fileformat.info/info/unicode/char/2026/index.htm
527
        strictEqual(data[i], '\u2026');
528
      }
529
    });
530
 
531
    const { promise, resolve } = Promise.withResolvers();
532
    file.on('close', function () {
533
      strictEqual(file.length, 10000);
534
      resolve();
535
    });
536
 
537
    await promise;
538
  },
539
};
540
 
541
export const readStreamTest18 = {
542
  async test() {
543
    const path = '/tmp/x.txt';
544
    writeFileSync(path, 'xyz\n');
545
    const options = { __proto__: { bufferSize: 1, start: 1, end: 2 } };
546
    const file = createReadStream(path, options);
547
    strictEqual(file.start, 1);
548
    strictEqual(file.end, 2);
549
    let contentRead = '';
550
    file.on('data', function (data) {
551
      contentRead += data.toString('utf-8');
552
    });
553
 
554
    const { promise, resolve } = Promise.withResolvers();
555
    file.on('end', function () {
556
      strictEqual(contentRead, 'yz');
557
      resolve();
558
    });
559
    await promise;
560
  },
561
};
562
 
563
export const readStreamTest19 = {
564
  async test() {
565
    const path = '/tmp/x.txt';
566
    writeFileSync(path, 'xyz\n');
567
    const options = { __proto__: { bufferSize: 1, start: 1 } };
568
    const file = createReadStream(path, options);
569
    strictEqual(file.start, 1);
570
    file.data = '';
571
    file.on('data', function (data) {
572
      file.data += data.toString('utf-8');
573
    });
574
    const { promise, resolve } = Promise.withResolvers();
575
    file.on('end', function () {
576
      strictEqual(file.data, 'yz\n');
577
      resolve();
578
    });
579
    await promise;
580
  },
581
};
582
 
583
// https://github.com/joyent/node/issues/2320
584
export const readStreamTest20 = {
585
  async test() {
586
    const path = '/tmp/x.txt';
587
    writeFileSync(path, 'xyz\n');
588
    const options = { __proto__: { bufferSize: 1.23, start: 1 } };
589
    const file = createReadStream(path, options);
590
    strictEqual(file.start, 1);
591
    file.data = '';
592
    file.on('data', function (data) {
593
      file.data += data.toString('utf-8');
594
    });
595
    const { promise, resolve } = Promise.withResolvers();
596
    file.on('end', function () {
597
      strictEqual(file.data, 'yz\n');
598
      resolve();
599
    });
600
    await promise;
601
  },
602
};
603
 
604
export const readStreamTest21 = {
605
  test() {
606
    const path = '/tmp/x.txt';
607
    throws(() => createReadStream(path, { __proto__: { start: 10, end: 2 } }), {
608
      code: 'ERR_OUT_OF_RANGE',
609
      name: 'RangeError',
610
    });
611
  },
612
};
613
 
614
export const readStreamTest22 = {
615
  async test() {
616
    const path = '/tmp/x.txt';
617
    writeFileSync(path, 'xyz\n');
618
    const options = { __proto__: { start: 0, end: 0 } };
619
    const stream = createReadStream(path, options);
620
    strictEqual(stream.start, 0);
621
    strictEqual(stream.end, 0);
622
    stream.data = '';
623
 
624
    stream.on('data', function (chunk) {
625
      stream.data += chunk;
626
    });
627
 
628
    const { promise, resolve } = Promise.withResolvers();
629
    stream.on('end', function () {
630
      strictEqual(stream.data, 'x');
631
      resolve();
632
    });
633
    await promise;
634
  },
635
};
636
 
637
export const readStreamTest23 = {
638
  async test() {
639
    const path = '/tmp/x.txt';
640
    let output = '';
641
    writeFileSync(path, 'hello world');
642
    const fd = openSync(path, 'r');
643
    const stream = createReadStream(null, { fd: fd, encoding: 'utf8' });
644
    strictEqual(stream.path, undefined);
645
    stream.on('data', (data) => {
646
      output += data;
647
    });
648
    const { promise, resolve } = Promise.withResolvers();
649
    stream.on('close', resolve);
650
    await promise;
651
    strictEqual(output, 'hello world');
652
  },
653
};
654
 
655
export const readStreamTest24 = {
656
  async test() {
657
    const path = '/tmp/x.txt';
658
    writeFileSync(path, 'hello world');
659
    const stream = createReadStream(path);
660
    const { promise: promise1, resolve: resolve1 } = Promise.withResolvers();
661
    const { promise: promise2, resolve: resolve2 } = Promise.withResolvers();
662
    stream.close(resolve1);
663
    stream.close(resolve2);
664
    await Promise.all([promise1, promise2]);
665
  },
666
};
667
 
668
export const readStreamTest25 = {
669
  async test() {
670
    const path = '/tmp/x.txt';
671
    writeFileSync(path, 'hello world');
672
    const stream = createReadStream(path);
673
    const { promise: promise1, resolve: resolve1 } = Promise.withResolvers();
674
    const { promise: promise2, resolve: resolve2 } = Promise.withResolvers();
675
    stream.destroy(null, resolve1);
676
    stream.destroy(null, resolve2);
677
    await Promise.all([promise1, promise2]);
678
  },
679
};
680
 
681
export const readStreamTest26 = {
682
  async test() {
683
    const path = '/tmp/x.txt';
684
    writeFileSync(path, 'hello world');
685
    const fh = await promises.open(path, 'r');
686
    const { promise: closePromise, resolve: closeResolve } =
687
      Promise.withResolvers();
688
    fh.on('close', closeResolve);
689
    const stream = createReadStream(null, { fd: fh, encoding: 'utf8' });
690
    let data = '';
691
    stream.on('data', (chunk) => (data += chunk));
692
    const { promise, resolve } = Promise.withResolvers();
693
    stream.on('end', () => resolve());
694
    await Promise.all([promise, closePromise]);
695
    strictEqual(data, 'hello world');
696
    strictEqual(fh.fd, undefined);
697
  },
698
};
699
 
700
export const readStreamTest27 = {
701
  async test() {
702
    const path = '/tmp/x.txt';
703
    writeFileSync(path, 'xyz\n');
704
    const handle = await promises.open(path, 'r');
705
    const { promise, resolve, reject } = Promise.withResolvers();
706
    const { promise: closePromise, resolve: closeResolve } =
707
      Promise.withResolvers();
708
    handle.on('close', resolve);
709
    const stream = createReadStream(null, { fd: handle });
710
    stream.on('data', reject);
711
    stream.on('close', closeResolve);
712
    handle.close();
713
    await Promise.all([promise, closePromise]);
714
  },
715
};
716
 
717
export const readStreamTest28 = {
718
  async test() {
719
    const path = '/tmp/x.txt';
720
    writeFileSync(path, 'xyz\n');
721
    const handle = await promises.open(path, 'r');
722
    const { promise: handlePromise, resolve: handleResolve } =
723
      Promise.withResolvers();
724
    const { promise: streamPromise, resolve: streamResolve } =
725
      Promise.withResolvers();
726
    handle.on('close', handleResolve);
727
    const stream = createReadStream(null, { fd: handle });
728
    stream.on('close', streamResolve);
729
    stream.on('data', () => handle.close());
730
    await Promise.all([handlePromise, streamPromise]);
731
  },
732
};
733
 
734
export const readStreamTest29 = {
735
  async test() {
736
    const path = '/tmp/x.txt';
737
    writeFileSync(path, 'xyz\n');
738
    const handle = await promises.open(path, 'r');
739
    const { promise: handlePromise, resolve: handleResolve } =
740
      Promise.withResolvers();
741
    const { promise: streamPromise, resolve: streamResolve } =
742
      Promise.withResolvers();
743
    handle.on('close', handleResolve);
744
    const stream = createReadStream(null, { fd: handle });
745
    stream.on('close', streamResolve);
746
    stream.close();
747
    await Promise.all([handlePromise, streamPromise]);
748
  },
749
};
750
 
751
export const readStreamTest30 = {
752
  async test() {
753
    const path = '/tmp/x.txt';
754
    writeFileSync(path, 'xyz\n');
755
    const handle = await promises.open(path, 'r');
756
    throws(() => createReadStream(null, { fd: handle, fs: {} }), {
757
      code: 'ERR_METHOD_NOT_IMPLEMENTED',
758
    });
759
    handle.close();
760
  },
761
};
762
 
763
export const readStreamTest31 = {
764
  async test() {
765
    // AbortSignal option test
766
    const path = '/tmp/x.txt';
767
    writeFileSync(path, 'xyz\n');
768
    const handle = await promises.open(path, 'r');
769
    const controller = new AbortController();
770
    const { signal } = controller;
771
    const stream = handle.createReadStream({ signal });
772
 
773
    stream.on('data', () => {
774
      throw new Error('boom');
775
    });
776
    stream.on('end', () => {
777
      throw new Error('boom');
778
    });
779
 
780
    const { promise: errorPromise, resolve: errorResolve } =
781
      Promise.withResolvers();
782
    const { promise: closePromise, resolve: closeResolve } =
783
      Promise.withResolvers();
784
    stream.on('error', (err) => {
785
      strictEqual(err.name, 'AbortError');
786
      errorResolve();
787
    });
788
 
789
    handle.on('close', closeResolve);
790
    stream.on('close', () => handle.close());
791
 
792
    controller.abort();
793
 
794
    await Promise.all([errorPromise, closePromise]);
795
  },
796
};
797
 
798
export const readStreamTest32 = {
799
  async test() {
800
    // Already-aborted signal test
801
    const path = '/tmp/x.txt';
802
    writeFileSync(path, 'xyz\n');
803
    const handle = await promises.open(path, 'r');
804
 
805
    const signal = AbortSignal.abort();
806
    const stream = handle.createReadStream({ signal });
807
 
808
    stream.on('data', () => {
809
      throw new Error('boom');
810
    });
811
    stream.on('end', () => {
812
      throw new Error('boom');
813
    });
814
 
815
    const { promise: errorPromise, resolve: errorResolve } =
816
      Promise.withResolvers();
817
    const { promise: closePromise, resolve: closeResolve } =
818
      Promise.withResolvers();
819
 
820
    stream.on('error', (err) => {
821
      strictEqual(err.name, 'AbortError');
822
      errorResolve();
823
    });
824
 
825
    handle.on('close', closeResolve);
826
    stream.on('close', () => handle.close());
827
 
828
    await Promise.all([errorPromise, closePromise]);
829
  },
830
};
831
 
832
export const readStreamTest33 = {
833
  async test() {
834
    // Invalid signal type test
835
    const path = '/tmp/x.txt';
836
    writeFileSync(path, 'xyz\n');
837
    const handle = await promises.open(path, 'r');
838
 
839
    for (const signal of [
840
      1,
841
      {},
842
      [],
843
      '',
844
      NaN,
845
      1n,
846
      () => {},
847
      Symbol(),
848
      false,
849
      true,
850
    ]) {
851
      throws(() => handle.createReadStream({ signal }), {
852
        code: 'ERR_INVALID_ARG_TYPE',
853
        name: 'TypeError',
854
      });
855
    }
856
    handle.close();
857
  },
858
};
859
 
860
export const readStreamTest34 = {
861
  async test() {
862
    // Custom abort reason test
863
    const path = '/tmp/x.txt';
864
    writeFileSync(path, 'xyz\n');
865
    const handle = await promises.open(path, 'r');
866
    const controller = new AbortController();
867
    const { signal } = controller;
868
    const reason = new Error('some silly abort reason');
869
    const stream = handle.createReadStream({ signal });
870
 
871
    const { promise: errorPromise, resolve: errorResolve } =
872
      Promise.withResolvers();
873
    const { promise: closePromise, resolve: closeResolve } =
874
      Promise.withResolvers();
875
 
876
    stream.on('error', (err) => {
877
      strictEqual(err.name, 'AbortError');
878
      strictEqual(err.cause, reason);
879
      errorResolve();
880
    });
881
 
882
    handle.on('close', closeResolve);
883
    stream.on('close', () => handle.close());
884
 
885
    controller.abort(reason);
886
 
887
    await Promise.all([errorPromise, closePromise]);
888
  },
889
};
890
 
891
export const emptyReadStreamTest = {
892
  async test() {
893
    writeFileSync('/tmp/empty.txt', '');
894
    const stream = createReadStream('/tmp/empty.txt');
895
    const { promise, resolve, reject } = Promise.withResolvers();
896
    stream.once('data', () => {
897
      reject(new Error('should not emit data'));
898
    });
899
    stream.once('end', resolve);
900
    await promise;
901
    strictEqual(stream.bytesRead, 0);
902
  },
903
};
904
 
905
export const fileHandleReadableWebStreamTest = {
906
  async test() {
907
    writeFileSync('/tmp/stream.txt', 'abcde'.repeat(1000));
908
    const fh = await promises.open('/tmp/stream.txt', 'r');
909
    const stream = fh.readableWebStream();
910
    const _enc = new TextEncoder();
911
    let data = '';
912
    for await (const chunk of stream) {
913
      strictEqual(chunk instanceof Uint8Array, true);
914
      data += new TextDecoder().decode(chunk);
915
    }
916
    strictEqual(data, 'abcde'.repeat(1000));
917
    strictEqual(fh.fd, undefined);
918
 
919
    // Should throw if the stream is closed.
920
    throws(() => fh.readableWebStream(), {
921
      code: 'EBADF',
922
    });
923
 
924
    const fh2 = await promises.open('/tmp/stream.txt', 'r');
925
    const stream2 = fh2.readableWebStream({ autoClose: false });
926
    await fh2.close();
927
    const res = await stream2.getReader().read();
928
    strictEqual(res.done, true);
929
    strictEqual(res.value, undefined);
930
  },
931
};
932
 
933
/**
934
 * TODO(node-fs): Revisit
935
 * Temporarily comment out. These are larger tests causing timeouts
936
 * In CI. Will move them out to separate tests in a follow on PR
937
export const readStreamTest98 = {
938
  async test() {
939
    const path = prepareFile();
940
    const content = readFileSync(path);
941
942
    const N = 20;
943
    let started = 0;
944
    let done = 0;
945
946
    const arrayBuffers = new Set();
947
    const promises = [];
948
949
    async function startRead() {
950
      ++started;
951
      const chunks = [];
952
      const promises = [];
953
      const { promise, resolve } = Promise.withResolvers();
954
      promises.push(promise);
955
      createReadStream(path)
956
        .on('data', (chunk) => {
957
          chunks.push(chunk);
958
          arrayBuffers.add(chunk.buffer);
959
        })
960
        .on('end', () => {
961
          if (started < N) promises.push(startRead());
962
          deepStrictEqual(Buffer.concat(chunks), content);
963
          if (++done === N) {
964
            const retainedMemory = [...arrayBuffers]
965
              .map((ab) => ab.byteLength)
966
              .reduce((a, b) => a + b);
967
            ok(
968
              retainedMemory / (N * content.length) <= 3,
969
              `Retaining ${retainedMemory} bytes in ABs for ${N} ` +
970
                `chunks of size ${content.length}`
971
            );
972
          }
973
          resolve();
974
        });
975
      await Promise.all(promises);
976
    }
977
978
    // Don’t start the reads all at once – that way we would have to allocate
979
    // a large amount of memory upfront.
980
    for (let i = 0; i < 6; ++i) {
981
      promises.push(startRead());
982
    }
983
    await Promise.all(promises);
984
  },
985
};
986
987
export const readStreamTest99 = {
988
  async test() {
989
    const path = '/tmp/read_stream_pos_test.txt';
990
    writeFileSync(path, '');
991
992
    let counter = 0;
993
994
    const writeInterval = setInterval(() => {
995
      counter = counter + 1;
996
      const line = `hello at ${counter}\n`;
997
      writeFileSync(path, line, { flag: 'a' });
998
    }, 1);
999
1000
    const hwm = 10;
1001
    let bufs = [];
1002
    let isLow = false;
1003
    let cur = 0;
1004
    let stream;
1005
1006
    const readInterval = setInterval(() => {
1007
      if (stream) return;
1008
1009
      stream = createReadStream(path, {
1010
        highWaterMark: hwm,
1011
        start: cur,
1012
      });
1013
      stream.on('data', (chunk) => {
1014
        cur += chunk.length;
1015
        bufs.push(chunk);
1016
        if (isLow) {
1017
          const brokenLines = Buffer.concat(bufs)
1018
            .toString()
1019
            .split('\n')
1020
            .filter((line) => {
1021
              const s = 'hello at'.slice(0, line.length);
1022
              if (line && !line.startsWith(s)) {
1023
                return true;
1024
              }
1025
              return false;
1026
            });
1027
          strictEqual(brokenLines.length, 0);
1028
          exitTest();
1029
          return;
1030
        }
1031
        if (chunk.length !== hwm) {
1032
          isLow = true;
1033
        }
1034
      });
1035
      stream.on('end', () => {
1036
        stream = null;
1037
        isLow = false;
1038
        bufs = [];
1039
      });
1040
    }, 10);
1041
1042
    // Time longer than 10 seconds to exit safely
1043
    await scheduler.wait(5_000);
1044
1045
    clearInterval(readInterval);
1046
    clearInterval(writeInterval);
1047
1048
    if (stream && !stream.destroyed) {
1049
      const { promise, resolve } = Promise.withResolvers();
1050
      stream.on('close', resolve);
1051
      stream.destroy();
1052
      await promise;
1053
    }
1054
  },
1055
};
1056
**/