Method readChunks

asyncbuffer/buffer.go:343–424

readChunks reads data from the upstream reader in background and stores them in the pool

()

// readChunks reads data from the upstream reader in background and stores them in the pool
func (ab *AsyncBuffer) readChunks() {
	defer func() {
		if ab.bytesRead.Load() < int64(ab.dataLen) {
			// If the reader has finished reading and we have not read enough data,
			// set err to io.ErrUnexpectedEOF
			ab.setErr(io.ErrUnexpectedEOF)
		}

		// Indicate that the reader has finished reading
		ab.finished.Store(true)
		ab.chunkCond.Close()

		// Close the upstream reader
		if err := ab.r.Close(); err != nil {
			slog.Warn(
				"error closing upstream reader",
				"error", err,
				"source", "asyncbuffer.AsyncBuffer.readChunks",
			)
		}

		ab.callFinishFn()
	}()

	r := ab.r.(io.Reader) //nolint:forcetypeassert
	if ab.dataLen > 0 {
		// If the data length is known, we read only that much data
		r = io.LimitReader(r, int64(ab.dataLen))
	}

	// Stop reading if the reader is closed
	for !ab.closed.Load() {
		// In case we are trying to read data beyond threshold and we are paused,
		// wait for pause to be released.
		if ab.bytesRead.Load() >= PauseThreshold {
			ab.paused.Wait()

			// If the reader has been closed while waiting, we can stop reading
			if ab.closed.Load() {
				return // No more data to read
			}
		}

		// Get a chunk from the pool
		// If the pool is empty, it will create a new byteChunk with ChunkSize
		chunk, ok := chunkPool.Get().(*byteChunk)
		if !ok {
			ab.setErr(errors.New("asyncbuffer.AsyncBuffer.readChunks: failed to get chunk from pool"))
			return
		}

		// Read data into the chunk's buffer
		// There is no way to guarantee that r.Read will abort on context cancellation,
		// unfortunately, this is how golang works.
		n, err := ioutil.TryReadFull(r, chunk.buf)

		// If it's not the EOF, we need to store the error
		if err != nil && !errors.Is(err, io.EOF) {