depager/v6/depager_test.go

package depager

import (
	"context"
	"fmt"
	"testing"
)

type NoopClient[T any] struct {
	err   error
	pages []*Aggr[T]
	m     int
	cnt   uint64
}

func (c *NoopClient[T]) NextPage(
	page Page[T],
	_offset uint64,
) (err error) {
	if len(c.pages) == 0 {
		return
	}
	if c.m >= len(c.pages) {
		err = fmt.Errorf("client: next page: exceeded max pages")
		return
	}
	src := *c.pages[c.m]
	dst := *page.(*Aggr[T])
	dst = dst[:min(cap(dst), len(src))]
	copy(dst, src)         // update values
	*page.(*Aggr[T]) = dst // update slice

	AggrCount = c.cnt
	err = c.err
	c.m++
	return
}

func NewNoopClient[T any](
	cnt int,
	err error,
	pages []*Aggr[T],
) Client[T] {
	return &NoopClient[T]{
		cnt:   uint64(cnt),
		err:   err,
		pages: pages,
	}
}

var AggrCount uint64

type Aggr[T any] []T

func (a *Aggr[T]) Elems() []T {
	return []T(*a)
}

func (a *Aggr[T]) URI() string {
	return ""
}

func (a *Aggr[T]) Count() uint64 {
	return AggrCount
}

func TestUsingNoopClient(t *testing.T) {
	client := NewNoopClient[any](1, nil, []*Aggr[any]{{}})
	pagePool := make(chan Page[any], 1)
	for i := 0; i < cap(pagePool); i++ {
		tmp := Aggr[any](make([]any, 0, 1))
		pagePool <- &tmp
	}
	pager := NewPager(context.Background(), client, pagePool)
	for range pager.Iter() {
	}
	if err := pager.LastErr(); err != nil {
		t.Errorf("unexpected error in pager with noop client: %v", err)
	}
}

func TestNoopClientReturnsError(t *testing.T) {
	client := NewNoopClient[any](0, fmt.Errorf("whomp"),
		[]*Aggr[any]{{}},
	)
	pagePool := make(chan Page[any], 1)
	for i := 0; i < cap(pagePool); i++ {
		tmp := Aggr[any](make([]any, 0))
		pagePool <- &tmp
	}
	pager := NewPager(context.Background(), client, pagePool)
	for range pager.Iter() {
	}
	if err := pager.LastErr(); err == nil {
		t.Errorf("unexpected success: %v", err)
	}
}

func TestClientReturnsNonemptyPage(t *testing.T) {
	itemCount := 3
	client := NewNoopClient[any](itemCount, nil,
		[]*Aggr[any]{{1, 2}, {3}},
	)
	pagePool := make(chan Page[any], 1)
	for i := 0; i < cap(pagePool); i++ {
		tmp := Aggr[any](make([]any, 0, 2))
		pagePool <- &tmp
	}
	pager := NewPager(context.Background(), client, pagePool)
	var elem int
	for e := range pager.Iter() {
		elem = e.(int)
	}
	if err := pager.LastErr(); err != nil {
		t.Errorf("unexpected error in pager: %v", err)
	}
	if elem != 3 {
		t.Errorf("unexpected value: '%v'", elem)
	}
}

func TestClientReturnsNonemptyPage2(t *testing.T) {
	itemCount := 3
	client := NewNoopClient[any](itemCount, nil,
		[]*Aggr[any]{{1, 2}, {3}},
	)
	pagePool := make(chan Page[any], 1)
	for i := 0; i < cap(pagePool); i++ {
		tmp := Aggr[any](make([]any, 0, 2))
		pagePool <- &tmp
	}
	pager := NewPager(context.Background(), client, pagePool)
	var elem int
	var i int
	for p := range pager.IterPages() {
		elem = p.Elems()[0].(int)
		i++
		pagePool <- p
	}
	if err := pager.LastErr(); err != nil {
		t.Errorf("unexpected error in pager: %v", err)
	}
	if elem != 3 {
		t.Errorf("unexpected value: '%v'", elem)
	}
}

func TestClientReturnsFewerPagesThanExpected(t *testing.T) {
	pageSize := 1
	itemCount := pageSize + 1
	client := NewNoopClient[any](itemCount, nil,
		[]*Aggr[any]{{0}},
	)
	pagePool := make(chan Page[any], 1)
	for i := 0; i < cap(pagePool); i++ {
		tmp := Aggr[any](make([]any, 0, 1))
		pagePool <- &tmp
	}
	pager := NewPager(context.Background(), client, pagePool)
	for range pager.Iter() {
	}
	if err := pager.LastErr(); err == nil {
		t.Errorf("unexpected success in pager: %v", err)
	}
}

func TestClientAbortsPagingItems(t *testing.T) {
	pageSize := 2
	itemCount := pageSize + 1
	client := NewNoopClient[any](itemCount, nil,
		[]*Aggr[any]{{0, 1}, {2}},
	)
	pagePool := make(chan Page[any], 2)
	pg := Aggr[any](make([]any, 0, 2))
	pagePool <- &pg
	pg = Aggr[any](make([]any, 0, 2))
	pagePool <- &pg

	pager := NewPager(context.Background(), client, pagePool)
	for range pager.Iter() {
		break
	}
	if err := pager.Abort(); err != nil {
		t.Errorf("unexpected result of Abort: %v", err)
	}
	if err := pager.LastErr(); err != nil {
		t.Errorf("unexpected error in pager: %v", err)
	}
	if ps := len(pagePool); ps != 2 {
		t.Errorf("unexpected number of pages in page pool: %d", ps)
	}
}

func clearChannel(
	pool chan Page[any],
	ch <-chan Page[any],
) {
}

func TestClientAbortsPagingPages(t *testing.T) {
	cases := []struct{ poolLen, poolCap int }{
		{1, 2},
		{2, 2},
	}
	for _, c := range cases {
		performAbortTest(t, c.poolLen, c.poolCap)
	}
}

func performAbortTest(t *testing.T, poolLen, poolCap int) {
	// Setup
	pageSize := 2
	itemCount := pageSize + 1
	client := NewNoopClient[any](itemCount, nil,
		[]*Aggr[any]{{0, 1}, {2}},
	)
	pagePool := make(chan Page[any], poolCap)
	var pg Aggr[any]

	for i := 0; i < poolLen; i++ {
		pg = Aggr[any](make([]any, 0, 2))
		pagePool <- &pg
	}

	// Abort paging prematurely
	pager := NewPager(context.Background(), client, pagePool)
	ch := pager.IterPages()
	for p := range ch {
		pagePool <- p
		break
	}
	if err := pager.Abort(); err != nil {
		t.Errorf("unexpected result of Abort: %v", err)
	}

	// Return pages to pool
	for len(ch) > 0 {
		pagePool <- <-ch
	}

	// Test our assumptions
	if err := pager.LastErr(); err != nil {
		t.Errorf("unexpected error in pager: %v", err)
	}
	if len(pagePool) != poolLen {
		t.Errorf("unexpected number of pages in page pool: %d", len(pagePool))
	}
	tmp := make(chan Page[any], poolLen)
	pages := make(map[Page[any]]struct{})
	for i := 0; i < len(pagePool); i++ {
		p := <-pagePool
		pages[p] = struct{}{}
		tmp <- p
	}
	if len(pages) != len(tmp) {
		t.Errorf("recovered pages are not unique: page pool length %d is not equal to the number of unique pages %d", len(pagePool), len(pages))
	}
}