1
import {
2
  Readable,
3
  Transform,
4
} from 'node:stream';
5
 
6
import {
7
  text,
8
} from 'node:stream/consumers';
9
 
10
import {
11
  pipeline,
12
} from 'node:stream/promises';
13
 
14
class MyTransform extends Transform {
15
  constructor() {
16
    super({ encoding: 'utf8' });
17
  }
18
  _transform(chunk, _, cb) {
19
    this.push(chunk.toString().toUpperCase());
20
    cb();
21
  }
22
  _flush(cb) {
23
    this.push('\n');
24
    cb();
25
  }
26
}
27
 
28
export default {
29
  async fetch() {
30
    const chunks = [
31
      "hello ",
32
      "from ",
33
      "the ",
34
      "wonderful ",
35
      "world ",
36
      "of ",
37
      "node.js ",
38
      "streams!"
39
    ];
40
 
41
    function nextChunk(readable) {
42
      readable.push(chunks.shift());
43
      if (chunks.length === 0) readable.push(null);
44
      else queueMicrotask(() => nextChunk(readable));
45
    }
46
 
47
    const readable = new Readable({
48
      encoding: 'utf8',
49
      read() { nextChunk(readable); }
50
    });
51
 
52
    const transform = new MyTransform();
53
 
54
    await pipeline(readable, transform);
55
 
56
    return new Response(await text(transform));
57
  }
58
};