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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
|
Streams are an important abstraction for working with binary data without loading it all into memory at once. They are commonly used for reading and writing files, sending and receiving network requests, and processing large amounts of data.
Bun implements the Web APIs [`ReadableStream`](https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream) and [`WritableStream`](https://developer.mozilla.org/en-US/docs/Web/API/WritableStream).
{% callout %}
Bun also implements the `node:stream` module, including [`Readable`](https://nodejs.org/api/stream.html#stream_readable_streams), [`Writable`](https://nodejs.org/api/stream.html#stream_writable_streams), and [`Duplex`](https://nodejs.org/api/stream.html#stream_duplex_and_transform_streams). For complete documentation, refer to the [Node.js docs](https://nodejs.org/api/stream.html).
{% /callout %}
To create a simple `ReadableStream`:
```ts
const stream = new ReadableStream({
start(controller) {
controller.enqueue("hello");
controller.enqueue("world");
controller.close();
},
});
```
The contents of a `ReadableStream` can be read chunk-by-chunk with `for await` syntax.
```ts
for await (const chunk of stream) {
console.log(chunk);
// => "hello"
// => "world"
}
```
## Direct `ReadableStream`
Bun implements an optimized version of `ReadableStream` that avoid unnecessary data copying & queue management logic. With a traditional `ReadableStream`, chunks of data are _enqueued_. Each chunk is copied into a queue, where it sits until the stream is ready to send more data.
```ts
const stream = new ReadableStream({
start(controller) {
controller.enqueue("hello");
controller.enqueue("world");
controller.close();
},
});
```
With a direct `ReadableStream`, chunks of data are written directly to the stream. No queueing happens, and there's no need to clone the chunk data into memory. The `controller` API is updated to reflect this; instead of `.enqueue()` you call `.write`.
```ts
const stream = new ReadableStream({
type: "direct",
pull(controller) {
controller.write("hello");
controller.write("world");
},
});
```
When using a direct `ReadableStream`, all chunk queueing is handled by the destination. The consumer of the stream receives exactly what is passed to `controller.write()`, without any encoding or modification.
## `Bun.ArrayBufferSink`
The `Bun.ArrayBufferSink` class is a fast incremental writer for constructing an `ArrayBuffer` of unknown size.
```ts
const sink = new Bun.ArrayBufferSink();
sink.write("h");
sink.write("e");
sink.write("l");
sink.write("l");
sink.write("o");
sink.end();
// ArrayBuffer(5) [ 104, 101, 108, 108, 111 ]
```
To instead retrieve the data as a `Uint8Array`, pass the `asUint8Array` option to the constructor.
```ts-diff
const sink = new Bun.ArrayBufferSink({
+ asUint8Array: true
});
sink.write("h");
sink.write("e");
sink.write("l");
sink.write("l");
sink.write("o");
sink.end();
// Uint8Array(5) [ 104, 101, 108, 108, 111 ]
```
The `.write()` method supports strings, typed arrays, `ArrayBuffer`, and `SharedArrayBuffer`.
```ts
sink.write("h");
sink.write(new Uint8Array([101, 108]));
sink.write(Buffer.from("lo").buffer);
sink.end();
```
Once `.end()` is called, no more data can be written to the `ArrayBufferSink`. However, in the context of buffering a stream, it's useful to continuously write data and periodically `.flush()` the contents (say, into a `WriteableStream`). To support this, pass `stream: true` to the constructor.
```ts
const sink = new Bun.ArrayBufferSink({
stream: true,
});
sink.write("h");
sink.write("e");
sink.write("l");
sink.flush();
// ArrayBuffer(5) [ 104, 101, 108 ]
sink.write("l");
sink.write("o");
sink.flush();
// ArrayBuffer(5) [ 108, 111 ]
```
The `.flush()` method returns the buffered data as an `ArrayBuffer` (or `Uint8Array` if `asUint8Array: true`) and clears internal buffer.
To manually set the size of the internal buffer in bytes, pass a value for `highWaterMark`:
```ts
const sink = new Bun.ArrayBufferSink({
highWaterMark: 1024 * 1024, // 1 MB
});
```
{% details summary="Reference" %}
```ts
/**
* Fast incremental writer that becomes an `ArrayBuffer` on end().
*/
export class ArrayBufferSink {
constructor();
start(options?: {
asUint8Array?: boolean;
/**
* Preallocate an internal buffer of this size
* This can significantly improve performance when the chunk size is small
*/
highWaterMark?: number;
/**
* On {@link ArrayBufferSink.flush}, return the written data as a `Uint8Array`.
* Writes will restart from the beginning of the buffer.
*/
stream?: boolean;
}): void;
write(
chunk: string | ArrayBufferView | ArrayBuffer | SharedArrayBuffer,
): number;
/**
* Flush the internal buffer
*
* If {@link ArrayBufferSink.start} was passed a `stream` option, this will return a `ArrayBuffer`
* If {@link ArrayBufferSink.start} was passed a `stream` option and `asUint8Array`, this will return a `Uint8Array`
* Otherwise, this will return the number of bytes written since the last flush
*
* This API might change later to separate Uint8ArraySink and ArrayBufferSink
*/
flush(): number | Uint8Array | ArrayBuffer;
end(): ArrayBuffer | Uint8Array;
}
```
{% /details %}
|
