depager/depager.go

package depager

import (
	"encoding/json"
	"fmt"
	"net/url"
)

// The `Page` interface must wrap server responses. This
// allows pagers to calculate page sizes and iterate over
// response aggregates.
//
// The underlying implementation must be a pointer to a
// struct containing the desired response fields, all tagged
// appropriately. Any fields corresponding to
// platform-specific error responses should also be
// included.
//
// `Count()` must return the total number of items to be
// paged. `Elems()` must return the items from the current
// page.
type Page[T any] interface {
	Count() uint64
	Elems() []T
}

type Client interface {
	Get(uri url.URL) ([]byte, error)
}

type PagedURI interface {
	PageURI(limit, offset uint64) url.URL
}

type Pager[T any] interface {
	Iter() <-chan T
	LastErr() error
}

func NewPager[T any](
	u PagedURI,
	c Client,
	pageSize uint64,
	newPage func() Page[T],
) Pager[T] {
	return &pager[T]{
		client:  c,
		uri:     u,
		m:       0,
		n:       pageSize,
		p:       2,
		newPage: newPage,
	}
}

// Retrieve n items in the range [m*n, m*n + n - 1],
// inclusive. Keep p pages buffered.
type pager[T any] struct {
	client  Client
	uri     PagedURI
	m       uint64
	n       uint64
	err     error
	p       int
	newPage func() Page[T]
}

func (p *pager[T]) iteratePages() <-chan Page[T] {
	ch := make(chan Page[T], p.p)
	go func() {
		defer close(ch)
		for {
			uri := p.uri.PageURI(p.n, p.m*p.n)

			body, err := p.client.Get(uri)
			if err != nil {
				// the client knows better than we do
				p.err = err
				return
			}

			page := p.newPage()

			// TODO: eventually allow other codecs
			err = json.Unmarshal(body, page)
			if err != nil {
				p.err = fmt.Errorf("pager: iterate pages: unmarshal response: %v", err)
				return
			}
			ch <- page

			if (p.m*p.n + p.n) >= page.Count() {
				return
			}
			p.m++
		}
	}()
	return ch
}

func (p *pager[T]) Iter() <-chan T {
	ch := make(chan T, p.n)
	go func() {
		defer close(ch)
		for page := range p.iteratePages() {
			for _, i := range page.Elems() {
				ch <- i
			}
			if p.err != nil {
				p.err = fmt.Errorf("pager: iterate items: %s", p.err)
				return
			}
		}
	}()
	return ch
}

func (p *pager[T]) LastErr() error {
	return p.err
}