Repository: ably/ably-go
Pull request #281 (2021-02-16 12:01:17)

Adapt Stats to follow new paginated result design
https://github.com/ably/ably-go/issues/278

The innards of paginated result handling have been simplified too.
(Only for stats; once all paginated results are refactored, the old
handling code can be removed and the "New" suffixes dropped.)

tcard: Note that this part is still missing:

```
// Paging implicit; just iterate over items.
items := rest.Foo(...).Items()
for items.Next(ctx) {
	item := items.Item()
	...
}
if err := r.Err(); err != nil {
	...
}
```

Patch:
diff --git a/README.md b/README.md
index 6992a4b..d0d14ee 100644
--- a/README.md
+++ b/README.md
@@ -243,6 +243,10 @@ if err != nil {
 
 ### Querying the History
 
+<!-- GO EXAMPLE
+{
+-->
+
 ```go
 page, err := channel.History(ctx, nil)
 for ; err == nil && page != nil; page, err = page.Next(ctx) {
@@ -255,10 +259,18 @@ if err != nil {
 }
 ```
 
+<!-- GO EXAMPLE
+}
+-->
+
 ### Presence on a channel
 
+<!-- GO EXAMPLE
+{
+-->
+
 ```go
-page, err = channel.Presence.Get(ctx, nil)
+page, err := channel.Presence.Get(ctx, nil)
 for ; err == nil && page != nil; page, err = page.Next(ctx) {
 	for _, presence := range page.PresenceMessages() {
 		fmt.Println(presence)
@@ -269,10 +281,18 @@ if err != nil {
 }
 ```
 
+<!-- GO EXAMPLE
+}
+-->
+
 ### Querying the Presence History
 
+<!-- GO EXAMPLE
+{
+-->
+
 ```go
-page, err = channel.Presence.History(ctx, nil)
+page, err := channel.Presence.History(ctx, nil)
 for ; err == nil && page != nil; page, err = page.Next(ctx) {
 	for _, presence := range page.PresenceMessages() {
 		fmt.Println(presence)
@@ -283,16 +303,27 @@ if err != nil {
 }
 ```
 
+<!-- GO EXAMPLE
+}
+-->
+
 ### Fetching your application's stats
 
+<!-- GO EXAMPLE
+{
+-->
+
 ```go
-page, err = client.Stats(ctx, &ably.PaginateParams{})
-for ; err == nil && page != nil; page, err = page.Next(ctx) {
-	for _, stat := range page.Stats() {
+pages, err := client.Stats().Pages(ctx)
+if err != nil {
+	panic(err)
+}
+for pages.Next(ctx) {
+	for _, stat := range pages.Items() {
 		fmt.Println(stat)
 	}
 }
-if err != nil {
+if err := pages.Err(); err != nil {
 	panic(err)
 }
 ```
@@ -301,6 +332,10 @@ if err != nil {
 }
 -->
 
+<!-- GO EXAMPLE
+}
+-->
+
 ## Feature support
 
 This library targets the Ably 1.2 [client library specification](https://www.ably.io/documentation/client-lib-development-guide/features).
diff --git a/ably/doc.go b/ably/doc.go
index e053372..1c0c474 100644
--- a/ably/doc.go
+++ b/ably/doc.go
@@ -33,4 +33,28 @@
 //
 // For messages and presence messages, "on" is called "subscribe" and "off" is
 // called "unsubscribe".
+//
+// Paginated results
+//
+// Most requests to the Ably REST API return a single page of results, with
+// hyperlinks to the first and next pages in the whole collection of results.
+// To facilitate navigating through these pages, the library provides access to
+// such paginated results though a common pattern.
+//
+// A method that prepares a paginated request returns a Request object with two
+// methods: Pages and Items. Pages returns a PaginatedResult, an iterator that,
+// on each iteration, yields a whole page of results. Items is simply a
+// convenience wrapper that yields single results insteads.
+//
+// In both cases, calling the method validates the request and may return an
+// error.
+//
+// Then, for accessing the results, the Next method from the resulting
+// iterator object must be called repeatedly; each time it returns true, the
+// result that has been retrieved can be inspected with the Items or Item method
+// from the iterator object. Finally, once it returns false, the Err method must
+// be called to check if the iterator stopped due to some error, or else, it
+// just finished going through all pages.
+//
+// See the PaginatedResults example.
 package ably
diff --git a/ably/paginated_result.go b/ably/paginated_result.go
index 8d6e406..f78713b 100644
--- a/ably/paginated_result.go
+++ b/ably/paginated_result.go
@@ -16,6 +16,161 @@ import (
 	"github.com/ably/ably-go/ably/proto"
 )
 
+type Direction string
+
+const (
+	Backwards Direction = "backwards"
+	Forwards  Direction = "forwards"
+)
+
+type paginatedRequestNew struct {
+	path   string
+	params url.Values
+
+	query queryFunc
+}
+
+func (r *REST) newPaginatedRequest(path string, params url.Values) paginatedRequestNew {
+	return paginatedRequestNew{
+		path:   path,
+		params: params,
+
+		query: query(r.get),
+	}
+}
+
+// PaginatedResultNew is a generic iterator for PaginatedResult pagination.
+// Items decoding is delegated to type-specific wrappers.
+//
+// See "Paginated results" section in the package-level documentation.
+type PaginatedResultNew struct {
+	basePath  string
+	nextLink  string
+	firstLink string
+	res       *http.Response
+	err       error
+
+	query queryFunc
+	first bool
+}
+
+// load loads the first page of results. Must be called from the type-specific
+// wrapper Pages method that creates the PaginatedResult object.
+func (p *PaginatedResultNew) load(ctx context.Context, r paginatedRequestNew) error {
+	p.basePath = path.Dir(r.path)
+	p.firstLink = (&url.URL{
+		Path:     r.path,
+		RawQuery: r.params.Encode(),
+	}).String()
+	p.query = r.query
+	return p.First(ctx)
+}
+
+// loadItems loads the first page of results and returns a next function. Must
+// be called from the type-specific wrapper Items method that creates the
+// PaginatedItems object.
+//
+// The returned next function must be called from the wrapper's Next method, and
+// returns the index of the object that should be returned by the Item method,
+// previously loading the next page if necessary.
+//
+// pageDecoder will be called each time a new page is retrieved under the hood.
+// It should return a destination object on which the page of results will be
+// decoded, and a pageLength function that, when called after the page has been
+// decoded, must return the length of the page.
+func (p *PaginatedResultNew) loadItems(
+	ctx context.Context,
+	r paginatedRequestNew,
+	pageDecoder func() (page interface{}, pageLength func() int),
+) (
+	next func(context.Context) (int, bool),
+	err error,
+) {
+	err = p.load(ctx, r)
+	if err != nil {
+		return nil, err
+	}
+
+	var page interface{}
+	var pageLen int
+	nextItem := 0
+
+	return func(ctx context.Context) (int, bool) {
+		if nextItem == 0 {
+			var getLen func() int
+			page, getLen = pageDecoder()
+			hasNext := p.next(ctx, &page)
+			if !hasNext {
+				return 0, false
+			}
+			pageLen = getLen()
+		}
+
+		idx := nextItem
+		nextItem = (nextItem + 1) % pageLen
+		return idx, true
+	}, nil
+}
+
+func (p *PaginatedResultNew) goTo(ctx context.Context, link string) error {
+	var err error
+	p.res, err = p.query(ctx, link)
+	if err != nil {
+		return err
+	}
+	p.nextLink = ""
+	for _, rawLink := range p.res.Header["Link"] {
+		m := relLinkRegexp.FindStringSubmatch(rawLink)
+		if len(m) == 0 {
+			continue
+		}
+		relPath, rel := m[1], m[2]
+		path := path.Join(p.basePath, relPath)
+		switch rel {
+		case "first":
+			p.firstLink = path
+		case "next":
+			p.nextLink = path
+		}
+	}
+	return nil
+}
+
+// next loads the next page of items, if there is one. It returns whether a page
+// was successfully loaded or not; after it returns false, Err should be
+// called to check for any errors.
+//
+// Items can then be inspected with the type-specific Items method.
+//
+// For items iterators, use the next function returned by loadItems instead.
+func (p *PaginatedResultNew) next(ctx context.Context, into interface{}) bool {
+	if !p.first {
+		if p.nextLink == "" {
+			return false
+		}
+		p.err = p.goTo(ctx, p.nextLink)
+		if p.err != nil {
+			return false
+		}
+	}
+	p.first = false
+
+	p.err = decodeResp(p.res, into)
+	return p.err == nil
+}
+
+// First loads the first page of items. Next should be called before inspecting
+// the Items.
+func (p *PaginatedResultNew) First(ctx context.Context) error {
+	p.first = true
+	return p.goTo(ctx, p.firstLink)
+}
+
+// Err returns the error that caused Next to fail, if there was one.
+func (p *PaginatedResultNew) Err() error {
+	return p.err
+}
+
 // relLinkRegexp is the regexp that matches our pagination format
 var relLinkRegexp = regexp.MustCompile(`<(?P<url>[^>]+)>; rel="(?P<rel>[^"]+)"`)
 
@@ -230,18 +385,9 @@ func (p *PaginatedResult) PresenceMessages() []*PresenceMessage {
 	return items
 }
 
-// Stats gives a slice of statistics for the current page. The method panics if
-// the underlying paginated result is not statistics.
-func (p *PaginatedResult) Stats() []*Stats {
-	items, ok := p.typItems.([]*proto.Stats)
-	if !ok {
-		panic(errInvalidType{typ: p.req.typ})
-	}
-	return items
-}
-
 type Stats = proto.Stats
 type StatsMessageTypes = proto.MessageTypes
+type StatsMessageCount = proto.MessageCount
 type StatsMessageTraffic = proto.MessageTraffic
 type StatsConnectionTypes = proto.ConnectionTypes
 type StatsResourceCount = proto.ResourceCount
diff --git a/ably/proto/stat.go b/ably/proto/stat.go
index 621262b..622d686 100644
--- a/ably/proto/stat.go
+++ b/ably/proto/stat.go
@@ -1,6 +1,9 @@
 package proto
 
-import "time"
+import (
+	"fmt"
+	"time"
+)
 
 const (
 	StatGranularityMinute = "minute"
@@ -46,7 +49,7 @@ type MessageCount struct {
 
 type MessageTypes struct {
 	All      MessageCount `json:"all" codec:"all"`
-	Plain    MessageCount `json:"plain" codec:"plain"`
+	Messages MessageCount `json:"messages" codec:"messages"`
 	Presence MessageCount `json:"presence" codec:"presence"`
 }
 
@@ -136,3 +139,7 @@ type Stats struct {
 	XchgConsumer  XchgMessages    `json:"xchgConsumer" codec:"xchgConsumer"`
 	PeakRates     Rates           `json:"peakRates" codec:"peakRates"`
 }
+
+func (s Stats) String() string {
+	return fmt.Sprintf("<Stats %v; unit=%v; count=%v>", s.IntervalID, s.Unit, s.Count)
+}
diff --git a/ably/realtime_client.go b/ably/realtime_client.go
index 98ca2ce..a904600 100644
--- a/ably/realtime_client.go
+++ b/ably/realtime_client.go
@@ -51,11 +51,9 @@ func (c *Realtime) Close() {
 	c.Connection.Close()
 }
 
-// Stats gives the clients metrics according to the given parameters. The
-// returned result can be inspected for the statistics via the Stats()
-// method.
-func (c *Realtime) Stats(ctx context.Context, params *PaginateParams) (*PaginatedResult, error) {
-	return c.rest.Stats(ctx, params)
+// Stats is the same as REST.Stats.
+func (c *Realtime) Stats(o ...StatsOption) StatsRequest {
+	return c.rest.Stats(o...)
 }
 
 // Time
diff --git a/ably/rest_client.go b/ably/rest_client.go
index 4615ce1..0a876b6 100644
--- a/ably/rest_client.go
+++ b/ably/rest_client.go
@@ -14,7 +14,9 @@ import (
 	"net/http"
 	"net/http/httptrace"
 	"net/http/httputil"
+	"net/url"
 	"reflect"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -25,7 +27,6 @@ import (
 
 var (
 	msgType     = reflect.TypeOf((*[]*proto.Message)(nil)).Elem()
-	statType    = reflect.TypeOf((*[]*proto.Stats)(nil)).Elem()
 	presMsgType = reflect.TypeOf((*[]*proto.PresenceMessage)(nil)).Elem()
 	arrayTyp    = reflect.TypeOf((*[]interface{})(nil)).Elem()
 )
@@ -154,11 +155,141 @@ func (c *REST) Time(ctx context.Context) (time.Time, error) {
 	return time.Unix(times[0]/1000, times[0]%1000), nil
 }
 
-// Stats gives the channel's metrics according to the given parameters.
-// The returned result can be inspected for the statistics via the Stats()
-// method.
-func (c *REST) Stats(ctx context.Context, params *PaginateParams) (*PaginatedResult, error) {
-	return newPaginatedResult(ctx, nil, paginatedRequest{typ: statType, path: "/stats", params: params, query: query(c.get), logger: c.logger(), respCheck: checkValidHTTPResponse})
+// Stats retrieves statistics about the Ably app's activity.
+func (c *REST) Stats(o ...StatsOption) StatsRequest {
+	params := (&statsOptions{}).apply(o...)
+	return StatsRequest{r: c.newPaginatedRequest("/stats", params)}
+}
+
+// A StatsOption configures a call to REST.Stats or Realtime.Stats.
+type StatsOption func(*statsOptions)
+
+func StatsWithStart(t time.Time) StatsOption {
+	return func(o *statsOptions) {
+		o.params.Set("start", strconv.FormatInt(unixMilli(t), 10))
+	}
+}
+
+func StatsWithEnd(t time.Time) StatsOption {
+	return func(o *statsOptions) {
+		o.params.Set("end", strconv.FormatInt(unixMilli(t), 10))
+	}
+}
+
+func StatsWithLimit(limit int) StatsOption {
+	return func(o *statsOptions) {
+		o.params.Set("limit", strconv.Itoa(limit))
+	}
+}
+
+func StatsWithDirection(d Direction) StatsOption {
+	return func(o *statsOptions) {
+		o.params.Set("direction", string(d))
+	}
+}
+
+type PeriodUnit string
+
+const (
+	PeriodMinute PeriodUnit = "minute"
+	PeriodHour   PeriodUnit = "hour"
+	PeriodDay    PeriodUnit = "day"
+	PeriodMonth  PeriodUnit = "month"
+)
+
+func StatsWithUnit(d PeriodUnit) StatsOption {
+	return func(o *statsOptions) {
+		o.params.Set("unit", string(d))
+	}
+}
+
+type statsOptions struct {
+	params url.Values
+}
+
+func (o *statsOptions) apply(opts ...StatsOption) url.Values {
+	o.params = make(url.Values)
+	for _, opt := range opts {
+		opt(o)
+	}
+	return o.params
+}
+
+// StatsRequest represents a request prepared by the REST.Stats or
+// Realtime.Stats method, ready to be performed by its Pages or Items methods.
+type StatsRequest struct {
+	r paginatedRequestNew
+}
+
+// Pages returns an iterator for whole pages of Stats.
+//
+// See "Paginated results" section in the package-level documentation.
+func (r StatsRequest) Pages(ctx context.Context) (*StatsPaginatedResult, error) {
+	var res StatsPaginatedResult
+	return &res, res.load(ctx, r.r)
+}
+
+// A StatsPaginatedResult is an iterator for the result of a Stats request.
+//
+// See "Paginated results" section in the package-level documentation.
+type StatsPaginatedResult struct {
+	PaginatedResultNew
+	items []*Stats
+}
+
+// Next retrieves the next page of results.
+//
+// See the "Paginated results" section in the package-level documentation.
+func (p *StatsPaginatedResult) Next(ctx context.Context) bool {
+	p.items = nil // avoid mutating already returned items
+	return p.next(ctx, &p.items)
+}
+
+// Items returns the current page of results.
+//
+// See the "Paginated results" section in the package-level documentation.
+func (p *StatsPaginatedResult) Items() []*Stats {
+	return p.items
+}
+
+// Items returns a convenience iterator for single Stats, over an underlying
+// paginated iterator.
+//
+// See "Paginated results" section in the package-level documentation.
+func (r StatsRequest) Items(ctx context.Context) (*StatsPaginatedItems, error) {
+	var res StatsPaginatedItems
+	var err error
+	res.next, err = res.loadItems(ctx, r.r, func() (interface{}, func() int) {
+		res.items = nil // avoid mutating already returned items
+		return &res.items, func() int { return len(res.items) }
+	})
+	return &res, err
+}
+
+type StatsPaginatedItems struct {
+	PaginatedResultNew
+	items []*Stats
+	item  *Stats
+	next  func(context.Context) (int, bool)
+}
+
+// Next retrieves the next result.
+//
+// See the "Paginated results" section in the package-level documentation.
+func (p *StatsPaginatedItems) Next(ctx context.Context) bool {
+	i, ok := p.next(ctx)
+	if !ok {
+		return false
+	}
+	p.item = p.items[i]
+	return true
+}
+
+// Item returns the current result.
+//
+// See the "Paginated results" section in the package-level documentation.
+func (p *StatsPaginatedItems) Item() *Stats {
+	return p.item
 }
 
 // request this contains fields necessary to compose http request that will be


Test patch:
diff --git a/ably/ablytest/pagination.go b/ably/ablytest/pagination.go
new file mode 100644
index 0000000..c4952cc
--- /dev/null
+++ b/ably/ablytest/pagination.go
@@ -0,0 +1,150 @@
+package ablytest
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+)
+
+func TestPagination(expected, request interface{}, perPage int) error {
+	var items []interface{}
+	rexpected := reflect.ValueOf(expected)
+	for i := 0; i < reflect.ValueOf(expected).Len(); i++ {
+		items = append(items, rexpected.Index(i).Interface())
+	}
+	return testPagination(reflect.ValueOf(request), items, perPage)
+}
+
+func testPagination(request reflect.Value, expectedItems []interface{}, perPage int) error {
+	getPages, getItems := generalizePagination(request)
+
+	var expectedPages [][]interface{}
+	var page []interface{}
+	for _, item := range expectedItems {
+		page = append(page, item)
+		if len(page) == perPage {
+			expectedPages = append(expectedPages, page)
+			page = nil
+		}
+	}
+	if len(page) > 0 {
+		expectedPages = append(expectedPages, page)
+	}
+
+	for i := 0; i < 2; i++ {
+		pages, err := getPages()
+		if err != nil {
+			return fmt.Errorf("calling Pages: %w", err)
+		}
+		var gotPages [][]interface{}
+		for pages.next() {
+			gotPages = append(gotPages, pages.items())
+		}
+		if err := pages.err(); err != nil {
+			return fmt.Errorf("iterating pages: %w", err)
+		}
+
+		if !reflect.DeepEqual(expectedPages, gotPages) {
+			return fmt.Errorf("expected pages: %+v, got: %+v", expectedPages, gotPages)
+		}
+
+		if err := pages.first(); err != nil {
+			return fmt.Errorf("going back to first page: %w", err)
+		}
+	}
+
+	for i := 0; i < 2; i++ {
+		items, err := getItems()
+		if err != nil {
+			return fmt.Errorf("calling Items: %w", err)
+		}
+		var gotItems []interface{}
+		for items.next() {
+			gotItems = append(gotItems, items.item())
+		}
+		if err := items.err(); err != nil {
+			return fmt.Errorf("iterating items: %w", err)
+		}
+
+		if !reflect.DeepEqual(expectedItems, gotItems) {
+			return fmt.Errorf("expected items: %+v, got: %+v", expectedItems, gotItems)
+		}
+
+		if err := items.first(); err != nil {
+			return fmt.Errorf("going back to first page: %w", err)
+		}
+	}
+
+	return nil
+}
+
+type paginated struct {
+	next  func() bool
+	first func() error
+	err   func() error
+}
+
+type paginatedResult struct {
+	paginated
+	items func() []interface{}
+}
+
+type paginatedItems struct {
+	paginated
+	item func() interface{}
+}
+
+func generalizePagination(request reflect.Value) (func() (paginatedResult, error), func() (paginatedItems, error)) {
+	ctx := reflect.ValueOf(context.Background())
+
+	generalizeCommon := func(r reflect.Value) paginated {
+		return paginated{
+			next: func() bool {
+				return r.MethodByName("Next").Call([]reflect.Value{ctx})[0].Bool()
+			},
+			first: func() error {
+				err, _ := r.MethodByName("First").Call([]reflect.Value{ctx})[0].Interface().(error)
+				return err
+			},
+			err: func() error {
+				err, _ := r.MethodByName("Err").Call(nil)[0].Interface().(error)
+				return err
+			},
+		}
+	}
+
+	pages := func() (paginatedResult, error) {
+		ret := request.MethodByName("Pages").Call([]reflect.Value{ctx})
+		if err, ok := ret[1].Interface().(error); ok && err != nil {
+			return paginatedResult{}, err
+		}
+		r := ret[0]
+		return paginatedResult{
+			paginated: generalizeCommon(r),
+			items: func() []interface{} {
+				ritems := r.MethodByName("Items").Call(nil)[0]
+				var items []interface{}
+				for i := 0; i < ritems.Len(); i++ {
+					items = append(items, ritems.Index(i).Interface())
+				}
+				return items
+			},
+		}, nil
+	}
+
+	items := func() (paginatedItems, error) {
+		ret := request.MethodByName("Items").Call([]reflect.Value{ctx})
+		if err, ok := ret[1].Interface().(error); ok && err != nil {
+			return paginatedItems{}, err
+		}
+		r := ret[0]
+		return paginatedItems{
+			paginated: generalizeCommon(r),
+			item: func() interface{} {
+				return r.MethodByName("Item").Call(nil)[0].Interface()
+			},
+		}, nil
+	}
+
+	return pages, items
+}
diff --git a/ably/auth_test.go b/ably/auth_test.go
index d00a4b3..a9de5b7 100644
--- a/ably/auth_test.go
+++ b/ably/auth_test.go
@@ -55,7 +55,7 @@ func TestAuth_BasicAuth(t *testing.T) {
 	if _, err := client.Time(context.Background()); err != nil {
 		t.Fatalf("client.Time()=%v", err)
 	}
-	if _, err := client.Stats(context.Background(), single()); err != nil {
+	if _, err := client.Stats().Pages(context.Background()); err != nil {
 		t.Fatalf("client.Stats()=%v", err)
 	}
 	if n := rec.Len(); n != 2 {
@@ -107,7 +107,7 @@ func TestAuth_TokenAuth(t *testing.T) {
 	if _, err := client.Time(context.Background()); err != nil {
 		t.Fatalf("client.Time()=%v", err)
 	}
-	if _, err := client.Stats(context.Background(), single()); err != nil {
+	if _, err := client.Stats().Pages(context.Background()); err != nil {
 		t.Fatalf("client.Stats()=%v", err)
 	}
 	// At this points there should be two requests recorded:
@@ -262,7 +262,7 @@ func TestAuth_TokenAuth_Renew(t *testing.T) {
 		t.Fatalf("want ttl=1s; got %v", ttl)
 	}
 	time.Sleep(2 * time.Second) // wait till expires
-	_, err = client.Stats(context.Background(), single())
+	_, err = client.Stats().Pages(context.Background())
 	if err != nil {
 		t.Fatalf("Stats()=%v", err)
 	}
@@ -296,7 +296,7 @@ func TestAuth_TokenAuth_Renew(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewREST()=%v", err)
 	}
-	if _, err := client.Stats(context.Background(), single()); err == nil {
+	if _, err := client.Stats().Pages(context.Background()); err == nil {
 		t.Fatal("want err!=nil")
 	}
 	// Ensure no requests were made to Ably servers.
@@ -460,7 +460,7 @@ func TestAuth_RequestToken(t *testing.T) {
 			t.Errorf("NewRealtime()=%v", err)
 			continue
 		}
-		if _, err = c.Stats(context.Background(), single()); err != nil {
+		if _, err = c.Stats().Pages(context.Background()); err != nil {
 			t.Errorf("c.Stats()=%v (method=%s)", err, method)
 		}
 	}
diff --git a/ably/doc_test.go b/ably/doc_test.go
new file mode 100644
index 0000000..7a7670b
--- /dev/null
+++ b/ably/doc_test.go
@@ -0,0 +1,8 @@
+package ably_test
+
+import "fmt"
+
+func Example_paginatedResults() {
+	fmt.Println("TODO once the Requests method is adapted")
+	// Output: TODO once the Requests method is adapted
+}
diff --git a/ably/readme_examples_test.go b/ably/readme_examples_test.go
index 5caad05..98a93cd 100644
--- a/ably/readme_examples_test.go
+++ b/ably/readme_examples_test.go
@@ -170,54 +170,70 @@ func TestReadmeExamples(t *testing.T) {
 			/* README.md:240 */ panic(err)
 			/* README.md:241 */
 		}
-		/* README.md:247 */ page, err := channel.History(ctx, nil)
-		/* README.md:248 */ for ; err == nil && page != nil; page, err = page.Next(ctx) {
-			/* README.md:249 */ for _, message := range page.Messages() {
-				/* README.md:250 */ fmt.Println(message)
-				/* README.md:251 */
+		/* README.md:247 */ {
+			/* README.md:251 */ page, err := channel.History(ctx, nil)
+			/* README.md:252 */ for ; err == nil && page != nil; page, err = page.Next(ctx) {
+				/* README.md:253 */ for _, message := range page.Messages() {
+					/* README.md:254 */ fmt.Println(message)
+					/* README.md:255 */
+				}
+				/* README.md:256 */
 			}
-			/* README.md:252 */
-		}
-		/* README.md:253 */ if err != nil {
-			/* README.md:254 */ panic(err)
-			/* README.md:255 */
-		}
-		/* README.md:261 */ page, err = channel.Presence.Get(ctx, nil)
-		/* README.md:262 */ for ; err == nil && page != nil; page, err = page.Next(ctx) {
-			/* README.md:263 */ for _, presence := range page.PresenceMessages() {
-				/* README.md:264 */ fmt.Println(presence)
-				/* README.md:265 */
+			/* README.md:257 */ if err != nil {
+				/* README.md:258 */ panic(err)
+				/* README.md:259 */
 			}
-			/* README.md:266 */
-		}
-		/* README.md:267 */ if err != nil {
-			/* README.md:268 */ panic(err)
-			/* README.md:269 */
+			/* README.md:263 */
 		}
-		/* README.md:275 */ page, err = channel.Presence.History(ctx, nil)
-		/* README.md:276 */ for ; err == nil && page != nil; page, err = page.Next(ctx) {
-			/* README.md:277 */ for _, presence := range page.PresenceMessages() {
-				/* README.md:278 */ fmt.Println(presence)
-				/* README.md:279 */
+		/* README.md:269 */ {
+			/* README.md:273 */ page, err := channel.Presence.Get(ctx, nil)
+			/* README.md:274 */ for ; err == nil && page != nil; page, err = page.Next(ctx) {
+				/* README.md:275 */ for _, presence := range page.PresenceMessages() {
+					/* README.md:276 */ fmt.Println(presence)
+					/* README.md:277 */
+				}
+				/* README.md:278 */
 			}
-			/* README.md:280 */
-		}
-		/* README.md:281 */ if err != nil {
-			/* README.md:282 */ panic(err)
-			/* README.md:283 */
+			/* README.md:279 */ if err != nil {
+				/* README.md:280 */ panic(err)
+				/* README.md:281 */
+			}
+			/* README.md:285 */
 		}
-		/* README.md:289 */ page, err = client.Stats(ctx, &ably.PaginateParams{})
-		/* README.md:290 */ for ; err == nil && page != nil; page, err = page.Next(ctx) {
-			/* README.md:291 */ for _, stat := range page.Stats() {
-				/* README.md:292 */ fmt.Println(stat)
-				/* README.md:293 */
+		/* README.md:291 */ {
+			/* README.md:295 */ page, err := channel.Presence.History(ctx, nil)
+			/* README.md:296 */ for ; err == nil && page != nil; page, err = page.Next(ctx) {
+				/* README.md:297 */ for _, presence := range page.PresenceMessages() {
+					/* README.md:298 */ fmt.Println(presence)
+					/* README.md:299 */
+				}
+				/* README.md:300 */
 			}
-			/* README.md:294 */
+			/* README.md:301 */ if err != nil {
+				/* README.md:302 */ panic(err)
+				/* README.md:303 */
+			}
+			/* README.md:307 */
 		}
-		/* README.md:295 */ if err != nil {
-			/* README.md:296 */ panic(err)
-			/* README.md:297 */
+		/* README.md:313 */ {
+			/* README.md:317 */ pages, err := client.Stats().Pages(ctx)
+			/* README.md:318 */ if err != nil {
+				/* README.md:319 */ panic(err)
+				/* README.md:320 */
+			}
+			/* README.md:321 */ for pages.Next(ctx) {
+				/* README.md:322 */ for _, stat := range pages.Items() {
+					/* README.md:323 */ fmt.Println(stat)
+					/* README.md:324 */
+				}
+				/* README.md:325 */
+			}
+			/* README.md:326 */ if err := pages.Err(); err != nil {
+				/* README.md:327 */ panic(err)
+				/* README.md:328 */
+			}
+			/* README.md:332 */
 		}
-		/* README.md:301 */
+		/* README.md:336 */
 	}
 }
diff --git a/ably/rest_client_test.go b/ably/rest_client_test.go
index f490bdf..3fa9e44 100644
--- a/ably/rest_client_test.go
+++ b/ably/rest_client_test.go
@@ -1,6 +1,7 @@
 package ably_test
 
 import (
+	"bytes"
 	"context"
 	"crypto/tls"
 	"encoding/json"
@@ -11,6 +12,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"reflect"
 	"regexp"
 	"strings"
 	"sync/atomic"
@@ -201,19 +203,20 @@ func TestRestClient(t *testing.T) {
 					errCh <- errors.New("timeout waiting for client.Stats to return nonempty value")
 					return
 				case <-tick:
-					page, err := client.Stats(context.Background(), &ably.PaginateParams{
-						Limit: 1,
-						ScopeParams: ably.ScopeParams{
-							Start: longAgo,
-							Unit:  proto.StatGranularityMinute,
-						},
-					})
+					page, err := client.Stats(
+						ably.StatsWithLimit(1),
+						ably.StatsWithStart(longAgo),
+						ably.StatsWithUnit(ably.PeriodMinute),
+					).Pages(context.Background())
 					if err != nil {
 						errCh <- err
 						return
 					}
-
-					if stats := page.Stats(); len(stats) != 0 {
+					if !page.Next(context.Background()) {
+						errCh <- page.Err()
+						return
+					}
+					if stats := page.Items(); len(stats) != 0 {
 						statsCh <- stats
 						return
 					}
@@ -595,3 +598,201 @@ func (c *connCloseTracker) Close() error {
 	atomic.StoreUintptr(&c.closed, 1)
 	return c.Conn.Close()
 }
+
+func TestStatsPagination_RSC6a_RSCb3(t *testing.T) {
+	t.Parallel()
+
+	for _, limit := range []int{2, 3, 20} {
+		t.Run(fmt.Sprintf("limit=%d", limit), func(t *testing.T) {
+			t.Parallel()
+			app, rest := ablytest.NewREST()
+			defer app.Close()
+
+			fixtures := statsFixtures()
+			postStats(app, fixtures)
+
+			err := ablytest.TestPagination(
+				reverseStats(fixtures),
+				rest.Stats(ably.StatsWithLimit(limit)),
+				limit,
+			)
+			if err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
+func TestStats_StartEnd_RSC6b1(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	app, rest := ablytest.NewREST()
+	defer app.Close()
+
+	fixtures := statsFixtures()
+	postStats(app, fixtures)
+
+	expected := reverseStats(fixtures[1:3])
+
+	pages, err := rest.Stats(
+		ably.StatsWithStart(time.Date(2020, time.January, 28, 14, 1, 0, 0, time.UTC)),
+		ably.StatsWithEnd(time.Date(2020, time.January, 28, 14, 2, 30, 0, time.UTC)),
+	).Pages(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got []*ably.Stats
+	for pages.Next(ctx) {
+		got = append(got, pages.Items()...)
+	}
+	if err := pages.Err(); err != nil {
+		t.Fatal(err)
+	}
+
+	if !reflect.DeepEqual(expected, got) {
+		t.Fatalf("expected: %+v; got: %+v", expected, got)
+	}
+}
+
+func TestStats_Direction_RSC6b2(t *testing.T) {
+	t.Parallel()
+
+	for _, c := range []struct {
+		direction ably.Direction
+		expected  []*ably.Stats
+	}{
+		{
+			direction: ably.Backwards,
+			expected:  reverseStats(statsFixtures()),
+		},
+		{
+			direction: ably.Forwards,
+			expected:  statsFixtures(),
+		},
+	} {
+		c := c
+		t.Run(fmt.Sprintf("direction=%v", c.direction), func(t *testing.T) {
+			ctx := context.Background()
+
+			app, rest := ablytest.NewREST()
+			defer app.Close()
+
+			fixtures := statsFixtures()
+			postStats(app, fixtures)
+
+			expected := c.expected
+
+			pages, err := rest.Stats(
+				ably.StatsWithLimit(len(expected)),
+				ably.StatsWithDirection(c.direction),
+			).Pages(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+			var got []*ably.Stats
+			for pages.Next(ctx) {
+				got = append(got, pages.Items()...)
+			}
+			if err := pages.Err(); err != nil {
+				t.Fatal(err)
+			}
+
+			if !reflect.DeepEqual(expected, got) {
+				t.Fatalf("expected: %+v; got: %+v", expected, got)
+			}
+		})
+	}
+}
+
+func TestStats_Unit_RSC6b4(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+
+	app, rest := ablytest.NewREST()
+	defer app.Close()
+
+	fixtures := statsFixtures()
+	postStats(app, fixtures)
+
+	pages, err := rest.Stats(
+		ably.StatsWithUnit(ably.PeriodMonth),
+	).Pages(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var got []*ably.Stats
+	for pages.Next(ctx) {
+		got = append(got, pages.Items()...)
+	}
+	if err := pages.Err(); err != nil {
+		t.Fatal(err)
+	}
+
+	if expected, got := 1, len(got); expected != got {
+		t.Fatalf("expected: %v; got: %v", expected, got)
+	}
+
+	stats := got[0]
+	if expected, got := "month", stats.Unit; expected != got {
+		t.Fatalf("expected: %v; got: %v", expected, got)
+	}
+}
+
+func statsFixtures() []*ably.Stats {
+	var fixtures []*ably.Stats
+	baseDate := time.Date(2020, time.January, 28, 14, 0, 0, 0, time.UTC)
+	msgCounts := ably.StatsMessageCount{
+		Count: 50,
+		Data:  5000,
+	}
+	msgTypes := ably.StatsMessageTypes{
+		All:      msgCounts,
+		Messages: msgCounts,
+	}
+	for i := time.Duration(0); i < 10; i++ {
+		fixtures = append(fixtures, &ably.Stats{
+			IntervalID: baseDate.Add(i * time.Minute).Format("2006-01-02:15:04"),
+			Unit:       "minute",
+			All:        msgTypes,
+			Inbound: ably.StatsMessageTraffic{
+				All:      msgTypes,
+				RealTime: msgTypes,
+			},
+		})
+	}
+	return fixtures
+}
+
+func postStats(app *ablytest.Sandbox, stats []*ably.Stats) error {
+	ctx, cancel := context.WithTimeout(context.Background(), ablytest.Timeout)
+	defer cancel()
+
+	statsJSON, err := json.Marshal(stats)
+	if err != nil {
+		return fmt.Errorf("marshaling stats: %w", err)
+	}
+
+	req, err := http.NewRequest("POST", "https://sandbox-rest.ably.io/stats", bytes.NewReader(statsJSON))
+	if err != nil {
+		return fmt.Errorf("creating request: %w", err)
+	}
+	req = req.WithContext(ctx)
+	req.SetBasicAuth(app.KeyParts())
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return fmt.Errorf("performing request: %w", err)
+	}
+	resp.Body.Close()
+	return nil
+}
+
+func reverseStats(stats []*ably.Stats) []*ably.Stats {
+	var reversed []*ably.Stats
+	for i := len(stats) - 1; i >= 0; i-- {
+		reversed = append(reversed, stats[i])
+	}
+	return reversed
+}