sendWebStream() — fastify Function Reference

Architecture documentation for the sendWebStream() function in reply.js from the fastify codebase.

Function javascript CoreKernel InstanceFactory calls 4 called by 1

Entity Profile

CoreKernel→ InstanceFactory→ sendWebStream() — fastify Function Reference

Dependency Diagram

graph TD
  e86de8a4_fcf2_726c_6613_1273524e94e2["sendWebStream()"]
  4bcd71dc_1ec2_5fe8_b8ff_4a371e392925["reply.js"]
  e86de8a4_fcf2_726c_6613_1273524e94e2 -->|defined in| 4bcd71dc_1ec2_5fe8_b8ff_4a371e392925
  8460cb7c_51e5_baba_0f16_9ea8d190d139["onSendEnd()"]
  8460cb7c_51e5_baba_0f16_9ea8d190d139 -->|calls| e86de8a4_fcf2_726c_6613_1273524e94e2
  18ed21df_f31c_a1a7_f5d2_fd033a6e96bc["logStreamError()"]
  e86de8a4_fcf2_726c_6613_1273524e94e2 -->|calls| 18ed21df_f31c_a1a7_f5d2_fd033a6e96bc
  944c1a61_ec49_25f7_bdf7_a5b89cd52706["sendTrailer()"]
  e86de8a4_fcf2_726c_6613_1273524e94e2 -->|calls| 944c1a61_ec49_25f7_bdf7_a5b89cd52706
  f26bdb28_b830_3aec_800b_087055150c6c["then()"]
  e86de8a4_fcf2_726c_6613_1273524e94e2 -->|calls| f26bdb28_b830_3aec_800b_087055150c6c
  d1d39f22_2d9a_4ca3_4bbe_11322ac6d36b["onErrorHook()"]
  e86de8a4_fcf2_726c_6613_1273524e94e2 -->|calls| d1d39f22_2d9a_4ca3_4bbe_11322ac6d36b
  style e86de8a4_fcf2_726c_6613_1273524e94e2 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

lib/reply.js lines 683–754

function sendWebStream (payload, res, reply) {
  if (payload.locked) {
    throw new FST_ERR_REP_READABLE_STREAM_LOCKED()
  }

  let sourceOpen = true
  let errorLogged = false
  let waitingDrain = false
  const reader = payload.getReader()

  eos(res, function (err) {
    if (sourceOpen) {
      if (err != null && res.headersSent && !errorLogged) {
        errorLogged = true
        logStreamError(reply.log, err, res)
      }
      reader.cancel().catch(noop)
    }
  })

  if (!res.headersSent) {
    for (const key in reply[kReplyHeaders]) {
      res.setHeader(key, reply[kReplyHeaders][key])
    }
  } else {
    reply.log.warn('response will send, but you shouldn\'t use res.writeHead in stream mode')
  }

  function onRead (result) {
    if (result.done) {
      sourceOpen = false
      sendTrailer(null, res, reply)
      return
    }
    /* c8 ignore next 5 - race condition: eos handler typically fires first */
    if (res.destroyed) {
      sourceOpen = false
      reader.cancel().catch(noop)
      return
    }
    const shouldContinue = res.write(result.value)
    if (shouldContinue === false) {
      waitingDrain = true
      res.once('drain', onDrain)
      return
    }
    reader.read().then(onRead, onReadError)
  }

  function onDrain () {
    if (!waitingDrain || !sourceOpen || res.destroyed) {
      return
    }
    waitingDrain = false
    reader.read().then(onRead, onReadError)
  }

  function onReadError (err) {
    sourceOpen = false
    if (res.headersSent || reply.request.raw.aborted === true) {
      if (!errorLogged) {
        errorLogged = true
        logStreamError(reply.log, err, reply)
      }
      res.destroy()
    } else {
      onErrorHook(reply, err)
    }
  }

  reader.read().then(onRead, onReadError)
}