@@ -32,6 +32,111 @@ export function readableStreamFromIterator(iterator, name = "iterator") {
3232 } ) ;
3333}
3434
35+ export const DEFAULT_BYTE_STREAM_CHUNK_SIZE = 64 * 1024 ;
36+
37+ /**
38+ * Creates a transferable ReadableStream from a byte-stream reader.
39+ *
40+ * Generated p3 stream readers expose `readMany(count)`, which performs one
41+ * bounded Canonical ABI stream read for up to `count` bytes. Prefer that path
42+ * so host byte sinks consume practical chunks instead of one byte at a time.
43+ *
44+ * @param {object } reader - Reader that provides `readMany()`, `[Symbol.asyncIterator]()`, or `read()`.
45+ * @param {object } [opts={}] - Optional adapter settings.
46+ * @param {number } [opts.chunkSize=DEFAULT_BYTE_STREAM_CHUNK_SIZE] - Maximum bytes to request per `readMany()` call.
47+ * @param {string } [opts.name="stream reader"] - Reader name used in error messages.
48+ * @returns {ReadableStream<Uint8Array> } A transferable stream of byte chunks.
49+ */
50+ export function readableByteStreamFromReader ( reader , opts = { } ) {
51+ const source = byteStreamSource ( reader , opts ) ;
52+ return new ReadableStream ( {
53+ async pull ( controller ) {
54+ const { done, value } = await source . read ( ) ;
55+ if ( done ) {
56+ controller . close ( ) ;
57+ return ;
58+ }
59+ controller . enqueue ( byteChunk ( value ) ) ;
60+ } ,
61+ cancel ( reason ) {
62+ return source . cancel ?. ( reason ) ;
63+ } ,
64+ } ) ;
65+ }
66+
67+ function byteStreamSource ( reader , opts ) {
68+ const name = opts . name ?? "stream reader" ;
69+
70+ if ( typeof reader ?. readMany === "function" ) {
71+ const chunkSize = opts . chunkSize ?? DEFAULT_BYTE_STREAM_CHUNK_SIZE ;
72+ if ( ! Number . isInteger ( chunkSize ) || chunkSize < 1 ) {
73+ throw new TypeError ( `invalid byte stream chunk size [${ chunkSize } ) ;
74+ }
75+ return {
76+ async read ( ) {
77+ const result = await reader . readMany ( chunkSize ) ;
78+ if ( result == null || typeof result !== "object" ) {
79+ throw new TypeError ( "readMany() must return an iterator result" ) ;
80+ }
81+ return result ;
82+ } ,
83+ } ;
84+ }
85+ if ( typeof reader ?. [ Symbol . asyncIterator ] === "function" ) {
86+ const iterator = reader [ Symbol . asyncIterator ] ( ) ;
87+ return {
88+ read ( ) {
89+ return iterator . next ( ) ;
90+ } ,
91+ cancel ( reason ) {
92+ return iterator . return ?. ( reason ) ;
93+ } ,
94+ } ;
95+ }
96+ if ( typeof reader ?. read === "function" ) {
97+ return {
98+ async read ( ) {
99+ const value = await reader . read ( ) ;
100+ return { value, done : value === null } ;
101+ } ,
102+ } ;
103+ }
104+
105+ throw new TypeError ( `${ name } ) ;
106+ }
107+
108+ function byteChunk ( value ) {
109+ if ( value instanceof Uint8Array ) {
110+ return value ;
111+ }
112+ if ( value instanceof ArrayBuffer ) {
113+ return new Uint8Array ( value ) ;
114+ }
115+ if ( Array . isArray ( value ) ) {
116+ for ( const byte of value ) {
117+ assertByte ( byte ) ;
118+ }
119+ return Uint8Array . from ( value ) ;
120+ }
121+ if ( typeof value === "number" ) {
122+ return Uint8Array . of ( assertByte ( value ) ) ;
123+ }
124+ if ( typeof value === "string" ) {
125+ return new TextEncoder ( ) . encode ( value ) ;
126+ }
127+
128+ throw new TypeError (
129+ "byte stream chunk must be a byte, byte array, ArrayBuffer, Uint8Array, or string" ,
130+ ) ;
131+ }
132+
133+ function assertByte ( value ) {
134+ if ( ! Number . isInteger ( value ) || value < 0 || value > 255 ) {
135+ throw new RangeError ( `Invalid byte stream value: ${ value } ) ;
136+ }
137+ return value ;
138+ }
139+
35140/**
36141 * Creates a bidirectional stream with separate reader and writer interfaces.
37142 *
0 commit comments