stream: minor stream/iter implementation edits

Signed-off-by: Renegade334 <contact.9a5d6388@renegade334.me.uk>
PR-URL: #63132
Backport-PR-URL: #64675
Reviewed-By: James M Snell <jasnell@gmail.com>
Reviewed-By: Matteo Collina <matteo.collina@gmail.com>

Author: Renegade334

lib/internal/streams/iter/classic.js

@@ -13,6 +13,7 @@
 
 const{
  ArrayIsArray,
+ ArrayPrototypePush,
  MathMax,
  NumberMAX_SAFE_INTEGER,
  Promise,
@@ -107,14 +108,14 @@ async function normalizeBatch(raw) {
 for(leti=0;i<raw.length;i++){
 constvalue=raw[i];
 if(isUint8Array(value)){
-batch.push(value);
+ArrayPrototypePush(batch,value);
 }else{
 // normalizeAsyncValue may await for async protocols (e.g.
 // toAsyncStreamable on yielded objects). Stream events during
 // the suspension are queued, not lost -- errors will surface
 // on the next loop iteration after this yield completes.
 forawait(constnormalizedofnormalizeAsyncValue(value)){
-batch.push(normalized);
+ArrayPrototypePush(batch,normalized);
 }
 }
 }
@@ -163,7 +164,7 @@ async function* createBatchedAsyncIterator(stream, normalize) {
 stream._readableState?.length>0){
 constc=stream.read();
 if(c===null)break;
-batch.push(c);
+ArrayPrototypePush(batch,c);
 }
 if(normalize!==null){
 constresult=awaitnormalize(batch);
@@ -495,7 +496,7 @@ function fromWritable(writable, options = kNullPrototype) {
 
 functionwaitForDrain(){
 const{ promise, resolve, reject }=PromiseWithResolvers();
-waiters.push({__proto__: null, resolve, reject });
+ArrayPrototypePush(waiters,{__proto__: null, resolve, reject });
 installListeners();
 returnpromise;
 }
@@ -686,7 +687,7 @@ function fromWritable(writable, options = kNullPrototype) {
 returnPromiseResolve(true);
 }
 const{ promise, resolve }=PromiseWithResolvers();
-waiters.push({
+ArrayPrototypePush(waiters,{
 __proto__: null,
 resolve(){resolve(true);},
 reject(){resolve(false);},

lib/internal/streams/iter/consumers.js

@@ -12,6 +12,7 @@ const {
  ArrayBufferPrototypeSlice,
  ArrayPrototypeMap,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  ArrayPrototypeSlice,
  Promise,
  PromisePrototypeThen,
@@ -477,7 +478,7 @@ function merge(...args) {
 
 // Drain ready queue synchronously
 while(ready.length>0){
-constitem=ready.shift();
+constitem=ArrayPrototypeShift(ready);
 if(item?.error){
 throwitem.error;
 }

lib/internal/streams/iter/from.js

@@ -31,8 +31,8 @@ const {
 
 const{
  isAnyArrayBuffer,
- isDataView,
  isPromise,
+ isTypedArray,
  isUint8Array,
 }=require('internal/util/types');
 
@@ -106,17 +106,21 @@ function primitiveToUint8Array(chunk) {
 returnchunk;
 }
 // Other ArrayBufferView types (Int8Array, DataView, etc.)
-if(isDataView(chunk)){
+returnarrayBufferViewToUint8Array(chunk);
+}
+
+functionarrayBufferViewToUint8Array(chunk){
+if(isTypedArray(chunk)){
 returnnewUint8Array(
-DataViewPrototypeGetBuffer(chunk),
-DataViewPrototypeGetByteOffset(chunk),
-DataViewPrototypeGetByteLength(chunk),
+TypedArrayPrototypeGetBuffer(chunk),
+TypedArrayPrototypeGetByteOffset(chunk),
+TypedArrayPrototypeGetByteLength(chunk),
 );
 }
 returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(chunk),
-TypedArrayPrototypeGetByteOffset(chunk),
-TypedArrayPrototypeGetByteLength(chunk),
+DataViewPrototypeGetBuffer(chunk),
+DataViewPrototypeGetByteOffset(chunk),
+DataViewPrototypeGetByteLength(chunk),
 );
 }
 
@@ -580,6 +584,7 @@ function from(input) {
 // =============================================================================
 
 module.exports={
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
  isAsyncIterable,

lib/internal/streams/iter/pull.js

@@ -33,9 +33,9 @@ const {
 const{ AbortController }=require('internal/abort_controller');
 
 const{
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
- primitiveToUint8Array,
  isSyncIterable,
  isAsyncIterable,
  isUint8ArrayBatch,
@@ -136,7 +136,7 @@ function* flattenTransformYieldSync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Must be Iterable<TransformYield>
@@ -170,7 +170,7 @@ async function* flattenTransformYieldAsync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Check for async iterable first
@@ -180,10 +180,10 @@ async function* flattenTransformYieldAsync(value) {
 }
 return;
 }
-// Must be sync Iterable<TransformYield>
+// Must be sync Iterable<TransformYield>, no nested async iterables
 if(isSyncIterable(value)){
 for(constitemofvalue){
-yield*flattenTransformYieldAsync(item);
+yield*flattenTransformYieldSync(item);
 }
 return;
 }
@@ -218,7 +218,7 @@ function* processTransformResultSync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -278,7 +278,7 @@ async function* processTransformResultAsync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -313,7 +313,9 @@ async function* processTransformResultAsync(result) {
 ArrayPrototypePush(batch,item);
 continue;
 }
-forawait(constchunkofflattenTransformYieldAsync(item)){
+// Note: This iteration is synchronous, since async iterables
+// may not be nested within sync iterables.
+for(constchunkofflattenTransformYieldSync(item)){
 ArrayPrototypePush(batch,chunk);
 }
 }
@@ -366,7 +368,7 @@ function* applyFusedStatelessSyncTransforms(source, run) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultSync(current);
 }
@@ -428,7 +430,7 @@ function* createSyncPipeline(source, transforms) {
 }
 current=applyStatefulSyncTransform(current,transform.transform);
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 if(statelessRun.length>0){
@@ -490,7 +492,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultAsync(current);
 }
@@ -531,9 +533,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
  * @yields {Uint8Array[]}
  */
 asyncfunction*withFlushAsync(source){
-forawait(constbatchofsource){
-yieldbatch;
-}
+yield*source;
 yieldnull;
 }
 
@@ -647,7 +647,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
 current,transform.transform,opts);
 }
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 // Flush remaining stateless run

lib/internal/streams/iter/utils.js

@@ -12,6 +12,7 @@ const {
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
+ TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
@@ -24,8 +25,6 @@ const {
 }=require('internal/errors');
 const{ isError }=require('internal/util');
 
-const{ Buffer }=require('buffer');
-
 const{ isSharedArrayBuffer, isUint8Array }=require('internal/util/types');
 
 const{ validateOneOf }=require('internal/validators');
@@ -127,26 +126,36 @@ function concatBytes(chunks) {
 if(chunks.length===0){
 returnnewUint8Array(0);
 }
-// Single chunk: return directly if it covers the entire backing buffer
+// Single chunk: return directly if it covers the entire backing buffer,
+// otherwise return a copy
 if(chunks.length===1){
 constchunk=chunks[0];
-constbuf=TypedArrayPrototypeGetBuffer(chunk);
-// SharedArrayBuffer is not available in primordials, so use
-// direct property access for its byteLength.
-constbufByteLength=isSharedArrayBuffer(buf) ?
-buf.byteLength :
-ArrayBufferPrototypeGetByteLength(buf);
-if(TypedArrayPrototypeGetByteOffset(chunk)===0&&
-TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
-returnchunk;
+// If non-zero offset, skip the remaining buffer checks.
+if(TypedArrayPrototypeGetByteOffset(chunk)===0){
+constbuf=TypedArrayPrototypeGetBuffer(chunk);
+// SharedArrayBuffer is not available in primordials, so use
+// direct property access for its byteLength.
+constbufByteLength=isSharedArrayBuffer(buf) ?
+buf.byteLength :
+ArrayBufferPrototypeGetByteLength(buf);
+if(TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
+returnchunk;
+}
 }
+returnnewUint8Array(chunk);
+}
+// Multiple chunks: concatenate
+lettotalByteLength=0;
+for(leti=0;i<chunks.length;i++){
+totalByteLength+=TypedArrayPrototypeGetByteLength(chunks[i]);
+}
+constconcatenated=newUint8Array(totalByteLength);
+letoffset=0;
+for(leti=0;i<chunks.length;i++){
+TypedArrayPrototypeSet(concatenated,chunks[i],offset);
+offset+=TypedArrayPrototypeGetByteLength(chunks[i]);
 }
-// Multiple chunks or shared buffer: concatenate
-constbuf=Buffer.concat(chunks);
-returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(buf),
-TypedArrayPrototypeGetByteOffset(buf),
-TypedArrayPrototypeGetByteLength(buf));
+returnconcatenated;
 }
 
 /**

lib/internal/streams/iter/classic.js

@@ -13,6 +13,7 @@
 
 const{
  ArrayIsArray,
+ ArrayPrototypePush,
  MathMax,
  NumberMAX_SAFE_INTEGER,
  Promise,
@@ -107,14 +108,14 @@ async function normalizeBatch(raw) {
 for(leti=0;i<raw.length;i++){
 constvalue=raw[i];
 if(isUint8Array(value)){
-batch.push(value);
+ArrayPrototypePush(batch,value);
 }else{
 // normalizeAsyncValue may await for async protocols (e.g.
 // toAsyncStreamable on yielded objects). Stream events during
 // the suspension are queued, not lost -- errors will surface
 // on the next loop iteration after this yield completes.
 forawait(constnormalizedofnormalizeAsyncValue(value)){
-batch.push(normalized);
+ArrayPrototypePush(batch,normalized);
 }
 }
 }
@@ -163,7 +164,7 @@ async function* createBatchedAsyncIterator(stream, normalize) {
 stream._readableState?.length>0){
 constc=stream.read();
 if(c===null)break;
-batch.push(c);
+ArrayPrototypePush(batch,c);
 }
 if(normalize!==null){
 constresult=awaitnormalize(batch);
@@ -495,7 +496,7 @@ function fromWritable(writable, options = kNullPrototype) {
 
 functionwaitForDrain(){
 const{ promise, resolve, reject }=PromiseWithResolvers();
-waiters.push({__proto__: null, resolve, reject });
+ArrayPrototypePush(waiters,{__proto__: null, resolve, reject });
 installListeners();
 returnpromise;
 }
@@ -686,7 +687,7 @@ function fromWritable(writable, options = kNullPrototype) {
 returnPromiseResolve(true);
 }
 const{ promise, resolve }=PromiseWithResolvers();
-waiters.push({
+ArrayPrototypePush(waiters,{
 __proto__: null,
 resolve(){resolve(true);},
 reject(){resolve(false);},

lib/internal/streams/iter/consumers.js

@@ -12,6 +12,7 @@ const {
  ArrayBufferPrototypeSlice,
  ArrayPrototypeMap,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  ArrayPrototypeSlice,
  Promise,
  PromisePrototypeThen,
@@ -477,7 +478,7 @@ function merge(...args) {
 
 // Drain ready queue synchronously
 while(ready.length>0){
-constitem=ready.shift();
+constitem=ArrayPrototypeShift(ready);
 if(item?.error){
 throwitem.error;
 }

lib/internal/streams/iter/from.js

@@ -31,8 +31,8 @@ const {
 
 const{
  isAnyArrayBuffer,
- isDataView,
  isPromise,
+ isTypedArray,
  isUint8Array,
 }=require('internal/util/types');
 
@@ -106,17 +106,21 @@ function primitiveToUint8Array(chunk) {
 returnchunk;
 }
 // Other ArrayBufferView types (Int8Array, DataView, etc.)
-if(isDataView(chunk)){
+returnarrayBufferViewToUint8Array(chunk);
+}
+
+functionarrayBufferViewToUint8Array(chunk){
+if(isTypedArray(chunk)){
 returnnewUint8Array(
-DataViewPrototypeGetBuffer(chunk),
-DataViewPrototypeGetByteOffset(chunk),
-DataViewPrototypeGetByteLength(chunk),
+TypedArrayPrototypeGetBuffer(chunk),
+TypedArrayPrototypeGetByteOffset(chunk),
+TypedArrayPrototypeGetByteLength(chunk),
 );
 }
 returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(chunk),
-TypedArrayPrototypeGetByteOffset(chunk),
-TypedArrayPrototypeGetByteLength(chunk),
+DataViewPrototypeGetBuffer(chunk),
+DataViewPrototypeGetByteOffset(chunk),
+DataViewPrototypeGetByteLength(chunk),
 );
 }
 
@@ -580,6 +584,7 @@ function from(input) {
 // =============================================================================
 
 module.exports={
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
  isAsyncIterable,

lib/internal/streams/iter/pull.js

@@ -33,9 +33,9 @@ const {
 const{ AbortController }=require('internal/abort_controller');
 
 const{
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
- primitiveToUint8Array,
  isSyncIterable,
  isAsyncIterable,
  isUint8ArrayBatch,
@@ -136,7 +136,7 @@ function* flattenTransformYieldSync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Must be Iterable<TransformYield>
@@ -170,7 +170,7 @@ async function* flattenTransformYieldAsync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Check for async iterable first
@@ -180,10 +180,10 @@ async function* flattenTransformYieldAsync(value) {
 }
 return;
 }
-// Must be sync Iterable<TransformYield>
+// Must be sync Iterable<TransformYield>, no nested async iterables
 if(isSyncIterable(value)){
 for(constitemofvalue){
-yield*flattenTransformYieldAsync(item);
+yield*flattenTransformYieldSync(item);
 }
 return;
 }
@@ -218,7 +218,7 @@ function* processTransformResultSync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -278,7 +278,7 @@ async function* processTransformResultAsync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -313,7 +313,9 @@ async function* processTransformResultAsync(result) {
 ArrayPrototypePush(batch,item);
 continue;
 }
-forawait(constchunkofflattenTransformYieldAsync(item)){
+// Note: This iteration is synchronous, since async iterables
+// may not be nested within sync iterables.
+for(constchunkofflattenTransformYieldSync(item)){
 ArrayPrototypePush(batch,chunk);
 }
 }
@@ -366,7 +368,7 @@ function* applyFusedStatelessSyncTransforms(source, run) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultSync(current);
 }
@@ -428,7 +430,7 @@ function* createSyncPipeline(source, transforms) {
 }
 current=applyStatefulSyncTransform(current,transform.transform);
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 if(statelessRun.length>0){
@@ -490,7 +492,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultAsync(current);
 }
@@ -531,9 +533,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
  * @yields {Uint8Array[]}
  */
 asyncfunction*withFlushAsync(source){
-forawait(constbatchofsource){
-yieldbatch;
-}
+yield*source;
 yieldnull;
 }
 
@@ -647,7 +647,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
 current,transform.transform,opts);
 }
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 // Flush remaining stateless run

lib/internal/streams/iter/utils.js

@@ -12,6 +12,7 @@ const {
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
+ TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
@@ -24,8 +25,6 @@ const {
 }=require('internal/errors');
 const{ isError }=require('internal/util');
 
-const{ Buffer }=require('buffer');
-
 const{ isSharedArrayBuffer, isUint8Array }=require('internal/util/types');
 
 const{ validateOneOf }=require('internal/validators');
@@ -127,26 +126,36 @@ function concatBytes(chunks) {
 if(chunks.length===0){
 returnnewUint8Array(0);
 }
-// Single chunk: return directly if it covers the entire backing buffer
+// Single chunk: return directly if it covers the entire backing buffer,
+// otherwise return a copy
 if(chunks.length===1){
 constchunk=chunks[0];
-constbuf=TypedArrayPrototypeGetBuffer(chunk);
-// SharedArrayBuffer is not available in primordials, so use
-// direct property access for its byteLength.
-constbufByteLength=isSharedArrayBuffer(buf) ?
-buf.byteLength :
-ArrayBufferPrototypeGetByteLength(buf);
-if(TypedArrayPrototypeGetByteOffset(chunk)===0&&
-TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
-returnchunk;
+// If non-zero offset, skip the remaining buffer checks.
+if(TypedArrayPrototypeGetByteOffset(chunk)===0){
+constbuf=TypedArrayPrototypeGetBuffer(chunk);
+// SharedArrayBuffer is not available in primordials, so use
+// direct property access for its byteLength.
+constbufByteLength=isSharedArrayBuffer(buf) ?
+buf.byteLength :
+ArrayBufferPrototypeGetByteLength(buf);
+if(TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
+returnchunk;
+}
 }
+returnnewUint8Array(chunk);
+}
+// Multiple chunks: concatenate
+lettotalByteLength=0;
+for(leti=0;i<chunks.length;i++){
+totalByteLength+=TypedArrayPrototypeGetByteLength(chunks[i]);
+}
+constconcatenated=newUint8Array(totalByteLength);
+letoffset=0;
+for(leti=0;i<chunks.length;i++){
+TypedArrayPrototypeSet(concatenated,chunks[i],offset);
+offset+=TypedArrayPrototypeGetByteLength(chunks[i]);
 }
-// Multiple chunks or shared buffer: concatenate
-constbuf=Buffer.concat(chunks);
-returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(buf),
-TypedArrayPrototypeGetByteOffset(buf),
-TypedArrayPrototypeGetByteLength(buf));
+returnconcatenated;
 }
 
 /**

lib/internal/streams/iter/classic.js

@@ -13,6 +13,7 @@
 
 const{
  ArrayIsArray,
+ ArrayPrototypePush,
  MathMax,
  NumberMAX_SAFE_INTEGER,
  Promise,
@@ -107,14 +108,14 @@ async function normalizeBatch(raw) {
 for(leti=0;i<raw.length;i++){
 constvalue=raw[i];
 if(isUint8Array(value)){
-batch.push(value);
+ArrayPrototypePush(batch,value);
 }else{
 // normalizeAsyncValue may await for async protocols (e.g.
 // toAsyncStreamable on yielded objects). Stream events during
 // the suspension are queued, not lost -- errors will surface
 // on the next loop iteration after this yield completes.
 forawait(constnormalizedofnormalizeAsyncValue(value)){
-batch.push(normalized);
+ArrayPrototypePush(batch,normalized);
 }
 }
 }
@@ -163,7 +164,7 @@ async function* createBatchedAsyncIterator(stream, normalize) {
 stream._readableState?.length>0){
 constc=stream.read();
 if(c===null)break;
-batch.push(c);
+ArrayPrototypePush(batch,c);
 }
 if(normalize!==null){
 constresult=awaitnormalize(batch);
@@ -495,7 +496,7 @@ function fromWritable(writable, options = kNullPrototype) {
 
 functionwaitForDrain(){
 const{ promise, resolve, reject }=PromiseWithResolvers();
-waiters.push({__proto__: null, resolve, reject });
+ArrayPrototypePush(waiters,{__proto__: null, resolve, reject });
 installListeners();
 returnpromise;
 }
@@ -686,7 +687,7 @@ function fromWritable(writable, options = kNullPrototype) {
 returnPromiseResolve(true);
 }
 const{ promise, resolve }=PromiseWithResolvers();
-waiters.push({
+ArrayPrototypePush(waiters,{
 __proto__: null,
 resolve(){resolve(true);},
 reject(){resolve(false);},

lib/internal/streams/iter/consumers.js

@@ -12,6 +12,7 @@ const {
  ArrayBufferPrototypeSlice,
  ArrayPrototypeMap,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  ArrayPrototypeSlice,
  Promise,
  PromisePrototypeThen,
@@ -477,7 +478,7 @@ function merge(...args) {
 
 // Drain ready queue synchronously
 while(ready.length>0){
-constitem=ready.shift();
+constitem=ArrayPrototypeShift(ready);
 if(item?.error){
 throwitem.error;
 }

lib/internal/streams/iter/from.js

@@ -31,8 +31,8 @@ const {
 
 const{
  isAnyArrayBuffer,
- isDataView,
  isPromise,
+ isTypedArray,
  isUint8Array,
 }=require('internal/util/types');
 
@@ -106,17 +106,21 @@ function primitiveToUint8Array(chunk) {
 returnchunk;
 }
 // Other ArrayBufferView types (Int8Array, DataView, etc.)
-if(isDataView(chunk)){
+returnarrayBufferViewToUint8Array(chunk);
+}
+
+functionarrayBufferViewToUint8Array(chunk){
+if(isTypedArray(chunk)){
 returnnewUint8Array(
-DataViewPrototypeGetBuffer(chunk),
-DataViewPrototypeGetByteOffset(chunk),
-DataViewPrototypeGetByteLength(chunk),
+TypedArrayPrototypeGetBuffer(chunk),
+TypedArrayPrototypeGetByteOffset(chunk),
+TypedArrayPrototypeGetByteLength(chunk),
 );
 }
 returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(chunk),
-TypedArrayPrototypeGetByteOffset(chunk),
-TypedArrayPrototypeGetByteLength(chunk),
+DataViewPrototypeGetBuffer(chunk),
+DataViewPrototypeGetByteOffset(chunk),
+DataViewPrototypeGetByteLength(chunk),
 );
 }
 
@@ -580,6 +584,7 @@ function from(input) {
 // =============================================================================
 
 module.exports={
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
  isAsyncIterable,

lib/internal/streams/iter/pull.js

@@ -33,9 +33,9 @@ const {
 const{ AbortController }=require('internal/abort_controller');
 
 const{
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
- primitiveToUint8Array,
  isSyncIterable,
  isAsyncIterable,
  isUint8ArrayBatch,
@@ -136,7 +136,7 @@ function* flattenTransformYieldSync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Must be Iterable<TransformYield>
@@ -170,7 +170,7 @@ async function* flattenTransformYieldAsync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Check for async iterable first
@@ -180,10 +180,10 @@ async function* flattenTransformYieldAsync(value) {
 }
 return;
 }
-// Must be sync Iterable<TransformYield>
+// Must be sync Iterable<TransformYield>, no nested async iterables
 if(isSyncIterable(value)){
 for(constitemofvalue){
-yield*flattenTransformYieldAsync(item);
+yield*flattenTransformYieldSync(item);
 }
 return;
 }
@@ -218,7 +218,7 @@ function* processTransformResultSync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -278,7 +278,7 @@ async function* processTransformResultAsync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -313,7 +313,9 @@ async function* processTransformResultAsync(result) {
 ArrayPrototypePush(batch,item);
 continue;
 }
-forawait(constchunkofflattenTransformYieldAsync(item)){
+// Note: This iteration is synchronous, since async iterables
+// may not be nested within sync iterables.
+for(constchunkofflattenTransformYieldSync(item)){
 ArrayPrototypePush(batch,chunk);
 }
 }
@@ -366,7 +368,7 @@ function* applyFusedStatelessSyncTransforms(source, run) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultSync(current);
 }
@@ -428,7 +430,7 @@ function* createSyncPipeline(source, transforms) {
 }
 current=applyStatefulSyncTransform(current,transform.transform);
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 if(statelessRun.length>0){
@@ -490,7 +492,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultAsync(current);
 }
@@ -531,9 +533,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
  * @yields {Uint8Array[]}
  */
 asyncfunction*withFlushAsync(source){
-forawait(constbatchofsource){
-yieldbatch;
-}
+yield*source;
 yieldnull;
 }
 
@@ -647,7 +647,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
 current,transform.transform,opts);
 }
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 // Flush remaining stateless run

lib/internal/streams/iter/utils.js

@@ -12,6 +12,7 @@ const {
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
+ TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
@@ -24,8 +25,6 @@ const {
 }=require('internal/errors');
 const{ isError }=require('internal/util');
 
-const{ Buffer }=require('buffer');
-
 const{ isSharedArrayBuffer, isUint8Array }=require('internal/util/types');
 
 const{ validateOneOf }=require('internal/validators');
@@ -127,26 +126,36 @@ function concatBytes(chunks) {
 if(chunks.length===0){
 returnnewUint8Array(0);
 }
-// Single chunk: return directly if it covers the entire backing buffer
+// Single chunk: return directly if it covers the entire backing buffer,
+// otherwise return a copy
 if(chunks.length===1){
 constchunk=chunks[0];
-constbuf=TypedArrayPrototypeGetBuffer(chunk);
-// SharedArrayBuffer is not available in primordials, so use
-// direct property access for its byteLength.
-constbufByteLength=isSharedArrayBuffer(buf) ?
-buf.byteLength :
-ArrayBufferPrototypeGetByteLength(buf);
-if(TypedArrayPrototypeGetByteOffset(chunk)===0&&
-TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
-returnchunk;
+// If non-zero offset, skip the remaining buffer checks.
+if(TypedArrayPrototypeGetByteOffset(chunk)===0){
+constbuf=TypedArrayPrototypeGetBuffer(chunk);
+// SharedArrayBuffer is not available in primordials, so use
+// direct property access for its byteLength.
+constbufByteLength=isSharedArrayBuffer(buf) ?
+buf.byteLength :
+ArrayBufferPrototypeGetByteLength(buf);
+if(TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
+returnchunk;
+}
 }
+returnnewUint8Array(chunk);
+}
+// Multiple chunks: concatenate
+lettotalByteLength=0;
+for(leti=0;i<chunks.length;i++){
+totalByteLength+=TypedArrayPrototypeGetByteLength(chunks[i]);
+}
+constconcatenated=newUint8Array(totalByteLength);
+letoffset=0;
+for(leti=0;i<chunks.length;i++){
+TypedArrayPrototypeSet(concatenated,chunks[i],offset);
+offset+=TypedArrayPrototypeGetByteLength(chunks[i]);
 }
-// Multiple chunks or shared buffer: concatenate
-constbuf=Buffer.concat(chunks);
-returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(buf),
-TypedArrayPrototypeGetByteOffset(buf),
-TypedArrayPrototypeGetByteLength(buf));
+returnconcatenated;
 }
 
 /**

lib/internal/streams/iter/classic.js

@@ -13,6 +13,7 @@
 
 const{
  ArrayIsArray,
+ ArrayPrototypePush,
  MathMax,
  NumberMAX_SAFE_INTEGER,
  Promise,
@@ -107,14 +108,14 @@ async function normalizeBatch(raw) {
 for(leti=0;i<raw.length;i++){
 constvalue=raw[i];
 if(isUint8Array(value)){
-batch.push(value);
+ArrayPrototypePush(batch,value);
 }else{
 // normalizeAsyncValue may await for async protocols (e.g.
 // toAsyncStreamable on yielded objects). Stream events during
 // the suspension are queued, not lost -- errors will surface
 // on the next loop iteration after this yield completes.
 forawait(constnormalizedofnormalizeAsyncValue(value)){
-batch.push(normalized);
+ArrayPrototypePush(batch,normalized);
 }
 }
 }
@@ -163,7 +164,7 @@ async function* createBatchedAsyncIterator(stream, normalize) {
 stream._readableState?.length>0){
 constc=stream.read();
 if(c===null)break;
-batch.push(c);
+ArrayPrototypePush(batch,c);
 }
 if(normalize!==null){
 constresult=awaitnormalize(batch);
@@ -495,7 +496,7 @@ function fromWritable(writable, options = kNullPrototype) {
 
 functionwaitForDrain(){
 const{ promise, resolve, reject }=PromiseWithResolvers();
-waiters.push({__proto__: null, resolve, reject });
+ArrayPrototypePush(waiters,{__proto__: null, resolve, reject });
 installListeners();
 returnpromise;
 }
@@ -686,7 +687,7 @@ function fromWritable(writable, options = kNullPrototype) {
 returnPromiseResolve(true);
 }
 const{ promise, resolve }=PromiseWithResolvers();
-waiters.push({
+ArrayPrototypePush(waiters,{
 __proto__: null,
 resolve(){resolve(true);},
 reject(){resolve(false);},

lib/internal/streams/iter/consumers.js

@@ -12,6 +12,7 @@ const {
  ArrayBufferPrototypeSlice,
  ArrayPrototypeMap,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  ArrayPrototypeSlice,
  Promise,
  PromisePrototypeThen,
@@ -477,7 +478,7 @@ function merge(...args) {
 
 // Drain ready queue synchronously
 while(ready.length>0){
-constitem=ready.shift();
+constitem=ArrayPrototypeShift(ready);
 if(item?.error){
 throwitem.error;
 }

lib/internal/streams/iter/from.js

@@ -31,8 +31,8 @@ const {
 
 const{
  isAnyArrayBuffer,
- isDataView,
  isPromise,
+ isTypedArray,
  isUint8Array,
 }=require('internal/util/types');
 
@@ -106,17 +106,21 @@ function primitiveToUint8Array(chunk) {
 returnchunk;
 }
 // Other ArrayBufferView types (Int8Array, DataView, etc.)
-if(isDataView(chunk)){
+returnarrayBufferViewToUint8Array(chunk);
+}
+
+functionarrayBufferViewToUint8Array(chunk){
+if(isTypedArray(chunk)){
 returnnewUint8Array(
-DataViewPrototypeGetBuffer(chunk),
-DataViewPrototypeGetByteOffset(chunk),
-DataViewPrototypeGetByteLength(chunk),
+TypedArrayPrototypeGetBuffer(chunk),
+TypedArrayPrototypeGetByteOffset(chunk),
+TypedArrayPrototypeGetByteLength(chunk),
 );
 }
 returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(chunk),
-TypedArrayPrototypeGetByteOffset(chunk),
-TypedArrayPrototypeGetByteLength(chunk),
+DataViewPrototypeGetBuffer(chunk),
+DataViewPrototypeGetByteOffset(chunk),
+DataViewPrototypeGetByteLength(chunk),
 );
 }
 
@@ -580,6 +584,7 @@ function from(input) {
 // =============================================================================
 
 module.exports={
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
  isAsyncIterable,

lib/internal/streams/iter/pull.js

@@ -33,9 +33,9 @@ const {
 const{ AbortController }=require('internal/abort_controller');
 
 const{
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
- primitiveToUint8Array,
  isSyncIterable,
  isAsyncIterable,
  isUint8ArrayBatch,
@@ -136,7 +136,7 @@ function* flattenTransformYieldSync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Must be Iterable<TransformYield>
@@ -170,7 +170,7 @@ async function* flattenTransformYieldAsync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Check for async iterable first
@@ -180,10 +180,10 @@ async function* flattenTransformYieldAsync(value) {
 }
 return;
 }
-// Must be sync Iterable<TransformYield>
+// Must be sync Iterable<TransformYield>, no nested async iterables
 if(isSyncIterable(value)){
 for(constitemofvalue){
-yield*flattenTransformYieldAsync(item);
+yield*flattenTransformYieldSync(item);
 }
 return;
 }
@@ -218,7 +218,7 @@ function* processTransformResultSync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -278,7 +278,7 @@ async function* processTransformResultAsync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -313,7 +313,9 @@ async function* processTransformResultAsync(result) {
 ArrayPrototypePush(batch,item);
 continue;
 }
-forawait(constchunkofflattenTransformYieldAsync(item)){
+// Note: This iteration is synchronous, since async iterables
+// may not be nested within sync iterables.
+for(constchunkofflattenTransformYieldSync(item)){
 ArrayPrototypePush(batch,chunk);
 }
 }
@@ -366,7 +368,7 @@ function* applyFusedStatelessSyncTransforms(source, run) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultSync(current);
 }
@@ -428,7 +430,7 @@ function* createSyncPipeline(source, transforms) {
 }
 current=applyStatefulSyncTransform(current,transform.transform);
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 if(statelessRun.length>0){
@@ -490,7 +492,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultAsync(current);
 }
@@ -531,9 +533,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
  * @yields {Uint8Array[]}
  */
 asyncfunction*withFlushAsync(source){
-forawait(constbatchofsource){
-yieldbatch;
-}
+yield*source;
 yieldnull;
 }
 
@@ -647,7 +647,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
 current,transform.transform,opts);
 }
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 // Flush remaining stateless run

lib/internal/streams/iter/utils.js

@@ -12,6 +12,7 @@ const {
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
+ TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
@@ -24,8 +25,6 @@ const {
 }=require('internal/errors');
 const{ isError }=require('internal/util');
 
-const{ Buffer }=require('buffer');
-
 const{ isSharedArrayBuffer, isUint8Array }=require('internal/util/types');
 
 const{ validateOneOf }=require('internal/validators');
@@ -127,26 +126,36 @@ function concatBytes(chunks) {
 if(chunks.length===0){
 returnnewUint8Array(0);
 }
-// Single chunk: return directly if it covers the entire backing buffer
+// Single chunk: return directly if it covers the entire backing buffer,
+// otherwise return a copy
 if(chunks.length===1){
 constchunk=chunks[0];
-constbuf=TypedArrayPrototypeGetBuffer(chunk);
-// SharedArrayBuffer is not available in primordials, so use
-// direct property access for its byteLength.
-constbufByteLength=isSharedArrayBuffer(buf) ?
-buf.byteLength :
-ArrayBufferPrototypeGetByteLength(buf);
-if(TypedArrayPrototypeGetByteOffset(chunk)===0&&
-TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
-returnchunk;
+// If non-zero offset, skip the remaining buffer checks.
+if(TypedArrayPrototypeGetByteOffset(chunk)===0){
+constbuf=TypedArrayPrototypeGetBuffer(chunk);
+// SharedArrayBuffer is not available in primordials, so use
+// direct property access for its byteLength.
+constbufByteLength=isSharedArrayBuffer(buf) ?
+buf.byteLength :
+ArrayBufferPrototypeGetByteLength(buf);
+if(TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
+returnchunk;
+}
 }
+returnnewUint8Array(chunk);
+}
+// Multiple chunks: concatenate
+lettotalByteLength=0;
+for(leti=0;i<chunks.length;i++){
+totalByteLength+=TypedArrayPrototypeGetByteLength(chunks[i]);
+}
+constconcatenated=newUint8Array(totalByteLength);
+letoffset=0;
+for(leti=0;i<chunks.length;i++){
+TypedArrayPrototypeSet(concatenated,chunks[i],offset);
+offset+=TypedArrayPrototypeGetByteLength(chunks[i]);
 }
-// Multiple chunks or shared buffer: concatenate
-constbuf=Buffer.concat(chunks);
-returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(buf),
-TypedArrayPrototypeGetByteOffset(buf),
-TypedArrayPrototypeGetByteLength(buf));
+returnconcatenated;
 }
 
 /**

lib/internal/streams/iter/classic.js

@@ -13,6 +13,7 @@
 
 const{
  ArrayIsArray,
+ ArrayPrototypePush,
  MathMax,
  NumberMAX_SAFE_INTEGER,
  Promise,
@@ -107,14 +108,14 @@ async function normalizeBatch(raw) {
 for(leti=0;i<raw.length;i++){
 constvalue=raw[i];
 if(isUint8Array(value)){
-batch.push(value);
+ArrayPrototypePush(batch,value);
 }else{
 // normalizeAsyncValue may await for async protocols (e.g.
 // toAsyncStreamable on yielded objects). Stream events during
 // the suspension are queued, not lost -- errors will surface
 // on the next loop iteration after this yield completes.
 forawait(constnormalizedofnormalizeAsyncValue(value)){
-batch.push(normalized);
+ArrayPrototypePush(batch,normalized);
 }
 }
 }
@@ -163,7 +164,7 @@ async function* createBatchedAsyncIterator(stream, normalize) {
 stream._readableState?.length>0){
 constc=stream.read();
 if(c===null)break;
-batch.push(c);
+ArrayPrototypePush(batch,c);
 }
 if(normalize!==null){
 constresult=awaitnormalize(batch);
@@ -495,7 +496,7 @@ function fromWritable(writable, options = kNullPrototype) {
 
 functionwaitForDrain(){
 const{ promise, resolve, reject }=PromiseWithResolvers();
-waiters.push({__proto__: null, resolve, reject });
+ArrayPrototypePush(waiters,{__proto__: null, resolve, reject });
 installListeners();
 returnpromise;
 }
@@ -686,7 +687,7 @@ function fromWritable(writable, options = kNullPrototype) {
 returnPromiseResolve(true);
 }
 const{ promise, resolve }=PromiseWithResolvers();
-waiters.push({
+ArrayPrototypePush(waiters,{
 __proto__: null,
 resolve(){resolve(true);},
 reject(){resolve(false);},

lib/internal/streams/iter/consumers.js

@@ -12,6 +12,7 @@ const {
  ArrayBufferPrototypeSlice,
  ArrayPrototypeMap,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  ArrayPrototypeSlice,
  Promise,
  PromisePrototypeThen,
@@ -477,7 +478,7 @@ function merge(...args) {
 
 // Drain ready queue synchronously
 while(ready.length>0){
-constitem=ready.shift();
+constitem=ArrayPrototypeShift(ready);
 if(item?.error){
 throwitem.error;
 }

lib/internal/streams/iter/from.js

@@ -31,8 +31,8 @@ const {
 
 const{
  isAnyArrayBuffer,
- isDataView,
  isPromise,
+ isTypedArray,
  isUint8Array,
 }=require('internal/util/types');
 
@@ -106,17 +106,21 @@ function primitiveToUint8Array(chunk) {
 returnchunk;
 }
 // Other ArrayBufferView types (Int8Array, DataView, etc.)
-if(isDataView(chunk)){
+returnarrayBufferViewToUint8Array(chunk);
+}
+
+functionarrayBufferViewToUint8Array(chunk){
+if(isTypedArray(chunk)){
 returnnewUint8Array(
-DataViewPrototypeGetBuffer(chunk),
-DataViewPrototypeGetByteOffset(chunk),
-DataViewPrototypeGetByteLength(chunk),
+TypedArrayPrototypeGetBuffer(chunk),
+TypedArrayPrototypeGetByteOffset(chunk),
+TypedArrayPrototypeGetByteLength(chunk),
 );
 }
 returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(chunk),
-TypedArrayPrototypeGetByteOffset(chunk),
-TypedArrayPrototypeGetByteLength(chunk),
+DataViewPrototypeGetBuffer(chunk),
+DataViewPrototypeGetByteOffset(chunk),
+DataViewPrototypeGetByteLength(chunk),
 );
 }
 
@@ -580,6 +584,7 @@ function from(input) {
 // =============================================================================
 
 module.exports={
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
  isAsyncIterable,

lib/internal/streams/iter/pull.js

@@ -33,9 +33,9 @@ const {
 const{ AbortController }=require('internal/abort_controller');
 
 const{
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
- primitiveToUint8Array,
  isSyncIterable,
  isAsyncIterable,
  isUint8ArrayBatch,
@@ -136,7 +136,7 @@ function* flattenTransformYieldSync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Must be Iterable<TransformYield>
@@ -170,7 +170,7 @@ async function* flattenTransformYieldAsync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Check for async iterable first
@@ -180,10 +180,10 @@ async function* flattenTransformYieldAsync(value) {
 }
 return;
 }
-// Must be sync Iterable<TransformYield>
+// Must be sync Iterable<TransformYield>, no nested async iterables
 if(isSyncIterable(value)){
 for(constitemofvalue){
-yield*flattenTransformYieldAsync(item);
+yield*flattenTransformYieldSync(item);
 }
 return;
 }
@@ -218,7 +218,7 @@ function* processTransformResultSync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -278,7 +278,7 @@ async function* processTransformResultAsync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -313,7 +313,9 @@ async function* processTransformResultAsync(result) {
 ArrayPrototypePush(batch,item);
 continue;
 }
-forawait(constchunkofflattenTransformYieldAsync(item)){
+// Note: This iteration is synchronous, since async iterables
+// may not be nested within sync iterables.
+for(constchunkofflattenTransformYieldSync(item)){
 ArrayPrototypePush(batch,chunk);
 }
 }
@@ -366,7 +368,7 @@ function* applyFusedStatelessSyncTransforms(source, run) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultSync(current);
 }
@@ -428,7 +430,7 @@ function* createSyncPipeline(source, transforms) {
 }
 current=applyStatefulSyncTransform(current,transform.transform);
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 if(statelessRun.length>0){
@@ -490,7 +492,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultAsync(current);
 }
@@ -531,9 +533,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
  * @yields {Uint8Array[]}
  */
 asyncfunction*withFlushAsync(source){
-forawait(constbatchofsource){
-yieldbatch;
-}
+yield*source;
 yieldnull;
 }
 
@@ -647,7 +647,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
 current,transform.transform,opts);
 }
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 // Flush remaining stateless run

lib/internal/streams/iter/utils.js

@@ -12,6 +12,7 @@ const {
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
+ TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
@@ -24,8 +25,6 @@ const {
 }=require('internal/errors');
 const{ isError }=require('internal/util');
 
-const{ Buffer }=require('buffer');
-
 const{ isSharedArrayBuffer, isUint8Array }=require('internal/util/types');
 
 const{ validateOneOf }=require('internal/validators');
@@ -127,26 +126,36 @@ function concatBytes(chunks) {
 if(chunks.length===0){
 returnnewUint8Array(0);
 }
-// Single chunk: return directly if it covers the entire backing buffer
+// Single chunk: return directly if it covers the entire backing buffer,
+// otherwise return a copy
 if(chunks.length===1){
 constchunk=chunks[0];
-constbuf=TypedArrayPrototypeGetBuffer(chunk);
-// SharedArrayBuffer is not available in primordials, so use
-// direct property access for its byteLength.
-constbufByteLength=isSharedArrayBuffer(buf) ?
-buf.byteLength :
-ArrayBufferPrototypeGetByteLength(buf);
-if(TypedArrayPrototypeGetByteOffset(chunk)===0&&
-TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
-returnchunk;
+// If non-zero offset, skip the remaining buffer checks.
+if(TypedArrayPrototypeGetByteOffset(chunk)===0){
+constbuf=TypedArrayPrototypeGetBuffer(chunk);
+// SharedArrayBuffer is not available in primordials, so use
+// direct property access for its byteLength.
+constbufByteLength=isSharedArrayBuffer(buf) ?
+buf.byteLength :
+ArrayBufferPrototypeGetByteLength(buf);
+if(TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
+returnchunk;
+}
 }
+returnnewUint8Array(chunk);
+}
+// Multiple chunks: concatenate
+lettotalByteLength=0;
+for(leti=0;i<chunks.length;i++){
+totalByteLength+=TypedArrayPrototypeGetByteLength(chunks[i]);
+}
+constconcatenated=newUint8Array(totalByteLength);
+letoffset=0;
+for(leti=0;i<chunks.length;i++){
+TypedArrayPrototypeSet(concatenated,chunks[i],offset);
+offset+=TypedArrayPrototypeGetByteLength(chunks[i]);
 }
-// Multiple chunks or shared buffer: concatenate
-constbuf=Buffer.concat(chunks);
-returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(buf),
-TypedArrayPrototypeGetByteOffset(buf),
-TypedArrayPrototypeGetByteLength(buf));
+returnconcatenated;
 }
 
 /**

lib/internal/streams/iter/classic.js

@@ -13,6 +13,7 @@
 
 const{
  ArrayIsArray,
+ ArrayPrototypePush,
  MathMax,
  NumberMAX_SAFE_INTEGER,
  Promise,
@@ -107,14 +108,14 @@ async function normalizeBatch(raw) {
 for(leti=0;i<raw.length;i++){
 constvalue=raw[i];
 if(isUint8Array(value)){
-batch.push(value);
+ArrayPrototypePush(batch,value);
 }else{
 // normalizeAsyncValue may await for async protocols (e.g.
 // toAsyncStreamable on yielded objects). Stream events during
 // the suspension are queued, not lost -- errors will surface
 // on the next loop iteration after this yield completes.
 forawait(constnormalizedofnormalizeAsyncValue(value)){
-batch.push(normalized);
+ArrayPrototypePush(batch,normalized);
 }
 }
 }
@@ -163,7 +164,7 @@ async function* createBatchedAsyncIterator(stream, normalize) {
 stream._readableState?.length>0){
 constc=stream.read();
 if(c===null)break;
-batch.push(c);
+ArrayPrototypePush(batch,c);
 }
 if(normalize!==null){
 constresult=awaitnormalize(batch);
@@ -495,7 +496,7 @@ function fromWritable(writable, options = kNullPrototype) {
 
 functionwaitForDrain(){
 const{ promise, resolve, reject }=PromiseWithResolvers();
-waiters.push({__proto__: null, resolve, reject });
+ArrayPrototypePush(waiters,{__proto__: null, resolve, reject });
 installListeners();
 returnpromise;
 }
@@ -686,7 +687,7 @@ function fromWritable(writable, options = kNullPrototype) {
 returnPromiseResolve(true);
 }
 const{ promise, resolve }=PromiseWithResolvers();
-waiters.push({
+ArrayPrototypePush(waiters,{
 __proto__: null,
 resolve(){resolve(true);},
 reject(){resolve(false);},

lib/internal/streams/iter/consumers.js

@@ -12,6 +12,7 @@ const {
  ArrayBufferPrototypeSlice,
  ArrayPrototypeMap,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  ArrayPrototypeSlice,
  Promise,
  PromisePrototypeThen,
@@ -477,7 +478,7 @@ function merge(...args) {
 
 // Drain ready queue synchronously
 while(ready.length>0){
-constitem=ready.shift();
+constitem=ArrayPrototypeShift(ready);
 if(item?.error){
 throwitem.error;
 }

lib/internal/streams/iter/from.js

@@ -31,8 +31,8 @@ const {
 
 const{
  isAnyArrayBuffer,
- isDataView,
  isPromise,
+ isTypedArray,
  isUint8Array,
 }=require('internal/util/types');
 
@@ -106,17 +106,21 @@ function primitiveToUint8Array(chunk) {
 returnchunk;
 }
 // Other ArrayBufferView types (Int8Array, DataView, etc.)
-if(isDataView(chunk)){
+returnarrayBufferViewToUint8Array(chunk);
+}
+
+functionarrayBufferViewToUint8Array(chunk){
+if(isTypedArray(chunk)){
 returnnewUint8Array(
-DataViewPrototypeGetBuffer(chunk),
-DataViewPrototypeGetByteOffset(chunk),
-DataViewPrototypeGetByteLength(chunk),
+TypedArrayPrototypeGetBuffer(chunk),
+TypedArrayPrototypeGetByteOffset(chunk),
+TypedArrayPrototypeGetByteLength(chunk),
 );
 }
 returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(chunk),
-TypedArrayPrototypeGetByteOffset(chunk),
-TypedArrayPrototypeGetByteLength(chunk),
+DataViewPrototypeGetBuffer(chunk),
+DataViewPrototypeGetByteOffset(chunk),
+DataViewPrototypeGetByteLength(chunk),
 );
 }
 
@@ -580,6 +584,7 @@ function from(input) {
 // =============================================================================
 
 module.exports={
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
  isAsyncIterable,

lib/internal/streams/iter/pull.js

@@ -33,9 +33,9 @@ const {
 const{ AbortController }=require('internal/abort_controller');
 
 const{
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
- primitiveToUint8Array,
  isSyncIterable,
  isAsyncIterable,
  isUint8ArrayBatch,
@@ -136,7 +136,7 @@ function* flattenTransformYieldSync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Must be Iterable<TransformYield>
@@ -170,7 +170,7 @@ async function* flattenTransformYieldAsync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Check for async iterable first
@@ -180,10 +180,10 @@ async function* flattenTransformYieldAsync(value) {
 }
 return;
 }
-// Must be sync Iterable<TransformYield>
+// Must be sync Iterable<TransformYield>, no nested async iterables
 if(isSyncIterable(value)){
 for(constitemofvalue){
-yield*flattenTransformYieldAsync(item);
+yield*flattenTransformYieldSync(item);
 }
 return;
 }
@@ -218,7 +218,7 @@ function* processTransformResultSync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -278,7 +278,7 @@ async function* processTransformResultAsync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -313,7 +313,9 @@ async function* processTransformResultAsync(result) {
 ArrayPrototypePush(batch,item);
 continue;
 }
-forawait(constchunkofflattenTransformYieldAsync(item)){
+// Note: This iteration is synchronous, since async iterables
+// may not be nested within sync iterables.
+for(constchunkofflattenTransformYieldSync(item)){
 ArrayPrototypePush(batch,chunk);
 }
 }
@@ -366,7 +368,7 @@ function* applyFusedStatelessSyncTransforms(source, run) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultSync(current);
 }
@@ -428,7 +430,7 @@ function* createSyncPipeline(source, transforms) {
 }
 current=applyStatefulSyncTransform(current,transform.transform);
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 if(statelessRun.length>0){
@@ -490,7 +492,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultAsync(current);
 }
@@ -531,9 +533,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
  * @yields {Uint8Array[]}
  */
 asyncfunction*withFlushAsync(source){
-forawait(constbatchofsource){
-yieldbatch;
-}
+yield*source;
 yieldnull;
 }
 
@@ -647,7 +647,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
 current,transform.transform,opts);
 }
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 // Flush remaining stateless run

lib/internal/streams/iter/utils.js

@@ -12,6 +12,7 @@ const {
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
+ TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
@@ -24,8 +25,6 @@ const {
 }=require('internal/errors');
 const{ isError }=require('internal/util');
 
-const{ Buffer }=require('buffer');
-
 const{ isSharedArrayBuffer, isUint8Array }=require('internal/util/types');
 
 const{ validateOneOf }=require('internal/validators');
@@ -127,26 +126,36 @@ function concatBytes(chunks) {
 if(chunks.length===0){
 returnnewUint8Array(0);
 }
-// Single chunk: return directly if it covers the entire backing buffer
+// Single chunk: return directly if it covers the entire backing buffer,
+// otherwise return a copy
 if(chunks.length===1){
 constchunk=chunks[0];
-constbuf=TypedArrayPrototypeGetBuffer(chunk);
-// SharedArrayBuffer is not available in primordials, so use
-// direct property access for its byteLength.
-constbufByteLength=isSharedArrayBuffer(buf) ?
-buf.byteLength :
-ArrayBufferPrototypeGetByteLength(buf);
-if(TypedArrayPrototypeGetByteOffset(chunk)===0&&
-TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
-returnchunk;
+// If non-zero offset, skip the remaining buffer checks.
+if(TypedArrayPrototypeGetByteOffset(chunk)===0){
+constbuf=TypedArrayPrototypeGetBuffer(chunk);
+// SharedArrayBuffer is not available in primordials, so use
+// direct property access for its byteLength.
+constbufByteLength=isSharedArrayBuffer(buf) ?
+buf.byteLength :
+ArrayBufferPrototypeGetByteLength(buf);
+if(TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
+returnchunk;
+}
 }
+returnnewUint8Array(chunk);
+}
+// Multiple chunks: concatenate
+lettotalByteLength=0;
+for(leti=0;i<chunks.length;i++){
+totalByteLength+=TypedArrayPrototypeGetByteLength(chunks[i]);
+}
+constconcatenated=newUint8Array(totalByteLength);
+letoffset=0;
+for(leti=0;i<chunks.length;i++){
+TypedArrayPrototypeSet(concatenated,chunks[i],offset);
+offset+=TypedArrayPrototypeGetByteLength(chunks[i]);
 }
-// Multiple chunks or shared buffer: concatenate
-constbuf=Buffer.concat(chunks);
-returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(buf),
-TypedArrayPrototypeGetByteOffset(buf),
-TypedArrayPrototypeGetByteLength(buf));
+returnconcatenated;
 }
 
 /**

lib/internal/streams/iter/classic.js

@@ -13,6 +13,7 @@
 
 const{
  ArrayIsArray,
+ ArrayPrototypePush,
  MathMax,
  NumberMAX_SAFE_INTEGER,
  Promise,
@@ -107,14 +108,14 @@ async function normalizeBatch(raw) {
 for(leti=0;i<raw.length;i++){
 constvalue=raw[i];
 if(isUint8Array(value)){
-batch.push(value);
+ArrayPrototypePush(batch,value);
 }else{
 // normalizeAsyncValue may await for async protocols (e.g.
 // toAsyncStreamable on yielded objects). Stream events during
 // the suspension are queued, not lost -- errors will surface
 // on the next loop iteration after this yield completes.
 forawait(constnormalizedofnormalizeAsyncValue(value)){
-batch.push(normalized);
+ArrayPrototypePush(batch,normalized);
 }
 }
 }
@@ -163,7 +164,7 @@ async function* createBatchedAsyncIterator(stream, normalize) {
 stream._readableState?.length>0){
 constc=stream.read();
 if(c===null)break;
-batch.push(c);
+ArrayPrototypePush(batch,c);
 }
 if(normalize!==null){
 constresult=awaitnormalize(batch);
@@ -495,7 +496,7 @@ function fromWritable(writable, options = kNullPrototype) {
 
 functionwaitForDrain(){
 const{ promise, resolve, reject }=PromiseWithResolvers();
-waiters.push({__proto__: null, resolve, reject });
+ArrayPrototypePush(waiters,{__proto__: null, resolve, reject });
 installListeners();
 returnpromise;
 }
@@ -686,7 +687,7 @@ function fromWritable(writable, options = kNullPrototype) {
 returnPromiseResolve(true);
 }
 const{ promise, resolve }=PromiseWithResolvers();
-waiters.push({
+ArrayPrototypePush(waiters,{
 __proto__: null,
 resolve(){resolve(true);},
 reject(){resolve(false);},

lib/internal/streams/iter/consumers.js

@@ -12,6 +12,7 @@ const {
  ArrayBufferPrototypeSlice,
  ArrayPrototypeMap,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  ArrayPrototypeSlice,
  Promise,
  PromisePrototypeThen,
@@ -477,7 +478,7 @@ function merge(...args) {
 
 // Drain ready queue synchronously
 while(ready.length>0){
-constitem=ready.shift();
+constitem=ArrayPrototypeShift(ready);
 if(item?.error){
 throwitem.error;
 }

lib/internal/streams/iter/from.js

@@ -31,8 +31,8 @@ const {
 
 const{
  isAnyArrayBuffer,
- isDataView,
  isPromise,
+ isTypedArray,
  isUint8Array,
 }=require('internal/util/types');
 
@@ -106,17 +106,21 @@ function primitiveToUint8Array(chunk) {
 returnchunk;
 }
 // Other ArrayBufferView types (Int8Array, DataView, etc.)
-if(isDataView(chunk)){
+returnarrayBufferViewToUint8Array(chunk);
+}
+
+functionarrayBufferViewToUint8Array(chunk){
+if(isTypedArray(chunk)){
 returnnewUint8Array(
-DataViewPrototypeGetBuffer(chunk),
-DataViewPrototypeGetByteOffset(chunk),
-DataViewPrototypeGetByteLength(chunk),
+TypedArrayPrototypeGetBuffer(chunk),
+TypedArrayPrototypeGetByteOffset(chunk),
+TypedArrayPrototypeGetByteLength(chunk),
 );
 }
 returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(chunk),
-TypedArrayPrototypeGetByteOffset(chunk),
-TypedArrayPrototypeGetByteLength(chunk),
+DataViewPrototypeGetBuffer(chunk),
+DataViewPrototypeGetByteOffset(chunk),
+DataViewPrototypeGetByteLength(chunk),
 );
 }
 
@@ -580,6 +584,7 @@ function from(input) {
 // =============================================================================
 
 module.exports={
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
  isAsyncIterable,

lib/internal/streams/iter/pull.js

@@ -33,9 +33,9 @@ const {
 const{ AbortController }=require('internal/abort_controller');
 
 const{
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
- primitiveToUint8Array,
  isSyncIterable,
  isAsyncIterable,
  isUint8ArrayBatch,
@@ -136,7 +136,7 @@ function* flattenTransformYieldSync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Must be Iterable<TransformYield>
@@ -170,7 +170,7 @@ async function* flattenTransformYieldAsync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Check for async iterable first
@@ -180,10 +180,10 @@ async function* flattenTransformYieldAsync(value) {
 }
 return;
 }
-// Must be sync Iterable<TransformYield>
+// Must be sync Iterable<TransformYield>, no nested async iterables
 if(isSyncIterable(value)){
 for(constitemofvalue){
-yield*flattenTransformYieldAsync(item);
+yield*flattenTransformYieldSync(item);
 }
 return;
 }
@@ -218,7 +218,7 @@ function* processTransformResultSync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -278,7 +278,7 @@ async function* processTransformResultAsync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -313,7 +313,9 @@ async function* processTransformResultAsync(result) {
 ArrayPrototypePush(batch,item);
 continue;
 }
-forawait(constchunkofflattenTransformYieldAsync(item)){
+// Note: This iteration is synchronous, since async iterables
+// may not be nested within sync iterables.
+for(constchunkofflattenTransformYieldSync(item)){
 ArrayPrototypePush(batch,chunk);
 }
 }
@@ -366,7 +368,7 @@ function* applyFusedStatelessSyncTransforms(source, run) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultSync(current);
 }
@@ -428,7 +430,7 @@ function* createSyncPipeline(source, transforms) {
 }
 current=applyStatefulSyncTransform(current,transform.transform);
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 if(statelessRun.length>0){
@@ -490,7 +492,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultAsync(current);
 }
@@ -531,9 +533,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
  * @yields {Uint8Array[]}
  */
 asyncfunction*withFlushAsync(source){
-forawait(constbatchofsource){
-yieldbatch;
-}
+yield*source;
 yieldnull;
 }
 
@@ -647,7 +647,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
 current,transform.transform,opts);
 }
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 // Flush remaining stateless run

lib/internal/streams/iter/utils.js

@@ -12,6 +12,7 @@ const {
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
+ TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
@@ -24,8 +25,6 @@ const {
 }=require('internal/errors');
 const{ isError }=require('internal/util');
 
-const{ Buffer }=require('buffer');
-
 const{ isSharedArrayBuffer, isUint8Array }=require('internal/util/types');
 
 const{ validateOneOf }=require('internal/validators');
@@ -127,26 +126,36 @@ function concatBytes(chunks) {
 if(chunks.length===0){
 returnnewUint8Array(0);
 }
-// Single chunk: return directly if it covers the entire backing buffer
+// Single chunk: return directly if it covers the entire backing buffer,
+// otherwise return a copy
 if(chunks.length===1){
 constchunk=chunks[0];
-constbuf=TypedArrayPrototypeGetBuffer(chunk);
-// SharedArrayBuffer is not available in primordials, so use
-// direct property access for its byteLength.
-constbufByteLength=isSharedArrayBuffer(buf) ?
-buf.byteLength :
-ArrayBufferPrototypeGetByteLength(buf);
-if(TypedArrayPrototypeGetByteOffset(chunk)===0&&
-TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
-returnchunk;
+// If non-zero offset, skip the remaining buffer checks.
+if(TypedArrayPrototypeGetByteOffset(chunk)===0){
+constbuf=TypedArrayPrototypeGetBuffer(chunk);
+// SharedArrayBuffer is not available in primordials, so use
+// direct property access for its byteLength.
+constbufByteLength=isSharedArrayBuffer(buf) ?
+buf.byteLength :
+ArrayBufferPrototypeGetByteLength(buf);
+if(TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
+returnchunk;
+}
 }
+returnnewUint8Array(chunk);
+}
+// Multiple chunks: concatenate
+lettotalByteLength=0;
+for(leti=0;i<chunks.length;i++){
+totalByteLength+=TypedArrayPrototypeGetByteLength(chunks[i]);
+}
+constconcatenated=newUint8Array(totalByteLength);
+letoffset=0;
+for(leti=0;i<chunks.length;i++){
+TypedArrayPrototypeSet(concatenated,chunks[i],offset);
+offset+=TypedArrayPrototypeGetByteLength(chunks[i]);
 }
-// Multiple chunks or shared buffer: concatenate
-constbuf=Buffer.concat(chunks);
-returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(buf),
-TypedArrayPrototypeGetByteOffset(buf),
-TypedArrayPrototypeGetByteLength(buf));
+returnconcatenated;
 }
 
 /**

lib/internal/streams/iter/classic.js

@@ -13,6 +13,7 @@
 
 const{
  ArrayIsArray,
+ ArrayPrototypePush,
  MathMax,
  NumberMAX_SAFE_INTEGER,
  Promise,
@@ -107,14 +108,14 @@ async function normalizeBatch(raw) {
 for(leti=0;i<raw.length;i++){
 constvalue=raw[i];
 if(isUint8Array(value)){
-batch.push(value);
+ArrayPrototypePush(batch,value);
 }else{
 // normalizeAsyncValue may await for async protocols (e.g.
 // toAsyncStreamable on yielded objects). Stream events during
 // the suspension are queued, not lost -- errors will surface
 // on the next loop iteration after this yield completes.
 forawait(constnormalizedofnormalizeAsyncValue(value)){
-batch.push(normalized);
+ArrayPrototypePush(batch,normalized);
 }
 }
 }
@@ -163,7 +164,7 @@ async function* createBatchedAsyncIterator(stream, normalize) {
 stream._readableState?.length>0){
 constc=stream.read();
 if(c===null)break;
-batch.push(c);
+ArrayPrototypePush(batch,c);
 }
 if(normalize!==null){
 constresult=awaitnormalize(batch);
@@ -495,7 +496,7 @@ function fromWritable(writable, options = kNullPrototype) {
 
 functionwaitForDrain(){
 const{ promise, resolve, reject }=PromiseWithResolvers();
-waiters.push({__proto__: null, resolve, reject });
+ArrayPrototypePush(waiters,{__proto__: null, resolve, reject });
 installListeners();
 returnpromise;
 }
@@ -686,7 +687,7 @@ function fromWritable(writable, options = kNullPrototype) {
 returnPromiseResolve(true);
 }
 const{ promise, resolve }=PromiseWithResolvers();
-waiters.push({
+ArrayPrototypePush(waiters,{
 __proto__: null,
 resolve(){resolve(true);},
 reject(){resolve(false);},

lib/internal/streams/iter/consumers.js

@@ -12,6 +12,7 @@ const {
  ArrayBufferPrototypeSlice,
  ArrayPrototypeMap,
  ArrayPrototypePush,
+ ArrayPrototypeShift,
  ArrayPrototypeSlice,
  Promise,
  PromisePrototypeThen,
@@ -477,7 +478,7 @@ function merge(...args) {
 
 // Drain ready queue synchronously
 while(ready.length>0){
-constitem=ready.shift();
+constitem=ArrayPrototypeShift(ready);
 if(item?.error){
 throwitem.error;
 }

lib/internal/streams/iter/from.js

@@ -31,8 +31,8 @@ const {
 
 const{
  isAnyArrayBuffer,
- isDataView,
  isPromise,
+ isTypedArray,
  isUint8Array,
 }=require('internal/util/types');
 
@@ -106,17 +106,21 @@ function primitiveToUint8Array(chunk) {
 returnchunk;
 }
 // Other ArrayBufferView types (Int8Array, DataView, etc.)
-if(isDataView(chunk)){
+returnarrayBufferViewToUint8Array(chunk);
+}
+
+functionarrayBufferViewToUint8Array(chunk){
+if(isTypedArray(chunk)){
 returnnewUint8Array(
-DataViewPrototypeGetBuffer(chunk),
-DataViewPrototypeGetByteOffset(chunk),
-DataViewPrototypeGetByteLength(chunk),
+TypedArrayPrototypeGetBuffer(chunk),
+TypedArrayPrototypeGetByteOffset(chunk),
+TypedArrayPrototypeGetByteLength(chunk),
 );
 }
 returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(chunk),
-TypedArrayPrototypeGetByteOffset(chunk),
-TypedArrayPrototypeGetByteLength(chunk),
+DataViewPrototypeGetBuffer(chunk),
+DataViewPrototypeGetByteOffset(chunk),
+DataViewPrototypeGetByteLength(chunk),
 );
 }
 
@@ -580,6 +584,7 @@ function from(input) {
 // =============================================================================
 
 module.exports={
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
  isAsyncIterable,

lib/internal/streams/iter/pull.js

@@ -33,9 +33,9 @@ const {
 const{ AbortController }=require('internal/abort_controller');
 
 const{
+ arrayBufferViewToUint8Array,
  from,
  fromSync,
- primitiveToUint8Array,
  isSyncIterable,
  isAsyncIterable,
  isUint8ArrayBatch,
@@ -136,7 +136,7 @@ function* flattenTransformYieldSync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Must be Iterable<TransformYield>
@@ -170,7 +170,7 @@ async function* flattenTransformYieldAsync(value) {
 return;
 }
 if(ArrayBufferIsView(value)){
-yieldprimitiveToUint8Array(value);
+yieldarrayBufferViewToUint8Array(value);
 return;
 }
 // Check for async iterable first
@@ -180,10 +180,10 @@ async function* flattenTransformYieldAsync(value) {
 }
 return;
 }
-// Must be sync Iterable<TransformYield>
+// Must be sync Iterable<TransformYield>, no nested async iterables
 if(isSyncIterable(value)){
 for(constitemofvalue){
-yield*flattenTransformYieldAsync(item);
+yield*flattenTransformYieldSync(item);
 }
 return;
 }
@@ -218,7 +218,7 @@ function* processTransformResultSync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -278,7 +278,7 @@ async function* processTransformResultAsync(result) {
 return;
 }
 if(ArrayBufferIsView(result)){
-yield[primitiveToUint8Array(result)];
+yield[arrayBufferViewToUint8Array(result)];
 return;
 }
 // Uint8Array[] batch
@@ -313,7 +313,9 @@ async function* processTransformResultAsync(result) {
 ArrayPrototypePush(batch,item);
 continue;
 }
-forawait(constchunkofflattenTransformYieldAsync(item)){
+// Note: This iteration is synchronous, since async iterables
+// may not be nested within sync iterables.
+for(constchunkofflattenTransformYieldSync(item)){
 ArrayPrototypePush(batch,chunk);
 }
 }
@@ -366,7 +368,7 @@ function* applyFusedStatelessSyncTransforms(source, run) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultSync(current);
 }
@@ -428,7 +430,7 @@ function* createSyncPipeline(source, transforms) {
 }
 current=applyStatefulSyncTransform(current,transform.transform);
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 if(statelessRun.length>0){
@@ -490,7 +492,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
 }elseif(isAnyArrayBuffer(current)){
 yield[newUint8Array(current)];
 }elseif(ArrayBufferIsView(current)){
-yield[primitiveToUint8Array(current)];
+yield[arrayBufferViewToUint8Array(current)];
 }else{
 yield*processTransformResultAsync(current);
 }
@@ -531,9 +533,7 @@ async function* applyFusedStatelessAsyncTransforms(source, run, signal) {
  * @yields {Uint8Array[]}
  */
 asyncfunction*withFlushAsync(source){
-forawait(constbatchofsource){
-yieldbatch;
-}
+yield*source;
 yieldnull;
 }
 
@@ -647,7 +647,7 @@ async function* createAsyncPipeline(source, transforms, signal) {
 current,transform.transform,opts);
 }
 }else{
-statelessRun.push(transform);
+ArrayPrototypePush(statelessRun,transform);
 }
 }
 // Flush remaining stateless run

lib/internal/streams/iter/utils.js

@@ -12,6 +12,7 @@ const {
  TypedArrayPrototypeGetBuffer,
  TypedArrayPrototypeGetByteLength,
  TypedArrayPrototypeGetByteOffset,
+ TypedArrayPrototypeSet,
  Uint8Array,
 }=primordials;
 
@@ -24,8 +25,6 @@ const {
 }=require('internal/errors');
 const{ isError }=require('internal/util');
 
-const{ Buffer }=require('buffer');
-
 const{ isSharedArrayBuffer, isUint8Array }=require('internal/util/types');
 
 const{ validateOneOf }=require('internal/validators');
@@ -127,26 +126,36 @@ function concatBytes(chunks) {
 if(chunks.length===0){
 returnnewUint8Array(0);
 }
-// Single chunk: return directly if it covers the entire backing buffer
+// Single chunk: return directly if it covers the entire backing buffer,
+// otherwise return a copy
 if(chunks.length===1){
 constchunk=chunks[0];
-constbuf=TypedArrayPrototypeGetBuffer(chunk);
-// SharedArrayBuffer is not available in primordials, so use
-// direct property access for its byteLength.
-constbufByteLength=isSharedArrayBuffer(buf) ?
-buf.byteLength :
-ArrayBufferPrototypeGetByteLength(buf);
-if(TypedArrayPrototypeGetByteOffset(chunk)===0&&
-TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
-returnchunk;
+// If non-zero offset, skip the remaining buffer checks.
+if(TypedArrayPrototypeGetByteOffset(chunk)===0){
+constbuf=TypedArrayPrototypeGetBuffer(chunk);
+// SharedArrayBuffer is not available in primordials, so use
+// direct property access for its byteLength.
+constbufByteLength=isSharedArrayBuffer(buf) ?
+buf.byteLength :
+ArrayBufferPrototypeGetByteLength(buf);
+if(TypedArrayPrototypeGetByteLength(chunk)===bufByteLength){
+returnchunk;
+}
 }
+returnnewUint8Array(chunk);
+}
+// Multiple chunks: concatenate
+lettotalByteLength=0;
+for(leti=0;i<chunks.length;i++){
+totalByteLength+=TypedArrayPrototypeGetByteLength(chunks[i]);
+}
+constconcatenated=newUint8Array(totalByteLength);
+letoffset=0;
+for(leti=0;i<chunks.length;i++){
+TypedArrayPrototypeSet(concatenated,chunks[i],offset);
+offset+=TypedArrayPrototypeGetByteLength(chunks[i]);
 }
-// Multiple chunks or shared buffer: concatenate
-constbuf=Buffer.concat(chunks);
-returnnewUint8Array(
-TypedArrayPrototypeGetBuffer(buf),
-TypedArrayPrototypeGetByteOffset(buf),
-TypedArrayPrototypeGetByteLength(buf));
+returnconcatenated;
 }
 
 /**