Там где раньше мы отдавали весь HTML, я пишу в стрим первым чанком всю разметку до <APP />:

<head>
<META />
<LINKS />
<SCRIPTS />
</head>
<body>
...

Для стриминга, на каждый запрос создаю и сохраняю в DI отдельный Duplex стрим, который и отдаю в ответ через Fastify reply.send(stream). Это позволяет легко переиспользовать его в разных трамвайных модулях.

Во время отдачи первого чанка с <head>, renderToPipeableStream уже был запущен (и для сохранения существующего жизненного цикла запроса и для ускорения ответа), и тут нам важно избежать гонки, реакт должен отдавать разметку в стрим строго на слот <APP />, после отдачи первого чанка.

Еще раз подчеркну, что renderToPipeableStream приходится запускать раньше чем мы можем использовать данные из него, то есть один из основных челленджей в задаче идет из необходимости поддержать текущую архитектуру с минимумом доработок.

Для решения проблемы завел сущность ResponseTaskManager с такими возможностями:

- метод для добавления асинхронной задачи (push)
- метод для запуска и ожидания всех задач (process)

Пока максимально простой, в будущем скорее всего понадобится добавлять различные приоритеты для задач, если менять текущую архитектуру со схемой HTML страницы.

Этот менеджер задач позволяет в любое время запушить в очередь таски с записью разметки в стрим, но выполнить их строго в определенный момент времени - в нашем случае после того как отдали клиенту открывающий тег <body> в первом чанке.

Итого, у нас примерно так выглядит ответ клиенту, после того как сгенерировали начальный HTML по схеме:

const [headAndBodyStart, bodyEnd] = html.split('<APP />')

// передаем стрим в ответ клиенту
reply.send(stream)

// пишем первый чанк с head и открывающим body
stream.push(headAndBodyStart)

// тут выполняются задачи, поставленные во время работы renderToPipeableStream
await taskManager.process()

// пишем закрывающий body
stream.push(bodyEnd)

// завершаем ответ
stream.push(null)

Для того что бы данные из renderToPipeableStream писать в стрим по требованию а не по факту их получения:

- Создаем кастомный Writable стрим, который и передаем в метод pipe в коллбэке onShellReady (когда готова первая часть HTML но все отложенные Suspense компоненты вернули fallback)
- Этот стрим на каждый write от реакта добавляет задачу в taskManager с записью HTML в поток
- Также добавляем еще одну задачу в taskManager которая зарезолвится после события finish у нашего Writable стрима (когда все отложенные Suspense компоненты зарезолвлены и реакт выполнил всю работу)

Пример кода без лишних подробностей:

// этот стрим откладывает запись разметки в стрим ответа
class HtmlWritable extends Writable {
_write(chunk, encoding, callback) {
// stream - это именно поток ответа клиенту с предыдущего шага
taskManager.push(() => {
stream.push(chunk)
})

callback()
}
}

// реализацию Deferred добавлю позже
const allReadyDeferred = Deferred();

// задача на ожидание окончания рендера
taskManager.push(() => {
return allReadyDeferred.promise;
})

// событие сработает когда pipe завершит работу
htmlWritable.on('finish', () => {
allReadyDeferred.resolve()
})

const { pipe } = renderToPipeableStream(<App />, {
onShellReady() {
// pipe готов передавать данные
pipe(htmlWritable)
}
})