internal/pagination: add new pagination package
The pagination package provides generic functions for iterating over a collection of items such that each incoming request is filled even after filtering is applied.
This commit is contained in:
Johan Brandhorst-Satzkorn committed
1 parent
dcb67ca616
commit
78541ad05e
6 files changed
+2080
No files matched your search
@@ -0,0 +1,4 @@
|
||||
## Boundary list pagination logic
|
||||
|
||||
This package encapsulates the logic used to attempt to fill every page
|
||||
in a request to a Boundary list endpoint. For example use, see [example_test.go](./example_test.go).
|
||||
@@ -0,0 +1,9 @@
|
||||
// Copyright (c) HashiCorp, Inc.
|
||||
// SPDX-License-Identifier: BUSL-1.1
|
||||
|
||||
// Package pagination implements generic functions for paginating
|
||||
// through a collection using callbacks. The functions will automatically
|
||||
// request additional items to match the input page size even when
|
||||
// some items may be filtered out.
|
||||
// See example_test.go for examples of using the library.
|
||||
package pagination
|
||||
@@ -0,0 +1,215 @@
|
||||
// Copyright (c) HashiCorp, Inc.
|
||||
// SPDX-License-Identifier: BUSL-1.1
|
||||
|
||||
package pagination_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/boundary/internal/boundary"
|
||||
"github.com/hashicorp/boundary/internal/db/timestamp"
|
||||
"github.com/hashicorp/boundary/internal/pagination"
|
||||
"github.com/hashicorp/boundary/internal/refreshtoken"
|
||||
"github.com/hashicorp/boundary/internal/types/resource"
|
||||
)
|
||||
|
||||
// An example resource. In a real example, this would
|
||||
// be a Target or Session or similar.
|
||||
type ExampleResource struct {
|
||||
boundary.Resource
|
||||
Value int
|
||||
}
|
||||
|
||||
func (e *ExampleResource) GetPublicId() string {
|
||||
return "er_1234567890"
|
||||
}
|
||||
|
||||
func (e *ExampleResource) GetUpdateTime() *timestamp.Timestamp {
|
||||
return timestamp.New(time.Now().Add(-10 * time.Hour))
|
||||
}
|
||||
|
||||
func (e *ExampleResource) GetResourceType() resource.Type {
|
||||
return resource.Unknown
|
||||
}
|
||||
|
||||
func ExampleList() {
|
||||
grantsHash := []byte("hash-of-grants") // Acquired from authorization logic
|
||||
pageSize := 10 // From request or service default
|
||||
filterItemFunc := func(ctx context.Context, item *ExampleResource) (bool, error) {
|
||||
// Inspect item to determine whether we want to
|
||||
// include it in the final list.
|
||||
return item.Value < 5, nil
|
||||
}
|
||||
listItemsFunc := func(ctx context.Context, prevPageLast *ExampleResource, limit int) ([]*ExampleResource, error) {
|
||||
// Do the listing of the resource, generally using a
|
||||
// repository method such as target.(*Repository).listTargets.
|
||||
// Use the input to set up any options, for example a limit
|
||||
// or a starting point.
|
||||
if prevPageLast == nil {
|
||||
// No previous page item means this is the first list request.
|
||||
// List using limit.
|
||||
// opts := target.Option{
|
||||
// target.WithLimit(limit),
|
||||
// }
|
||||
} else {
|
||||
// Use the previous page last item to start pagination from the
|
||||
// next page.
|
||||
// opts := target.Option{
|
||||
// target.WithLimit(limit),
|
||||
// target.WithStartPageAfterItem(prevPageLast.GetPublicId(), prevPageLast.GetUpdateTime()),
|
||||
// }
|
||||
}
|
||||
// Example result from the repository
|
||||
return []*ExampleResource{
|
||||
{nil, 0},
|
||||
{nil, 1},
|
||||
{nil, 2},
|
||||
{nil, 3},
|
||||
{nil, 4},
|
||||
{nil, 5},
|
||||
{nil, 6},
|
||||
{nil, 7},
|
||||
{nil, 8},
|
||||
{nil, 9},
|
||||
{nil, 10},
|
||||
}, nil
|
||||
}
|
||||
estimatedCountFunc := func(ctx context.Context) (int, error) {
|
||||
// Get an estimate from the database of the total number
|
||||
// of entries for this resource, usually using some
|
||||
// repository method.
|
||||
return 1000, nil
|
||||
}
|
||||
resp, err := pagination.List(context.Background(), grantsHash, pageSize, filterItemFunc, listItemsFunc, estimatedCountFunc)
|
||||
if err != nil {
|
||||
fmt.Println("failed to paginate", err)
|
||||
return
|
||||
}
|
||||
fmt.Println("Got results:")
|
||||
for _, item := range resp.Items {
|
||||
fmt.Printf("\tValue: %d\n", item.Value)
|
||||
}
|
||||
if resp.CompleteListing {
|
||||
fmt.Println("Listing was complete")
|
||||
} else {
|
||||
fmt.Println("Listing was not complete")
|
||||
}
|
||||
fmt.Println("There are an estimated", resp.EstimatedItemCount, "total items available")
|
||||
// Output: Got results:
|
||||
// Value: 0
|
||||
// Value: 1
|
||||
// Value: 2
|
||||
// Value: 3
|
||||
// Value: 4
|
||||
// Value: 0
|
||||
// Value: 1
|
||||
// Value: 2
|
||||
// Value: 3
|
||||
// Value: 4
|
||||
// Listing was not complete
|
||||
// There are an estimated 1000 total items available
|
||||
}
|
||||
|
||||
func ExampleListRefresh() {
|
||||
grantsHash := []byte("hash-of-grants") // Acquired from authorization logic
|
||||
pageSize := 10 // From request or service default
|
||||
refreshToken, err := refreshtoken.New( // Normally from incoming request
|
||||
context.Background(),
|
||||
time.Now(),
|
||||
time.Now(),
|
||||
resource.Unknown,
|
||||
grantsHash,
|
||||
"er_1234567890",
|
||||
time.Now().Add(-time.Hour),
|
||||
)
|
||||
if err != nil {
|
||||
fmt.Println("failed to paginate", err)
|
||||
return
|
||||
}
|
||||
filterItemFunc := func(ctx context.Context, item *ExampleResource) (bool, error) {
|
||||
// Inspect item to determine whether we want to
|
||||
// include it in the final list.
|
||||
return item.Value < 5, nil
|
||||
}
|
||||
listItemsFunc := func(ctx context.Context, tok *refreshtoken.Token, prevPageLast *ExampleResource, limit int) ([]*ExampleResource, error) {
|
||||
// Do the listing of the resource, generally using a
|
||||
// repository method such as target.(*Repository).listTargets.
|
||||
// Use the input to set up any options, for example a limit
|
||||
// or a starting point.
|
||||
if prevPageLast == nil {
|
||||
// No previous page item means use the values from the refresh token.
|
||||
// opts := target.Option{
|
||||
// target.WithLimit(limit),
|
||||
// target.WithStartPageAfterItem(tok.GetPublicId(), tok.GetUpdateTime()),
|
||||
// }
|
||||
} else {
|
||||
// Use the previous page last item to start pagination from the next page.
|
||||
// opts := target.Option{
|
||||
// target.WithLimit(limit),
|
||||
// target.WithStartPageAfterItem(prevPageLast.GetPublicId(), prevPageLast.GetUpdateTime()),
|
||||
// }
|
||||
}
|
||||
// Example result from the repository
|
||||
return []*ExampleResource{
|
||||
{nil, 0},
|
||||
{nil, 1},
|
||||
{nil, 2},
|
||||
{nil, 3},
|
||||
{nil, 4},
|
||||
{nil, 5},
|
||||
{nil, 6},
|
||||
{nil, 7},
|
||||
{nil, 8},
|
||||
{nil, 9},
|
||||
{nil, 10},
|
||||
}, nil
|
||||
}
|
||||
estimatedCountFunc := func(ctx context.Context) (int, error) {
|
||||
// Get an estimate from the database of the total number
|
||||
// of entries for this resource, usually using some
|
||||
// repository method.
|
||||
return 1000, nil
|
||||
}
|
||||
deletedIdsFunc := func(ctx context.Context, since time.Time) ([]string, time.Time, error) {
|
||||
// Return IDs of resources that have been deleted since the provided timestamp.
|
||||
// Also return a timestamp for which this list was created, allowing the next
|
||||
// invocation to start from the point where this invocation left off.
|
||||
return []string{"er_0123456789"}, time.Now(), nil
|
||||
}
|
||||
resp, err := pagination.ListRefresh(context.Background(), grantsHash, pageSize, filterItemFunc, listItemsFunc, estimatedCountFunc, deletedIdsFunc, refreshToken)
|
||||
if err != nil {
|
||||
fmt.Println("failed to paginate", err)
|
||||
return
|
||||
}
|
||||
fmt.Println("Got results:")
|
||||
for _, item := range resp.Items {
|
||||
fmt.Printf("\tValue: %d\n", item.Value)
|
||||
}
|
||||
if resp.CompleteListing {
|
||||
fmt.Println("Listing was complete")
|
||||
} else {
|
||||
fmt.Println("Listing was not complete")
|
||||
}
|
||||
fmt.Println("There are an estimated", resp.EstimatedItemCount, "total items available")
|
||||
fmt.Println("The following resources have been deleted since we last saw them:")
|
||||
for _, id := range resp.DeletedIds {
|
||||
fmt.Println("\t" + id)
|
||||
}
|
||||
// Output: Got results:
|
||||
// Value: 0
|
||||
// Value: 1
|
||||
// Value: 2
|
||||
// Value: 3
|
||||
// Value: 4
|
||||
// Value: 0
|
||||
// Value: 1
|
||||
// Value: 2
|
||||
// Value: 3
|
||||
// Value: 4
|
||||
// Listing was not complete
|
||||
// There are an estimated 1000 total items available
|
||||
// The following resources have been deleted since we last saw them:
|
||||
// er_0123456789
|
||||
}
|
||||
@@ -0,0 +1,18 @@
|
||||
// Copyright (c) HashiCorp, Inc.
|
||||
// SPDX-License-Identifier: BUSL-1.1
|
||||
|
||||
package pagination
|
||||
|
||||
import (
|
||||
"github.com/hashicorp/boundary/internal/db/timestamp"
|
||||
"github.com/hashicorp/boundary/internal/types/resource"
|
||||
)
|
||||
|
||||
// Item defines a subset of a boundary.Resource that can
|
||||
// be used as an input to a DB operation for the purposes
|
||||
// of pagination and sorting.
|
||||
type Item interface {
|
||||
GetPublicId() string
|
||||
GetUpdateTime() *timestamp.Timestamp
|
||||
GetResourceType() resource.Type
|
||||
}
|
||||
@@ -0,0 +1,260 @@
|
||||
// Copyright (c) HashiCorp, Inc.
|
||||
// SPDX-License-Identifier: BUSL-1.1
|
||||
|
||||
package pagination
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/hashicorp/boundary/internal/boundary"
|
||||
"github.com/hashicorp/boundary/internal/errors"
|
||||
"github.com/hashicorp/boundary/internal/refreshtoken"
|
||||
)
|
||||
|
||||
// ListResponse represents the response from the paginated list operation.
|
||||
type ListResponse[T boundary.Resource] struct {
|
||||
// Items contains the page of items. They have
|
||||
// been filtered according to the behavior of
|
||||
// the filter item func, and will always have as
|
||||
// many items as requested in the page size, unless
|
||||
// there were not enough elements available, in
|
||||
// which case it will contain all elements that
|
||||
// were available.
|
||||
Items []T
|
||||
// CompleteListing signifies whether this page contains
|
||||
// the final item currently available. This indicates
|
||||
// that it may be appropriate to wait some time before
|
||||
// requesting additional pages.
|
||||
CompleteListing bool
|
||||
// RefreshToken is the token that the caller can use
|
||||
// to request a new page of items. The items in the
|
||||
// new page will have been updated more recently
|
||||
// than all the items in the previous page. This
|
||||
// field may be empty if there were no results for a
|
||||
// List call.
|
||||
RefreshToken *refreshtoken.Token
|
||||
// DeletedIds contains a list of item IDs that have been
|
||||
// deleted since the last request for items. This can happen
|
||||
// both during the initial pagination or when requesting a
|
||||
// refresh. This is always empty for the initial List call.
|
||||
DeletedIds []string
|
||||
// EstimatedItemCount is an estimate on exactly how many
|
||||
// items matching the filter function are available. If
|
||||
// a List call is complete, this number is equal to
|
||||
// the number of items returned. Otherwise, the
|
||||
// estimated count function is consulted for an estimate.
|
||||
EstimatedItemCount int
|
||||
}
|
||||
|
||||
// ListFilterFunc is a callback used to filter out resources that don't match
|
||||
// some criteria. The function must return true for items that should be included in the final
|
||||
// result. Returning an error results in an error being returned from the pagination.
|
||||
type ListFilterFunc[T boundary.Resource] func(ctx context.Context, item T) (bool, error)
|
||||
|
||||
// ListItemsFunc returns a slice of T that have been updated since prevPageLastItem.
|
||||
// If prevPageLastItem is empty, it returns a slice of T starting with the least recently updated.
|
||||
type ListItemsFunc[T boundary.Resource] func(ctx context.Context, prevPageLastItem T, limit int) ([]T, error)
|
||||
|
||||
// ListRefreshItemsFunc returns a slice of T that have been updated since prevPageLastItem.
|
||||
// If prevPageLastItem is empty, it returns a slice of T that have been updated since the
|
||||
// item in the refresh token.
|
||||
type ListRefreshItemsFunc[T boundary.Resource] func(ctx context.Context, tok *refreshtoken.Token, prevPageLastItem T, limit int) ([]T, error)
|
||||
|
||||
// EstimatedCountFunc is used to estimate the total number of items
|
||||
// available for the resource that is being listed.
|
||||
type EstimatedCountFunc func(ctx context.Context) (int, error)
|
||||
|
||||
// ListDeletedIDsFunc is used to list the IDs of the resources deleted since
|
||||
// the given timestamp. It returns a slice of IDs and the timestamp of the
|
||||
// instant in which the slice was created.
|
||||
type ListDeletedIDsFunc func(ctx context.Context, since time.Time) ([]string, time.Time, error)
|
||||
|
||||
// List returns a ListResponse. The response will contain at most a
|
||||
// number of items equal to the pageSize. Items are fetched using the
|
||||
// listItemsFn and then items are checked using the filterItemFn
|
||||
// to determine if they should be included in the response.
|
||||
// The response includes a new refresh token based on the grants and items.
|
||||
// The estimatedCountFn is used to provide an estimated total number of
|
||||
// items that can be returned by making additional requests using the provided
|
||||
// refresh token.
|
||||
func List[T boundary.Resource](
|
||||
ctx context.Context,
|
||||
grantsHash []byte,
|
||||
pageSize int,
|
||||
filterItemFn ListFilterFunc[T],
|
||||
listItemsFn ListItemsFunc[T],
|
||||
estimatedCountFn EstimatedCountFunc,
|
||||
) (*ListResponse[T], error) {
|
||||
const op = "pagination.List"
|
||||
|
||||
if len(grantsHash) == 0 {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
|
||||
}
|
||||
if pageSize < 1 {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
|
||||
}
|
||||
if filterItemFn == nil {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
|
||||
}
|
||||
if listItemsFn == nil {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing list items callback")
|
||||
}
|
||||
if estimatedCountFn == nil {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing estimated count callback")
|
||||
}
|
||||
|
||||
items, completeListing, err := list(ctx, pageSize, filterItemFn, listItemsFn)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(ctx, err, op)
|
||||
}
|
||||
|
||||
resp := &ListResponse[T]{
|
||||
Items: items,
|
||||
CompleteListing: completeListing,
|
||||
EstimatedItemCount: len(items),
|
||||
}
|
||||
|
||||
if !completeListing {
|
||||
// If this was not a complete listing, get an estimate
|
||||
// of the total items from the DB.
|
||||
var err error
|
||||
resp.EstimatedItemCount, err = estimatedCountFn(ctx)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(ctx, err, op)
|
||||
}
|
||||
}
|
||||
|
||||
if len(items) > 0 {
|
||||
resp.RefreshToken = refreshtoken.FromResource(items[len(items)-1], grantsHash)
|
||||
}
|
||||
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// ListRefresh returns a ListResponse. The response will contain at most a
|
||||
// number of items equal to the pageSize. Items are fetched using the
|
||||
// listRefreshItemsFn and then items are checked using the filterItemFn
|
||||
// to determine if they should be included in the response.
|
||||
// The response includes a new refresh token based on the grants and items.
|
||||
// The estimatedCountFn is used to provide an estimated total number of
|
||||
// items that can be returned by making additional requests using the provided
|
||||
// refresh token. The listDeletedIDsFn is used to list the IDs of any
|
||||
// resources that have been deleted since the refresh token was last used.
|
||||
func ListRefresh[T boundary.Resource](
|
||||
ctx context.Context,
|
||||
grantsHash []byte,
|
||||
pageSize int,
|
||||
filterItemFn ListFilterFunc[T],
|
||||
listRefreshItemsFn ListRefreshItemsFunc[T],
|
||||
estimatedCountFn EstimatedCountFunc,
|
||||
listDeletedIDsFn ListDeletedIDsFunc,
|
||||
tok *refreshtoken.Token,
|
||||
) (*ListResponse[T], error) {
|
||||
const op = "pagination.ListRefresh"
|
||||
|
||||
if len(grantsHash) == 0 {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing grants hash")
|
||||
}
|
||||
if pageSize < 1 {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "page size must be at least 1")
|
||||
}
|
||||
if filterItemFn == nil {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing filter item callback")
|
||||
}
|
||||
if listRefreshItemsFn == nil {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing list refresh items callback")
|
||||
}
|
||||
if estimatedCountFn == nil {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing estimated count callback")
|
||||
}
|
||||
if listDeletedIDsFn == nil {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing list deleted IDs callback")
|
||||
}
|
||||
if tok == nil {
|
||||
return nil, errors.New(ctx, errors.InvalidParameter, op, "missing refresh token")
|
||||
}
|
||||
|
||||
deletedIds, transactionTimestamp, err := listDeletedIDsFn(ctx, tok.UpdatedTime)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(ctx, err, op)
|
||||
}
|
||||
|
||||
listItemsFn := func(ctx context.Context, prevPageLast T, limit int) ([]T, error) {
|
||||
return listRefreshItemsFn(ctx, tok, prevPageLast, limit)
|
||||
}
|
||||
|
||||
items, completeListing, err := list(ctx, pageSize, filterItemFn, listItemsFn)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(ctx, err, op)
|
||||
}
|
||||
|
||||
resp := &ListResponse[T]{
|
||||
Items: items,
|
||||
CompleteListing: completeListing,
|
||||
DeletedIds: deletedIds,
|
||||
}
|
||||
|
||||
resp.EstimatedItemCount, err = estimatedCountFn(ctx)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(ctx, err, op)
|
||||
}
|
||||
|
||||
if len(items) > 0 {
|
||||
resp.RefreshToken = tok.RefreshLastItem(items[len(items)-1], transactionTimestamp)
|
||||
} else {
|
||||
resp.RefreshToken = tok.Refresh(transactionTimestamp)
|
||||
}
|
||||
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
func list[T boundary.Resource](
|
||||
ctx context.Context,
|
||||
pageSize int,
|
||||
filterItemFn ListFilterFunc[T],
|
||||
listItemsFn ListItemsFunc[T],
|
||||
) ([]T, bool, error) {
|
||||
const op = "pagination.list"
|
||||
|
||||
var lastItem T
|
||||
limit := pageSize + 1
|
||||
items := make([]T, 0, limit)
|
||||
dbLoop:
|
||||
for {
|
||||
// Request another page from the DB until we fill the final items
|
||||
page, err := listItemsFn(ctx, lastItem, limit)
|
||||
if err != nil {
|
||||
return nil, false, errors.Wrap(ctx, err, op)
|
||||
}
|
||||
for _, item := range page {
|
||||
ok, err := filterItemFn(ctx, item)
|
||||
if err != nil {
|
||||
return nil, false, errors.Wrap(ctx, err, op)
|
||||
}
|
||||
if ok {
|
||||
items = append(items, item)
|
||||
// If we filled the items after filtering,
|
||||
// we're done.
|
||||
if len(items) == cap(items) {
|
||||
break dbLoop
|
||||
}
|
||||
}
|
||||
}
|
||||
// If the current page was shorter than the limit, stop iterating
|
||||
if len(page) < limit {
|
||||
break dbLoop
|
||||
}
|
||||
|
||||
lastItem = page[len(page)-1]
|
||||
}
|
||||
// If we couldn't fill the items, it was a complete listing.
|
||||
completeListing := len(items) < cap(items)
|
||||
if !completeListing {
|
||||
// Items is of size pageSize+1, so
|
||||
// truncate if it was filled.
|
||||
items = items[:pageSize]
|
||||
}
|
||||
|
||||
return items, completeListing, nil
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user