Method Read

pkg/streamclient/streamreader.go:149–187

(p []byte)

}

func (r *Reader) Read(p []byte) (int, error) {
	r.lock.Lock()
	defer r.lock.Unlock()

	for len(r.buffer) == 0 && !r.eof && r.err == nil && !r.closed {
		r.cond.Wait()
	}

	if r.closed {
		return 0, io.ErrClosedPipe
	}

	if r.err != nil {
		return 0, r.err
	}

	if len(r.buffer) == 0 && r.eof {
		return 0, io.EOF
	}

	n := copy(p, r.buffer)
	r.buffer = r.buffer[n:]

	if n > 0 {
		currentRwnd := r.readWindow - int64(len(r.buffer))
		if currentRwnd < 0 {
			currentRwnd = 0
		}

		threshold := r.readWindow / 5
		rwndDiff := currentRwnd - r.lastRwndSent

		if len(r.buffer) == 0 || rwndDiff >= threshold {
			r.sendAckLocked(false, false, "")
		}
	}

	return n, nil
}

func (r *Reader) UpdateNextSeq(newSeq int64) {
	r.lock.Lock()