Skip to content

Node.js streams: Apply comments from #92252 #92530

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
timneutkens wants to merge 1 commit into
base: canary
Choose a base branch
from
Draft

Node.js streams: Apply comments from #92252 #92530

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion packages/next/src/server/app-render/debug-channel-server.web.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,5 +54,5 @@ export function createWebDebugChannel(): DebugChannelPair {
* which expects debugChannel to be a Node.js stream with a .write() method.
*/
export function createNodeDebugChannel(): DebugChannelPair {
throw new Error('not implemented')
throw new Error('Not implemented')
}
60 changes: 8 additions & 52 deletions packages/next/src/server/app-render/stream-ops.node.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,6 @@ import {
streamToString as webStreamToString,
createDocumentClosingStream as webCreateDocumentClosingStream,
createRuntimePrefetchTransformStream,
CLOSE_TAG,
} from '../stream-utils/node-web-streams-helper'
import { indexOfUint8Array } from '../stream-utils/uint8array-helpers'
import { ENCODED_TAGS } from '../stream-utils/encoded-tags'
Expand Down Expand Up @@ -210,9 +209,7 @@ function createFlightDataInjectionTransform(
const nodeTransform = new Transform({
transform(chunk, _encoding, callback) {
this.push(chunk)
if (delayDataUntilFirstHtmlChunk) {
startOrContinuePulling(this)
}
startOrContinuePulling(this)
callback()
},
flush(callback) {
Expand Down Expand Up @@ -263,7 +260,7 @@ function createHeadInsertionTransform(
if (index !== -1) {
if (insertion) {
const encodedInsertion = Buffer.from(insertion)
const merged = Buffer.allocUnsafe(
const merged = Buffer.alloc(
chunk.length + encodedInsertion.length
)
merged.set(chunk.slice(0, index))
Expand Down Expand Up @@ -336,6 +333,8 @@ function createMetadataTransform(
}

iconMarkLength = ENCODED_TAGS.META.ICON_MARK.length
// 47 is `/` – handle self-closing `<nxt-icon/>` (length +2 for `/>`)
// vs non-self-closing `<nxt-icon>` (length +1 for `>`)
if (chunk[iconMarkIndex + iconMarkLength] === 47) {
iconMarkLength += 2
} else {
Expand All @@ -345,7 +344,7 @@ function createMetadataTransform(
if (chunkIndex === 0) {
closedHeadIndex = indexOfUint8Array(chunk, ENCODED_TAGS.CLOSED.HEAD)
if (iconMarkIndex < closedHeadIndex) {
const replaced = Buffer.allocUnsafe(chunk.length - iconMarkLength)
const replaced = Buffer.alloc(chunk.length - iconMarkLength)
replaced.set(chunk.subarray(0, iconMarkIndex))
replaced.set(
chunk.subarray(iconMarkIndex + iconMarkLength),
Expand All @@ -356,7 +355,7 @@ function createMetadataTransform(
const insertion = await insert()
const encodedInsertion = Buffer.from(insertion)
const insertionLength = encodedInsertion.length
const replaced = Buffer.allocUnsafe(
const replaced = Buffer.alloc(
chunk.length - iconMarkLength + insertionLength
)
replaced.set(chunk.subarray(0, iconMarkIndex))
Expand All @@ -372,7 +371,7 @@ function createMetadataTransform(
const insertion = await insert()
const encodedInsertion = Buffer.from(insertion)
const insertionLength = encodedInsertion.length
const replaced = Buffer.allocUnsafe(
const replaced = Buffer.alloc(
chunk.length - iconMarkLength + insertionLength
)
replaced.set(chunk.subarray(0, iconMarkIndex))
Expand All @@ -393,36 +392,6 @@ function createMetadataTransform(
})
}

// ---------------------------------------------------------------------------
// Deferred suffix – Node.js Transform that appends a suffix string after the
// first HTML chunk, deferring via queueMicrotask so the chunk flushes first.
// ---------------------------------------------------------------------------

function createDeferredSuffixTransform(suffix: string): Transform {
let flushed = false
const encodedSuffix = Buffer.from(suffix)

return new Transform({
transform(chunk, _encoding, callback) {
this.push(chunk)

if (!flushed) {
flushed = true
queueMicrotask(() => {
this.push(encodedSuffix)
})
}
callback()
},
flush(callback) {
if (!flushed) {
this.push(encodedSuffix)
}
callback()
},
})
}

// ---------------------------------------------------------------------------
// Move suffix – Node.js Transform that strips </body></html> from its
// original position and re-appends it at the very end of the stream, so any
Expand Down Expand Up @@ -495,9 +464,7 @@ function createHtmlDataDplIdTransform(dplId: string): Transform {

const insertionPoint = htmlTagIndex + ENCODED_TAGS.OPENING.HTML.length
const encodedAttribute = Buffer.from(` data-dpl-id="${dplId}"`)
const modified = Buffer.allocUnsafe(
chunk.length + encodedAttribute.length
)
const modified = Buffer.alloc(chunk.length + encodedAttribute.length)

modified.set(chunk.subarray(0, insertionPoint))
modified.set(encodedAttribute, insertionPoint)
Expand Down Expand Up @@ -693,7 +660,6 @@ export async function resumeAndAbort(
export async function continueFizzStream(
renderStream: AnyStream,
{
suffix,
inlinedDataStream,
isStaticGeneration,
allReady,
Expand All @@ -703,9 +669,6 @@ export async function continueFizzStream(
validateRootLayout,
}: import('./stream-ops.web').ContinueFizzStreamOptions
): Promise<Readable> {
// Suffix itself might contain close tags at the end, so we need to split it.
const suffixUnclosed = suffix ? suffix.split(CLOSE_TAG, 1)[0] : null

if (isStaticGeneration) {
if (allReady) {
await allReady
Expand Down Expand Up @@ -736,13 +699,6 @@ export async function continueFizzStream(
source.pipe(metadata)
source = metadata

// Insert suffix content
if (suffixUnclosed != null && suffixUnclosed.length > 0) {
const deferredSuffix = createDeferredSuffixTransform(suffixUnclosed)
source.pipe(deferredSuffix)
source = deferredSuffix
}

// Flight data injection – interleaves RSC data chunks with the HTML stream
if (inlinedDataStream) {
const flightInjection = createFlightDataInjectionTransform(
Expand Down
7 changes: 3 additions & 4 deletions packages/next/src/server/app-render/stream-ops.web.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,6 @@ export type ContinueFizzStreamOptions = ContinueStreamSharedOptions & {
isStaticGeneration: boolean
allReady?: Promise<void>
validateRootLayout?: boolean
suffix?: string
}

export type ContinueStaticPrerenderOptions = ContinueStreamSharedOptions & {
Expand Down Expand Up @@ -158,7 +157,7 @@ export function continueDynamicHTMLResumeNode(
_renderStream: AnyStream,
_opts: ContinueDynamicHTMLResumeOptions
): Promise<AnyStream> {
throw new Error('not implemented')
throw new Error('Not implemented')
}

export async function streamToBuffer(stream: AnyStream): Promise<Buffer> {
Expand Down Expand Up @@ -200,7 +199,7 @@ export function createNodeInlinedDataStream(
_nonce: string | undefined,
_formState: unknown | null
): AnyStream {
throw new Error('not implemented')
throw new Error('Not implemented')
}

export function createPendingStream(): AnyStream {
Expand Down Expand Up @@ -235,7 +234,7 @@ export function renderToNodeFlightStream(
_clientModules: FlightClientModules,
_opts: FlightRenderOptions
): AnyStream {
throw new Error('not implemented')
throw new Error('Not implemented')
}

export function renderToWebFlightStream(
Expand Down
60 changes: 0 additions & 60 deletions packages/next/src/server/stream-utils/node-web-streams-helper.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -666,53 +666,6 @@ export function createInstantTestScriptInsertionTransformStream(
})
}

// Suffix after main body content - scripts before </body>,
// but wait for the major chunks to be enqueued.
export function createDeferredSuffixStream(
suffix: string
): TransformStream<Uint8Array, Uint8Array> {
let flushed = false
let pending: DetachedPromise<void> | undefined

const flush = (controller: TransformStreamDefaultController) => {
const detached = new DetachedPromise<void>()
pending = detached

scheduleImmediate(() => {
try {
controller.enqueue(encoder.encode(suffix))
} catch {
// If an error occurs while enqueuing it can't be due to this
// transformers fault. It's likely due to the controller being
// errored due to the stream being cancelled.
} finally {
pending = undefined
detached.resolve()
}
})
}

return new TransformStream({
transform(chunk, controller) {
controller.enqueue(chunk)

// If we've already flushed, we're done.
if (flushed) return

// Schedule the flush to happen.
flushed = true
flush(controller)
},
flush(controller) {
if (pending) return pending.promise
if (flushed) return

// Flush now.
controller.enqueue(encoder.encode(suffix))
},
})
}

export function createFlightDataInjectionTransformStream(
stream: ReadableStream<Uint8Array>,
delayDataUntilFirstHtmlChunk: boolean
Expand Down Expand Up @@ -996,16 +949,11 @@ export type ContinueStreamOptions = {
getServerInsertedHTML: () => Promise<string>
getServerInsertedMetadata: () => Promise<string>
validateRootLayout?: boolean
/**
* Suffix to inject after the buffered data, but before the close tags.
*/
suffix?: string | undefined
}

export async function continueFizzStream(
renderStream: ReactDOMServerReadableStream,
{
suffix,
inlinedDataStream,
isStaticGeneration,
deploymentId,
Expand All @@ -1014,9 +962,6 @@ export async function continueFizzStream(
validateRootLayout,
}: ContinueStreamOptions
): Promise<ReadableStream<Uint8Array>> {
// Suffix itself might contain close tags at the end, so we need to split it.
const suffixUnclosed = suffix ? suffix.split(CLOSE_TAG, 1)[0] : null

if (isStaticGeneration) {
// If we're generating static HTML we need to wait for it to resolve before continuing.
await renderStream.allReady
Expand All @@ -1036,11 +981,6 @@ export async function continueFizzStream(
// Transform metadata
createMetadataTransformStream(getServerInsertedMetadata),

// Insert suffix content
suffixUnclosed != null && suffixUnclosed.length > 0
? createDeferredSuffixStream(suffixUnclosed)
: null,

// Insert the inlined data (Flight data, form state, etc.) stream into the HTML
inlinedDataStream
? createFlightDataInjectionTransformStream(inlinedDataStream, true)
Expand Down
Loading