throw errorExceptionToDeliver!!
        }

        return -1L // This source is exhausted.
      }
    }

    private fun updateConnectionFlowControl(read: Long) {
      lock.assertNotHeld()

      connection.updateConnectionFlowControl(read)
    }

    /**
     * Accept bytes on the connection's reader thread. This function avoids holding locks while it

      // The removed stream may be blocked on a connection-wide window update.
      condition.signalAll()

      return stream
    }
  }

  internal fun updateConnectionFlowControl(read: Long) {
    this.withLock {
      readBytes.update(total = read)
      val readBytesToAcknowledge = readBytes.unacknowledged
      if (readBytesToAcknowledge >= okHttpSettings.initialWindowSize / 2) {