bool process_next_body_read(JSContext *cx) {
  if (pending_body_reads->length() == 0)
    return true;

  RootedObject owner(cx);
  RootedObject controller(cx);
  {
    HandleObject streamSource = (*pending_body_reads)[0];
    owner = builtins::NativeStreamSource::owner(streamSource);
    controller = builtins::NativeStreamSource::controller(streamSource);
    pending_body_reads->erase(const_cast<JSObject **>(streamSource.address()));
  }

  BodyHandle body = RequestOrResponse::body_handle(owner);
  char *bytes = static_cast<char *>(JS_malloc(cx, HANDLE_READ_CHUNK_SIZE));
  if (!bytes) {
    return error_stream_controller_with_pending_exception(cx, controller);
  }
  size_t nwritten;
  int result = xqd_body_read(body, bytes, HANDLE_READ_CHUNK_SIZE, &nwritten);
  if (!HANDLE_RESULT(cx, result)) {
    JS_free(cx, bytes);
    return error_stream_controller_with_pending_exception(cx, controller);
  }

  if (nwritten == 0) {
    RootedValue r(cx);
    return JS::Call(cx, controller, "close", HandleValueArray::empty(), &r);
  }

  char *new_bytes = static_cast<char *>(JS_realloc(cx, bytes, HANDLE_READ_CHUNK_SIZE, nwritten));
  if (!new_bytes) {
    JS_free(cx, bytes);
    return error_stream_controller_with_pending_exception(cx, controller);
  }
  bytes = new_bytes;

  RootedObject buffer(cx, JS::NewArrayBufferWithContents(cx, nwritten, bytes));
  if (!buffer) {
    JS_free(cx, bytes);
    return error_stream_controller_with_pending_exception(cx, controller);
  }

  RootedObject byte_array(cx, JS_NewUint8ArrayWithBuffer(cx, buffer, 0, nwritten));
  if (!byte_array)
    return false;

  JS::RootedValueArray<1> enqueue_args(cx);
  enqueue_args[0].setObject(*byte_array);
  RootedValue r(cx);
  if (!JS::Call(cx, controller, "enqueue", enqueue_args, &r)) {
    return error_stream_controller_with_pending_exception(cx, controller);
  }

  return true;
}
bool process_next_body_read(JSContext *cx) {
  if (pending_body_reads->length() == 0)
    return true;

  RootedObject owner(cx);
  RootedObject controller(cx);
  {
    HandleObject streamSource = (*pending_body_reads)[0];
    owner = builtins::NativeStreamSource::owner(streamSource);
    controller = builtins::NativeStreamSource::controller(streamSource);
    pending_body_reads->erase(const_cast<JSObject **>(streamSource.address()));
  }

  BodyHandle body = RequestOrResponse::body_handle(owner);
  char *bytes = static_cast<char *>(JS_malloc(cx, HANDLE_READ_CHUNK_SIZE));
  if (!bytes) {
    return error_stream_controller_with_pending_exception(cx, controller);
  }
  size_t nwritten;
  int result = xqd_body_read(body, bytes, HANDLE_READ_CHUNK_SIZE, &nwritten);
  if (!HANDLE_RESULT(cx, result)) {
    JS_free(cx, bytes);
    return error_stream_controller_with_pending_exception(cx, controller);
  }

  if (nwritten == 0) {
    RootedValue r(cx);
    return JS::Call(cx, controller, "close", HandleValueArray::empty(), &r);
  }

  char *new_bytes = static_cast<char *>(JS_realloc(cx, bytes, HANDLE_READ_CHUNK_SIZE, nwritten));
  if (!new_bytes) {
    JS_free(cx, bytes);
    return error_stream_controller_with_pending_exception(cx, controller);
  }
  bytes = new_bytes;

  RootedObject buffer(cx, JS::NewArrayBufferWithContents(cx, nwritten, bytes));
  if (!buffer) {
    JS_free(cx, bytes);
    return error_stream_controller_with_pending_exception(cx, controller);
  }

  RootedObject byte_array(cx, JS_NewUint8ArrayWithBuffer(cx, buffer, 0, nwritten));
  if (!byte_array)
    return false;

  JS::RootedValueArray<1> enqueue_args(cx);
  enqueue_args[0].setObject(*byte_array);
  RootedValue r(cx);
  if (!JS::Call(cx, controller, "enqueue", enqueue_args, &r)) {
    return error_stream_controller_with_pending_exception(cx, controller);
  }

  return true;
}