refactor(api/fetch): rework streaming for 'text/*' types

Author: jwerle

api/fetch/fetch.js

@@ -348,7 +348,7 @@ function normalizeMethod(method) {
   return methods.indexOf(upcased) > -1 ? upcased : method
 }
 
-export function Request(input, options) {
+export function Request(input, options, xhr) {
   if (!(this instanceof Request)) {
     throw new TypeError('Please use the "new" operator, this DOM object constructor cannot be called as a function.')
   }
@@ -393,7 +393,7 @@ export function Request(input, options) {
   if ((this.method === 'GET' || this.method === 'HEAD') && body) {
     throw new TypeError('Body not allowed for GET or HEAD requests')
   }
-  this._initBody(body, options)
+  this._initBody(body, options, xhr)
 
   if (this.method === 'GET' || this.method === 'HEAD') {
     if (options.cache === 'no-store' || options.cache === 'no-cache') {
@@ -461,7 +461,7 @@ function parseHeaders(rawHeaders) {
 
 Body.call(Request.prototype)
 
-export function Response(bodyInit, options) {
+export function Response(bodyInit, options, xhr) {
   if (!(this instanceof Response)) {
     throw new TypeError('Please use the "new" operator, this DOM object constructor cannot be called as a function.')
   }
@@ -478,7 +478,7 @@ export function Response(bodyInit, options) {
   this.statusText = options.statusText === undefined ? '' : '' + options.statusText
   this.headers = new Headers(options.headers)
   this.url = options.url || ''
-  this._initBody(bodyInit, options)
+  this._initBody(bodyInit, options, xhr)
 }
 
 Body.call(Response.prototype)
@@ -526,14 +526,14 @@ try {
 
 export function fetch(input, init) {
   return new Promise(function(resolve, reject) {
-    var request = new Request(input, init)
+    var xhr = new XMLHttpRequest()
+    var sent = false
+    var request = new Request(input, init, xhr)
 
     if (request.signal && request.signal.aborted) {
       return reject(new DOMException('Aborted', 'AbortError'))
     }
 
-    var xhr = new XMLHttpRequest()
-
     function abortXhr() {
       xhr.abort()
     }
@@ -553,7 +553,9 @@ export function fetch(input, init) {
       options.url = 'responseURL' in xhr ? xhr.responseURL : options.headers.get('X-Request-URL')
       var body = 'response' in xhr ? xhr.response : xhr.responseText
       setTimeout(function() {
-        resolve(new Response(body, options))
+        if (sent) return
+        sent = true
+        resolve(new Response(body, options, xhr))
       }, 0)
     }
 
@@ -575,10 +577,6 @@ export function fetch(input, init) {
       }, 0)
     }
 
-    if (init?.onxhr) {
-      init.onxhr(xhr)
-    }
-
     function fixUrl(url) {
       try {
         return url === '' && g.location.href ? g.location.href : url
@@ -633,6 +631,26 @@ export function fetch(input, init) {
       }
     }
 
+    // XXX(@jwerle): introduced to support 'text/event-stream' or any other streaming text sources
+    xhr.addEventListener('progress', () => {
+      if (sent) return
+      var options = {
+        statusText: xhr.statusText,
+        headers: parseHeaders(xhr.getAllResponseHeaders() || ''),
+        status: xhr.status,
+        url: ''
+      }
+      if ((options.headers.get('content-type') || '').startsWith('text/')) {
+        options.url = 'responseURL' in xhr ? xhr.responseURL : options.headers.get('X-Request-URL')
+        sent = true
+        resolve(new Response(null, options, xhr))
+      }
+    }, { once: true })
+
+    if (init?.onxhr) {
+      init.onxhr(xhr)
+    }
+
     xhr.send(typeof request._bodyInit === 'undefined' ? null : request._bodyInit)
   })
 }

api/fetch/index.js

@@ -30,7 +30,7 @@ const textEncoder = new TextEncoder()
 Response.prototype._initBody = initBody
 Request.prototype._initBody = initBody
 
-async function initBody (body) {
+async function initBody (body, options, xhr) {
   this.body = null
 
   if (
@@ -97,57 +97,68 @@ async function initBody (body) {
     if (this._bodyArrayBuffer) {
       const arrayBuffer = this._bodyArrayBuffer
       this.body = new ReadableStream({
-        async start (controller) {
+        start (controller) {
           controller.enqueue(arrayBuffer)
           controller.close()
         }
       })
     } else if (this._bodyBlob) {
       const blob = this._bodyBlob
       this.body = new ReadableStream({
-        type: 'bytes',
         async start (controller) {
-          const stream = await blob.stream()
-          if (controller.byobRequest) {
-            const reader = stream.getReader({ mode: 'byob' })
-            while (true) {
-              const { done, value } = await reader.read(controller.byobRequest.view)
-              if (done) {
-                break
-              }
-
-              if (value?.byteLength > 0) {
-                controller.byobRequest.respond(value.byteLength)
-              }
-            }
-          } else {
-            const reader = stream.getReader()
-            while (true) {
-              const { done, value } = await reader.read()
-              if (done) {
-                break
-              }
-
-              controller.enqueue(value)
-            }
-          }
-
+          controller.enqueue(await blob.arrayBuffer())
           controller.close()
         }
       })
     } else if (this._bodyText) {
       const text = this._bodyText
-      const encoded = textEncoder.encode(text)
       this.body = new ReadableStream({
-        async start (controller) {
-          controller.enqueue(encoded)
+        start (controller) {
+          controller.enqueue(textEncoder.encode(text))
           controller.close()
         }
       })
     } else {
       this.body = null
     }
   }
+
+  if (this instanceof Request) {
+    options.onxhr = (xhr) => {
+      const accept = this.headers.get('accept') || ''
+      if (accept.startsWith('text/')) {
+        try {
+          xhr.responseType = 'text'
+        } catch {}
+      }
+    }
+  }
+
+  const contentType = this.headers.get('content-type') || ''
+  if (xhr && contentType.startsWith('text/')) {
+    let controller
+    let byteOffset = 0
+    this.body = new ReadableStream({
+      type: 'bytes',
+      start (c) {
+        controller = c
+      }
+    })
+    xhr.addEventListener('load', () => {
+      if (controller) {
+        controller.close()
+      }
+    }, { once: true })
+    xhr.onprogress = () => {
+      try {
+        if (controller) {
+          const buffer = textEncoder.encode(xhr.responseText.slice(byteOffset))
+          byteOffset = xhr.responseText.length
+          controller.enqueue(buffer)
+        }
+      } catch {}
+    }
+  }
 }
 
 Object.defineProperties(Request.prototype, {