Function TestAsyncBufferThreshold

asyncbuffer/buffer_test.go:441–485

(t *testing.T)

}

func TestAsyncBufferThreshold(t *testing.T) {
	_, bytesReader := generateSourceData(t, asyncbuffer.PauseThreshold*3)
	ab := asyncbuffer.New(bytesReader, -1)
	defer ab.Close()

	target := make([]byte, asyncbuffer.ChunkSize)
	n, err := ab.ReadAt(target, 0)
	require.NoError(t, err)
	assert.Equal(t, asyncbuffer.ChunkSize, n)

	// Ensure that buffer hits the pause threshold
	require.Eventually(t, func() bool {
		return bytesReader.BytesRead() >= asyncbuffer.PauseThreshold
	}, 300*time.Millisecond, 10*time.Millisecond)

	// Ensure that buffer never reaches the end of the stream
	require.Never(t, func() bool {
		return bytesReader.BytesRead() >= asyncbuffer.PauseThreshold*2-1
	}, 300*time.Millisecond, 10*time.Millisecond)

	// Let's hit the pause threshold
	target = make([]byte, asyncbuffer.PauseThreshold)
	n, err = ab.ReadAt(target, 0)
	require.NoError(t, err)
	require.Equal(t, asyncbuffer.PauseThreshold, n)

	// Ensure that buffer never reaches the end of the stream
	require.Never(t, func() bool {
		return bytesReader.BytesRead() >= asyncbuffer.PauseThreshold*2-1
	}, 300*time.Millisecond, 10*time.Millisecond)

	// Let's hit the pause threshold
	target = make([]byte, asyncbuffer.PauseThreshold+1)
	n, err = ab.ReadAt(target, 0)
	require.NoError(t, err)

	// It usually returns only pauseThreshold bytes because this exact operation unpauses the reader,
	// but the initial offset is before the threshold, data beyond the threshold may not be available.
	assert.GreaterOrEqual(t, asyncbuffer.PauseThreshold, n)

	// Ensure that buffer hits the end of the stream
	require.Eventually(t, func() bool {
		return bytesReader.BytesRead() >= asyncbuffer.PauseThreshold*2
	}, 300*time.Millisecond, 10*time.Millisecond)
}

func TestAsyncBufferThresholdInstantBeyondAccess(t *testing.T) {
	_, bytesReader := generateSourceData(t, asyncbuffer.PauseThreshold*3)