Repository: ably/ably-go
Pull request #292 (2021-03-08 15:42:35)

Adapt History to follow new paginated result design
Based on #281.

Mainly just copying what #281 did for stats, but also changed the way message data is encoded to require less tightly-coupled, reflect-based decoding.


Patch:
diff --git a/README.md b/README.md
index d0d14ee..b17ac97 100644
--- a/README.md
+++ b/README.md
@@ -248,15 +248,19 @@ if err != nil {
 -->
 
 ```go
-page, err := channel.History(ctx, nil)
-for ; err == nil && page != nil; page, err = page.Next(ctx) {
-	for _, message := range page.Messages() {
+pages, err := channel.History().Pages(ctx)
+if err != nil {
+	panic(err)
+}
+for pages.Next(ctx) {
+	for _, message := range pages.Items() {
 		fmt.Println(message)
 	}
 }
-if err != nil {
+if err := pages.Err(); err != nil {
 	panic(err)
 }
+
 ```
 
 <!-- GO EXAMPLE
@@ -292,13 +296,16 @@ if err != nil {
 -->
 
 ```go
-page, err := channel.Presence.History(ctx, nil)
-for ; err == nil && page != nil; page, err = page.Next(ctx) {
-	for _, presence := range page.PresenceMessages() {
+pages, err := channel.Presence.History().Pages(ctx)
+if err != nil {
+	panic(err)
+}
+for pages.Next(ctx) {
+	for _, presence := range pages.Items() {
 		fmt.Println(presence)
 	}
 }
-if err != nil {
+if err := pages.Err(); err != nil {
 	panic(err)
 }
 ```
diff --git a/ably/paginated_result.go b/ably/paginated_result.go
index f78713b..cfbffa3 100644
--- a/ably/paginated_result.go
+++ b/ably/paginated_result.go
@@ -104,6 +104,9 @@ func (p *PaginatedResultNew) loadItems(
 				return 0, false
 			}
 			pageLen = getLen()
+			if pageLen == 0 { // compatible with hasNext if first page is empty
+				return 0, false
+			}
 		}
 
 		idx := nextItem
@@ -215,52 +218,11 @@ type paginatedRequest struct {
 }
 
 func decodePaginatedResult(opts *proto.ChannelOptions, typ reflect.Type, resp *http.Response) (interface{}, error) {
-	switch typ {
-	case msgType:
-		var o []map[string]interface{}
-		err := decodeResp(resp, &o)
-		if err != nil {
-			return nil, err
-		}
-		rs := make([]*proto.Message, len(o))
-		for k, v := range o {
-			m := &proto.Message{
-				ChannelOptions: opts,
-			}
-			err = m.FromMap(v)
-			if err != nil {
-				return nil, err
-			}
-			rs[k] = m
-		}
-		return rs, nil
-	case presMsgType:
-		var o []map[string]interface{}
-		err := decodeResp(resp, &o)
-		if err != nil {
-			return nil, err
-		}
-		rs := make([]*proto.PresenceMessage, len(o))
-		for k, v := range o {
-			m := &proto.PresenceMessage{
-				Message: proto.Message{
-					ChannelOptions: opts,
-				},
-			}
-			err = m.FromMap(v)
-			if err != nil {
-				return nil, err
-			}
-			rs[k] = m
-		}
-		return rs, nil
-	default:
-		v := reflect.New(typ)
-		if err := decodeResp(resp, v.Interface()); err != nil {
-			return nil, err
-		}
-		return v.Elem().Interface(), nil
+	v := reflect.New(typ)
+	if err := decodeResp(resp, v.Interface()); err != nil {
+		return nil, err
 	}
+	return v.Elem().Interface(), nil
 }
 
 func newPaginatedResult(ctx context.Context, opts *proto.ChannelOptions, req paginatedRequest) (*PaginatedResult, error) {
@@ -365,16 +327,6 @@ func (p *PaginatedResult) Items() []interface{} {
 	return p.items
 }
 
-// Messages gives a slice of messages for the current page. The method panics if
-// the underlying paginated result is not a message.
-func (p *PaginatedResult) Messages() []*Message {
-	items, ok := p.typItems.([]*proto.Message)
-	if !ok {
-		panic(errInvalidType{typ: p.req.typ})
-	}
-	return items
-}
-
 // PresenceMessages gives a slice of presence messages for the current path.
 // The method panics if the underlying paginated result is not a presence message.
 func (p *PaginatedResult) PresenceMessages() []*PresenceMessage {
diff --git a/ably/proto/crypto.go b/ably/proto/crypto.go
index 13aebcc..525ec45 100644
--- a/ably/proto/crypto.go
+++ b/ably/proto/crypto.go
@@ -81,6 +81,9 @@ type ChannelOptions struct {
 // GetCipher returns a ChannelCipher based on the algorithms set in the
 // ChannelOptions.CipherParams.
 func (c *ChannelOptions) GetCipher() (ChannelCipher, error) {
+	if c == nil {
+		return nil, errors.New("no cipher configured")
+	}
 	if c.cipher != nil {
 		return c.cipher, nil
 	}
diff --git a/ably/proto/message.go b/ably/proto/message.go
index 89c72b8..2667918 100644
--- a/ably/proto/message.go
+++ b/ably/proto/message.go
@@ -1,15 +1,13 @@
 package proto
 
 import (
+	"bytes"
 	"crypto/aes"
 	"encoding/base64"
 	"encoding/json"
-	"errors"
 	"fmt"
-	"reflect"
 	"strings"
-
-	"github.com/ugorji/go/codec"
+	"unicode/utf8"
 )
 
 // encodings
@@ -21,309 +19,117 @@ const (
 )
 
 type Message struct {
-	ID              string                 `json:"id,omitempty" codec:"id,omitempty"`
-	ClientID        string                 `json:"clientId,omitempty" codec:"clientId,omitempty"`
-	ConnectionID    string                 `json:"connectionId,omitempty" codec:"connectionID,omitempty"`
-	Name            string                 `json:"name,omitempty" codec:"name,omitempty"`
-	Data            interface{}            `json:"data,omitempty" codec:"data,omitempty"`
-	Encoding        string                 `json:"encoding,omitempty" codec:"encoding,omitempty"`
-	Timestamp       int64                  `json:"timestamp" codec:"timestamp"`
-	Extras          map[string]interface{} `json:"extras" codec:"extras"`
-	*ChannelOptions `json:"-" codec:"-"`
+	ID           string                 `json:"id,omitempty" codec:"id,omitempty"`
+	ClientID     string                 `json:"clientId,omitempty" codec:"clientId,omitempty"`
+	ConnectionID string                 `json:"connectionId,omitempty" codec:"connectionID,omitempty"`
+	Name         string                 `json:"name,omitempty" codec:"name,omitempty"`
+	Data         interface{}            `json:"data,omitempty" codec:"data,omitempty"`
+	Encoding     string                 `json:"encoding,omitempty" codec:"encoding,omitempty"`
+	Timestamp    int64                  `json:"timestamp,omitempty" codec:"timestamp,omitempty"`
+	Extras       map[string]interface{} `json:"extras,omitempty" codec:"extras,omitempty"`
 }
 
-func (m *Message) maybeJSONEncode() error {
-	if m.Data == nil {
-		return nil
-	}
-	switch m.Data.(type) {
-	case json.Marshaler:
-		bs, err := m.Data.(json.Marshaler).MarshalJSON()
-		if err != nil {
-			return err
-		}
-		m.Data = string(bs)
-		m.Encoding = mergeEncoding(m.Encoding, JSON)
-		return nil
-	case map[string]interface{}:
-		bs, err := json.Marshal(m.Data)
-		if err != nil {
-			return err
-		}
-		m.Data = string(bs)
-		m.Encoding = mergeEncoding(m.Encoding, JSON)
-		return nil
-	}
-	dataType := reflect.TypeOf(m.Data)
-
-	// marshal any sort of slice except for []byte (i.e. []uint8)
-	if dataType.Kind() == reflect.Slice && dataType.Elem().Kind() != reflect.Uint8 {
-		bs, err := json.Marshal(m.Data)
-		if err != nil {
-			return err
-		}
-		m.Data = string(bs)
-		m.Encoding = mergeEncoding(m.Encoding, JSON)
-	}
-	return nil
+func (m Message) String() string {
+	return fmt.Sprintf("<Message %q data=%v>", m.Name, m.Data)
 }
 
-func (m Message) HasCipher() bool {
-	if m.ChannelOptions != nil {
-		c, _ := m.ChannelOptions.GetCipher()
-		return c != nil
-	}
-	return false
+func unencodableDataErr(data interface{}) error {
+	return fmt.Errorf("message data type %T must be string, []byte, or a value that can be encoded as a JSON object or array", data)
 }
 
-func (m Message) encode() (Message, error) {
+func (m Message) WithEncodedData(cipher ChannelCipher) (Message, error) {
+	// TODO: Unexport once proto gets merged into package ably.
 	if m.Data == nil {
 		return m, nil
 	}
-	err := m.maybeJSONEncode()
-	if err != nil {
-		return m, err
-	}
-	switch m.Data.(type) {
+
+	switch d := m.Data.(type) {
 	case string:
-		if m.HasCipher() {
+		if utf8.ValidString(d) {
 			m.Encoding = mergeEncoding(m.Encoding, UTF8)
-		}
-	case []byte:
-		// ok
-	default:
-		return Message{}, errors.New("unsupported payload type")
-	}
-	if m.ChannelOptions != nil {
-		if cipher, err := m.GetCipher(); err == nil {
-			// since we know that m.Data is either []byte or string at this point, coerceBytes is always
-			// safe here
-			bs, err := coerceBytes(m.Data)
-			if err != nil {
-				return Message{}, err
-			}
-			e, err := cipher.Encrypt(bs)
-			if err != nil {
-				return Message{}, err
-			}
-			m.Data = e
-			m.Encoding = mergeEncoding(m.Encoding, cipher.GetAlgorithm())
+		} else {
+			// If string isn't UTF-8, convert to []byte to encode it as base64
+			// below.
+			m.Data = []byte(d)
 		}
 	}
-
-	return m, nil
-}
-
-func (m Message) encodeJSON() (Message, error) {
-	e, err := m.encode()
-	if err != nil {
-		return m, err
-	}
-	if d, ok := e.Data.([]byte); ok {
-		b := base64.StdEncoding.EncodeToString(d)
-		e.Data = b
-		e.Encoding = mergeEncoding(e.Encoding, Base64)
-	}
-	return e, nil
-}
-
-func coerceString(i interface{}) (string, error) {
-	switch v := i.(type) {
-	case []byte:
-		return string(v), nil
+	switch d := m.Data.(type) {
 	case string:
-		return v, nil
-	default:
-		return "", fmt.Errorf("UTF8 encoding can only handle types string or []byte, but got %T", i)
-	}
-}
-
-func coerceBytes(i interface{}) ([]byte, error) {
-	switch v := i.(type) {
 	case []byte:
-		return v, nil
-	case string:
-		return []byte(v), nil
+		m.Data = base64.StdEncoding.EncodeToString(d)
+		m.Encoding = mergeEncoding(m.Encoding, Base64)
 	default:
-		return nil, fmt.Errorf("coerceBytes should never get anything but strings or []byte, got %T", i)
-	}
-}
-
-// ToMap returns a map of all message field names to their respective value if
-// the field are set.
-//
-// Spec RSL1j
-func (m Message) ToMap() map[string]interface{} {
-	ctx := make(map[string]interface{})
-	if m.ID != "" {
-		ctx["id"] = m.ID
-	}
-	if m.ClientID != "" {
-		ctx["clientId"] = m.ClientID
-	}
-	if m.ConnectionID != "" {
-		ctx["connectionId"] = m.ConnectionID
-	}
-	if m.Name != "" {
-		ctx["name"] = m.Name
-	}
-	encoding := m.Encoding
-	if m.Data != nil {
-		ctx["data"] = m.Data
-	}
-	if encoding != "" {
-		ctx["encoding"] = encoding
-	}
-	if m.Timestamp != 0 {
-		ctx["timestamp"] = m.Timestamp
-	}
-	if m.Extras != nil {
-		ctx["extras"] = m.Extras
-	}
-	return ctx
-}
-
-func (m Message) MarshalJSON() ([]byte, error) {
-	e, err := m.encodeJSON()
-	if err != nil {
-		return nil, err
-	}
-	return json.Marshal(e.ToMap())
-}
-
-func (m *Message) UnmarshalJSON(data []byte) error {
-	var ctx map[string]interface{}
-	if err := json.Unmarshal(data, &ctx); err != nil {
-		return err
-	}
-	return m.FromMap(ctx)
-}
-
-func (m Message) CodecEncodeSelf(encoder *codec.Encoder) {
-	e, err := m.encode()
-	if err != nil {
-		panic(err)
-	}
-	encoder.MustEncode(e.ToMap())
-}
-
-// CodecDecodeSelf implements codec.Selfer interface for msgpack decoding.
-func (m *Message) CodecDecodeSelf(decoder *codec.Decoder) {
-	ctx := make(map[string]interface{})
-	decoder.MustDecode(&ctx)
-	m.FromMap(ctx)
-}
-
-func (m *Message) FromMap(ctx map[string]interface{}) error {
-	if v, ok := ctx["id"]; ok {
-		x, err := coerceString(v)
+		// RSL4c3, RSL4d3: JSON is only for objects and arrays. So marshal data
+		// into JSON, then check if if's one of those.
+		b, err := json.Marshal(d)
 		if err != nil {
-			return err
+			return Message{}, fmt.Errorf("%s; encoding as JSON: %w", unencodableDataErr(d), err)
 		}
-		m.ID = string(x)
-	}
-	if v, ok := ctx["clientId"]; ok {
-		x, err := coerceString(v)
-		if err != nil {
-			return err
-		}
-		m.ClientID = string(x)
-	}
-	if v, ok := ctx["connectionId"]; ok {
-		x, err := coerceString(v)
-		if err != nil {
-			return err
+		token, _ := json.NewDecoder(bytes.NewReader(b)).Token()
+		if token != json.Delim('[') && token != json.Delim('{') {
+			return Message{}, fmt.Errorf("%s; encoded as JSON %T", unencodableDataErr(d), token)
 		}
-		m.ConnectionID = string(x)
-	}
-	if v, ok := ctx["name"]; ok {
-		x, err := coerceString(v)
-		if err != nil {
-			return err
-		}
-		m.Name = string(x)
-	}
-	if v, ok := ctx["encoding"]; ok {
-		x, err := coerceString(v)
-		if err != nil {
-			return err
-		}
-		m.Encoding = string(x)
-	}
-	if v, ok := ctx["data"]; ok {
-		m.Data = v
-		dec, err := m.decode()
-		if err != nil {
-			return err
-		}
-		*m = dec
-	}
-	if v, ok := ctx["timestamp"]; ok {
-		switch e := v.(type) {
-		case float64:
-			m.Timestamp = int64(e)
-		case uint64:
-			m.Timestamp = int64(e)
-		case int64:
-			m.Timestamp = e
-		}
-	}
-	if v, ok := ctx["extras"]; ok {
-		m.Extras = v.(map[string]interface{})
+		m.Data = string(b)
+		m.Encoding = mergeEncoding(m.Encoding, JSON)
 	}
-	return nil
-}
-
-// MemberKey returns string that allows to uniquely identify connected clients.
-func (m *Message) MemberKey() string {
-	return m.ConnectionID + ":" + m.ClientID
-}
 
-func (m Message) Decrypt() (interface{}, error) {
-	cipher, err := m.GetCipher()
-	if err != nil {
-		return nil, err
+	if cipher == nil {
+		return m, nil
 	}
 
-	d, err := coerceBytes(m.Data)
+	// since we know that m.Data is either []byte or string at this point, coerceBytes is always
+	// safe here
+	bs, err := coerceBytes(m.Data)
 	if err != nil {
-		return nil, err
+		panic(err)
 	}
-	v, err := cipher.Decrypt(d)
+	e, err := cipher.Encrypt(bs)
 	if err != nil {
-		return nil, err
+		return Message{}, fmt.Errorf("encrypting message data: %w", err)
+	}
+	if _, ok := m.Data.(string); ok {
+		// Not sure why this is useful, and can't find the requirement in the
+		// spec, but test fixtures expect this.
+		m.Encoding = mergeEncoding(m.Encoding, UTF8)
 	}
-	return v, nil
+	m.Data = base64.StdEncoding.EncodeToString(e)
+	m.Encoding = mergeEncoding(m.Encoding, cipher.GetAlgorithm())
+	m.Encoding = mergeEncoding(m.Encoding, Base64)
+	return m, nil
 }
 
-func (m Message) decode() (Message, error) {
+func (m Message) WithDecodedData(cipher ChannelCipher) (Message, error) {
+	// TODO: Unexport once proto gets merged into package ably.
+
 	// strings.Split on empty string returns []string{""}
 	if m.Data == nil || m.Encoding == "" {
 		return m, nil
 	}
 	encodings := strings.Split(m.Encoding, "/")
-	for i := len(encodings) - 1; i >= 0; i-- {
-		switch encodings[i] {
+	for len(encodings) > 0 {
+		encoding := encodings[len(encodings)-1]
+		encodings = encodings[:len(encodings)-1]
+		switch encoding {
 		case Base64:
 			d, err := coerceString(m.Data)
 			if err != nil {
-				return Message{}, err
+				return m, err
 			}
 			data, err := base64.StdEncoding.DecodeString(d)
 			if err != nil {
-				return Message{}, err
+				return m, err
 			}
 			m.Data = data
 		case UTF8:
 			d, err := coerceString(m.Data)
 			if err != nil {
-				return Message{}, err
+				return m, err
 			}
 			m.Data = d
 		case JSON:
 			d, err := coerceBytes(m.Data)
 			if err != nil {
-				return Message{}, err
+				return m, err
 			}
 			var result interface{}
 			if err := json.Unmarshal(d, &result); err != nil {
@@ -331,22 +137,55 @@ func (m Message) decode() (Message, error) {
 			}
 			m.Data = result
 		default:
-			switch {
-			case strings.HasPrefix(encodings[i], Cipher):
-				d, err := m.Decrypt()
+			if strings.HasPrefix(encoding, Cipher) {
+				if cipher == nil {
+					return m, fmt.Errorf("message data is encrypted as %s, but cipher wasn't provided", encoding)
+				}
+				d, err := coerceBytes(m.Data)
 				if err != nil {
 					return m, err
 				}
+				d, err = cipher.Decrypt(d)
+				if err != nil {
+					return m, fmt.Errorf("decrypting message data: %w", err)
+				}
 				m.Data = d
-			default:
-				return m, fmt.Errorf("unknown encoding %s", encodings[i])
+			} else {
+				return m, fmt.Errorf("unknown encoding %s", encoding)
 			}
-
 		}
+		m.Encoding = strings.Join(encodings, "/")
 	}
 	return m, nil
 }
 
+func coerceString(i interface{}) (string, error) {
+	switch v := i.(type) {
+	case []byte:
+		return string(v), nil
+	case string:
+		return v, nil
+	default:
+		return "", fmt.Errorf("UTF8 encoding can only handle types string or []byte, but got %T", i)
+	}
+}
+
+func coerceBytes(i interface{}) ([]byte, error) {
+	switch v := i.(type) {
+	case []byte:
+		return v, nil
+	case string:
+		return []byte(v), nil
+	default:
+		return nil, fmt.Errorf("coerceBytes should never get anything but strings or []byte, got %T", i)
+	}
+}
+
+// MemberKey returns string that allows to uniquely identify connected clients.
+func (m *Message) MemberKey() string {
+	return m.ConnectionID + ":" + m.ClientID
+}
+
 func addPadding(src []byte) []byte {
 	padlen := byte(aes.BlockSize - (len(src) % aes.BlockSize))
 	data := make([]byte, len(src)+int(padlen))
diff --git a/ably/proto/presence_message.go b/ably/proto/presence_message.go
index 78b423e..ff40e2c 100644
--- a/ably/proto/presence_message.go
+++ b/ably/proto/presence_message.go
@@ -1,9 +1,7 @@
 package proto
 
 import (
-	"encoding/json"
-
-	"github.com/ugorji/go/codec"
+	"fmt"
 )
 
 type PresenceAction int64
@@ -21,57 +19,12 @@ type PresenceMessage struct {
 	Action PresenceAction `json:"action" codec:"action"`
 }
 
-func (m PresenceMessage) MarshalJSON() ([]byte, error) {
-	e, err := m.encodeJSON()
-	if err != nil {
-		return nil, err
-	}
-	ctx := e.ToMap()
-	ctx["action"] = m.Action
-	return json.Marshal(ctx)
-}
-
-func (m *PresenceMessage) UnmarshalJSON(data []byte) error {
-	var ctx map[string]interface{}
-	if err := json.Unmarshal(data, &ctx); err != nil {
-		return err
-	}
-	return m.FromMap(ctx)
-}
-
-// CodecEncodeSelf encodes PresenceMessage into a msgpack format.
-func (m PresenceMessage) CodecEncodeSelf(encoder *codec.Encoder) {
-	e, err := m.encode()
-	if err != nil {
-		panic(err)
-	}
-	ctx := e.ToMap()
-	ctx["action"] = m.Action
-	encoder.MustEncode(ctx)
-}
-
-// CodecDecodeSelf implements codec.Selfer interface for msgpack decoding.
-func (m *PresenceMessage) CodecDecodeSelf(decoder *codec.Decoder) {
-	ctx := make(map[string]interface{})
-	decoder.MustDecode(&ctx)
-	if err := m.FromMap(ctx); err != nil {
-		panic(err)
-	}
-}
-
-func (m *PresenceMessage) FromMap(ctx map[string]interface{}) error {
-	msg := &m.Message
-	if err := msg.FromMap(ctx); err != nil {
-		return err
-	}
-	if v, ok := ctx["action"]; ok {
-		m.Action = PresenceAction(coerceInt64(v))
-	}
-	return nil
-}
-
-func (m PresenceMessage) ToMap() map[string]interface{} {
-	ctx := m.Message.ToMap()
-	ctx["action"] = m.Action
-	return ctx
+func (m PresenceMessage) String() string {
+	return fmt.Sprintf("<PresenceMessage %v data=%v>", [...]string{
+		"absent",
+		"present",
+		"enter",
+		"leave",
+		"update",
+	}[m.Action], m.Data)
 }
diff --git a/ably/proto/protocol_message.go b/ably/proto/protocol_message.go
index a9db03b..a543b45 100644
--- a/ably/proto/protocol_message.go
+++ b/ably/proto/protocol_message.go
@@ -1,7 +1,6 @@
 package proto
 
 import (
-	"encoding/json"
 	"fmt"
 	"time"
 )
@@ -93,81 +92,6 @@ type ProtocolMessage struct {
 	Flags             Flag               `json:"flags,omitempty" codec:"flags,omitempty"`
 }
 
-func (p *ProtocolMessage) UnmarshalJSON(b []byte) error {
-	ctx := make(map[string]interface{})
-	err := json.Unmarshal(b, &ctx)
-	if err != nil {
-		return err
-	}
-	p.FromMap(ctx)
-	return nil
-}
-
-func (p *ProtocolMessage) FromMap(ctx map[string]interface{}) {
-	if v, ok := ctx["messages"]; ok {
-		i := v.([]interface{})
-		for _, v := range i {
-			var msg Message
-			msg.FromMap(v.(map[string]interface{}))
-			p.Messages = append(p.Messages, &msg)
-		}
-	}
-	if v, ok := ctx["presence"]; ok {
-		i := v.([]interface{})
-		for _, v := range i {
-			var msg PresenceMessage
-			msg.FromMap(v.(map[string]interface{}))
-			p.Presence = append(p.Presence, &msg)
-		}
-	}
-	if v, ok := ctx["id"]; ok {
-		p.ID = v.(string)
-	}
-	if v, ok := ctx["applicationId"]; ok {
-		p.ApplicationID = v.(string)
-	}
-	if v, ok := ctx["connectionId"]; ok {
-		p.ConnectionID = v.(string)
-	}
-	if v, ok := ctx["connectionKey"]; ok {
-		p.ConnectionKey = v.(string)
-	}
-	if v, ok := ctx["channel"]; ok {
-		p.Channel = v.(string)
-	}
-	if v, ok := ctx["channelSerial"]; ok {
-		p.ChannelSerial = v.(string)
-	}
-	if v, ok := ctx["connectionDetails"]; ok {
-		c := &ConnectionDetails{}
-		c.FromMap(v.(map[string]interface{}))
-		p.ConnectionDetails = c
-	}
-	if v, ok := ctx["error"]; ok {
-		c := &ErrorInfo{}
-		c.FromMap(v.(map[string]interface{}))
-		p.Error = c
-	}
-	if v, ok := ctx["msgSerial"]; ok {
-		p.MsgSerial = coerceInt64(v)
-	}
-	if v, ok := ctx["connectionSerial"]; ok {
-		p.ConnectionSerial = coerceInt64(v)
-	}
-	if v, ok := ctx["timestamp"]; ok {
-		p.Timestamp = coerceInt64(v)
-	}
-	if v, ok := ctx["count"]; ok {
-		p.Count = int(coerceInt64(v))
-	}
-	if v, ok := ctx["action"]; ok {
-		p.Action = Action(coerceInt8(v))
-	}
-	if v, ok := ctx["flags"]; ok {
-		p.Flags = Flag(coerceInt64(v))
-	}
-}
-
 func (msg *ProtocolMessage) String() string {
 	switch msg.Action {
 	case ActionHeartbeat:
diff --git a/ably/realtime_channel.go b/ably/realtime_channel.go
index 9ebecb5..7a1a676 100644
--- a/ably/realtime_channel.go
+++ b/ably/realtime_channel.go
@@ -431,11 +431,9 @@ func (c *RealtimeChannel) PublishBatch(ctx context.Context, messages []*Message)
 	return res.Wait(ctx)
 }
 
-// History gives the channel's message history according to the given parameters.
-// The returned result can be inspected for the messages via the Messages()
-// method.
-func (c *RealtimeChannel) History(ctx context.Context, params *PaginateParams) (*PaginatedResult, error) {
-	return c.client.rest.Channels.Get(c.Name).History(ctx, params)
+// History is equivalent to RESTChannel.History.
+func (c *RealtimeChannel) History(o ...HistoryOption) HistoryRequest {
+	return c.client.rest.Channels.Get(c.Name).History(o...)
 }
 
 func (c *RealtimeChannel) send(msg *proto.ProtocolMessage) (result, error) {
diff --git a/ably/rest_channel.go b/ably/rest_channel.go
index 41000ff..d35613d 100644
--- a/ably/rest_channel.go
+++ b/ably/rest_channel.go
@@ -2,11 +2,15 @@ package ably
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/url"
+	"strconv"
 	"strings"
+	"time"
 
 	"github.com/ably/ably-go/ably/internal/ablyutil"
+	"github.com/ugorji/go/codec"
 
 	"github.com/ably/ably-go/ably/proto"
 )
@@ -81,18 +85,16 @@ func PublishBatchWithParams(params map[string]string) PublishBatchOption {
 
 // PublishBatchWithOptions is PublishBatch with optional parameters.
 func (c *RESTChannel) PublishBatchWithOptions(ctx context.Context, messages []*Message, options ...PublishBatchOption) error {
-	// TODO: Use context
 	var publishOpts publishBatchOptions
 	for _, o := range options {
 		o(&publishOpts)
 	}
-	msgPtrs := make([]*proto.Message, 0, len(messages))
-	for _, m := range messages {
-		msgPtrs = append(msgPtrs, (*proto.Message)(m))
-	}
-	if c.options != nil {
-		for _, v := range messages {
-			v.ChannelOptions = c.options
+	for i, m := range messages {
+		cipher, _ := c.options.GetCipher()
+		var err error
+		*m, err = (*m).WithEncodedData(cipher)
+		if err != nil {
+			return fmt.Errorf("encoding data for message #%d: %w", i, err)
 		}
 	}
 	useIdempotent := c.client.opts.idempotentRESTPublishing()
@@ -141,16 +143,179 @@ func (c *RESTChannel) PublishBatchWithOptions(ctx context.Context, messages []*M
 	return res.Body.Close()
 }
 
-// History gives the channel's message history according to the given parameters.
-// The returned result can be inspected for the messages via the Messages()
-// method.
-func (c *RESTChannel) History(ctx context.Context, params *PaginateParams) (*PaginatedResult, error) {
-	path := c.baseURL + "/history"
-	rst, err := newPaginatedResult(ctx, c.options, paginatedRequest{typ: msgType, path: path, params: params, query: query(c.client.get), logger: c.logger(), respCheck: checkValidHTTPResponse})
+// History gives the channel's message history.
+func (c *RESTChannel) History(o ...HistoryOption) HistoryRequest {
+	params := (&historyOptions{}).apply(o...)
+	return HistoryRequest{
+		r:       c.client.newPaginatedRequest("/channels/"+c.Name+"/history", params),
+		channel: c,
+	}
+}
+
+// A HistoryOption configures a call to RESTChannel.History or RealtimeChannel.History.
+type HistoryOption func(*historyOptions)
+
+func HistoryWithStart(t time.Time) HistoryOption {
+	return func(o *historyOptions) {
+		o.params.Set("start", strconv.FormatInt(unixMilli(t), 10))
+	}
+}
+
+func HistoryWithEnd(t time.Time) HistoryOption {
+	return func(o *historyOptions) {
+		o.params.Set("end", strconv.FormatInt(unixMilli(t), 10))
+	}
+}
+
+func HistoryWithLimit(limit int) HistoryOption {
+	return func(o *historyOptions) {
+		o.params.Set("limit", strconv.Itoa(limit))
+	}
+}
+
+func HistoryWithDirection(d Direction) HistoryOption {
+	return func(o *historyOptions) {
+		o.params.Set("direction", string(d))
+	}
+}
+
+type historyOptions struct {
+	params url.Values
+}
+
+func (o *historyOptions) apply(opts ...HistoryOption) url.Values {
+	o.params = make(url.Values)
+	for _, opt := range opts {
+		opt(o)
+	}
+	return o.params
+}
+
+// HistoryRequest represents a request prepared by the RESTChannel.History or
+// RealtimeChannel.History method, ready to be performed by its Pages or Items methods.
+type HistoryRequest struct {
+	r       paginatedRequestNew
+	channel *RESTChannel
+}
+
+// Pages returns an iterator for whole pages of History.
+//
+// See "Paginated results" section in the package-level documentation.
+func (r HistoryRequest) Pages(ctx context.Context) (*MessagesPaginatedResult, error) {
+	var res MessagesPaginatedResult
+	return &res, res.load(ctx, r.r)
+}
+
+// A MessagesPaginatedResult is an iterator for the result of a History request.
+//
+// See "Paginated results" section in the package-level documentation.
+type MessagesPaginatedResult struct {
+	PaginatedResultNew
+	items []*Message
+}
+
+// Next retrieves the next page of results.
+//
+// See the "Paginated results" section in the package-level documentation.
+func (p *MessagesPaginatedResult) 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 *MessagesPaginatedResult) Items() []*Message {
+	return p.items
+}
+
+// Items returns a convenience iterator for single History, over an underlying
+// paginated iterator.
+//
+// See "Paginated results" section in the package-level documentation.
+func (r HistoryRequest) Items(ctx context.Context) (*MessagesPaginatedItems, error) {
+	var res MessagesPaginatedItems
+	var err error
+	res.next, err = res.loadItems(ctx, r.r, func() (interface{}, func() int) {
+		res.items = nil // avoid mutating already returned Items
+		return r.channel.fullMessagesDecoder(&res.items), func() int {
+			return len(res.items)
+		}
+	})
+	return &res, err
+}
+
+// fullMessagesDecoder wraps a destination slice of messages in a decoder value
+// that decodes both the message itself from the transport-level encoding and
+// the data field within from its message-specific encoding.
+func (c *RESTChannel) fullMessagesDecoder(dst *[]*Message) interface{} {
+	return &fullMessagesDecoder{dst: dst, c: c}
+}
+
+type fullMessagesDecoder struct {
+	dst *[]*Message
+	c   *RESTChannel
+}
+
+func (t *fullMessagesDecoder) UnmarshalJSON(b []byte) error {
+	err := json.Unmarshal(b, &t.dst)
 	if err != nil {
-		return nil, err
+		return err
 	}
-	return rst, nil
+	t.decodeMessagesData()
+	return nil
+}
+
+func (t *fullMessagesDecoder) CodecEncodeSelf(*codec.Encoder) {
+	panic("messagesDecoderForChannel cannot be used as encoder")
+}
+
+func (t *fullMessagesDecoder) CodecDecodeSelf(decoder *codec.Decoder) {
+	decoder.MustDecode(&t.dst)
+	t.decodeMessagesData()
+}
+
+var _ interface {
+	json.Unmarshaler
+	codec.Selfer
+} = (*fullMessagesDecoder)(nil)
+
+func (t *fullMessagesDecoder) decodeMessagesData() {
+	cipher, _ := t.c.options.GetCipher()
+	for _, m := range *t.dst {
+		var err error
+		*m, err = m.WithDecodedData(cipher)
+		if err != nil {
+			// RSL6b
+			t.c.logger().sugar().Errorf("Couldn't fully decode message data from channel %q: %w", t.c.Name, err)
+		}
+	}
+}
+
+type MessagesPaginatedItems struct {
+	PaginatedResultNew
+	items []*Message
+	item  *Message
+	next  func(context.Context) (int, bool)
+}
+
+// Next retrieves the next result.
+//
+// See the "Paginated results" section in the package-level documentation.
+func (p *MessagesPaginatedItems) 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 *MessagesPaginatedItems) Item() *Message {
+	return p.item
 }
 
 func (c *RESTChannel) logger() *LoggerOptions {
diff --git a/ably/rest_client.go b/ably/rest_client.go
index 0a876b6..0a3ecbb 100644
--- a/ably/rest_client.go
+++ b/ably/rest_client.go
@@ -677,5 +677,7 @@ func decodeResp(resp *http.Response, out interface{}) error {
 	if err != nil {
 		return err
 	}
-	return decode(typ, resp.Body, out)
+	b, _ := ioutil.ReadAll(resp.Body)
+
+	return decode(typ, bytes.NewReader(b), out)
 }
diff --git a/ably/rest_presence.go b/ably/rest_presence.go
index 0adad4b..a67d67d 100644
--- a/ably/rest_presence.go
+++ b/ably/rest_presence.go
@@ -1,6 +1,14 @@
 package ably
 
-import "context"
+import (
+	"context"
+	"encoding/json"
+	"net/url"
+	"strconv"
+	"time"
+
+	"github.com/ugorji/go/codec"
+)
 
 type RESTPresence struct {
 	client  *REST
@@ -15,14 +23,181 @@ func (p *RESTPresence) Get(ctx context.Context, params *PaginateParams) (*Pagina
 	return newPaginatedResult(ctx, nil, paginatedRequest{typ: presMsgType, path: path, params: params, query: query(p.client.get), logger: p.logger(), respCheck: checkValidHTTPResponse})
 }
 
-// History gives the channel's presence messages history according to the given
-// parameters. The returned result can be inspected for the presence messages
-// via the PresenceMessages() method.
-func (p *RESTPresence) History(ctx context.Context, params *PaginateParams) (*PaginatedResult, error) {
-	path := p.channel.baseURL + "/presence/history"
-	return newPaginatedResult(ctx, nil, paginatedRequest{typ: presMsgType, path: path, params: params, query: query(p.client.get), logger: p.logger(), respCheck: checkValidHTTPResponse})
-}
-
 func (p *RESTPresence) logger() *LoggerOptions {
 	return p.client.logger()
 }
+
+// History gives the channel's presence history.
+func (c *RESTPresence) History(o ...PresenceHistoryOption) PresenceHistoryRequest {
+	params := (&presenceHistoryOptions{}).apply(o...)
+	return PresenceHistoryRequest{
+		r:       c.client.newPaginatedRequest("/channels/"+c.channel.Name+"/presence/history", params),
+		channel: c.channel,
+	}
+}
+
+// A HistoryOption configures a call to RESTChannel.History or RealtimeChannel.History.
+type PresenceHistoryOption func(*presenceHistoryOptions)
+
+func PresenceHistoryWithStart(t time.Time) PresenceHistoryOption {
+	return func(o *presenceHistoryOptions) {
+		o.params.Set("start", strconv.FormatInt(unixMilli(t), 10))
+	}
+}
+
+func PresenceHistoryWithEnd(t time.Time) PresenceHistoryOption {
+	return func(o *presenceHistoryOptions) {
+		o.params.Set("end", strconv.FormatInt(unixMilli(t), 10))
+	}
+}
+
+func PresenceHistoryWithLimit(limit int) PresenceHistoryOption {
+	return func(o *presenceHistoryOptions) {
+		o.params.Set("limit", strconv.Itoa(limit))
+	}
+}
+
+func PresenceHistoryWithDirection(d Direction) PresenceHistoryOption {
+	return func(o *presenceHistoryOptions) {
+		o.params.Set("direction", string(d))
+	}
+}
+
+type presenceHistoryOptions struct {
+	params url.Values
+}
+
+func (o *presenceHistoryOptions) apply(opts ...PresenceHistoryOption) url.Values {
+	o.params = make(url.Values)
+	for _, opt := range opts {
+		opt(o)
+	}
+	return o.params
+}
+
+// PresenceHistoryRequest represents a request prepared by the RESTPresence.History or
+// RealtimePresence.History method, ready to be performed by its Pages or Items methods.
+type PresenceHistoryRequest struct {
+	r       paginatedRequestNew
+	channel *RESTChannel
+}
+
+// Pages returns an iterator for whole pages of presence messages.
+//
+// See "Paginated results" section in the package-level documentation.
+func (r PresenceHistoryRequest) Pages(ctx context.Context) (*PresencePaginatedResult, error) {
+	res := PresencePaginatedResult{decoder: r.channel.fullPresenceDecoder}
+	return &res, res.load(ctx, r.r)
+}
+
+// A PresencePaginatedResult is an iterator for the result of a PresenceHistory request.
+//
+// See "Paginated results" section in the package-level documentation.
+type PresencePaginatedResult struct {
+	PaginatedResultNew
+	items   []*PresenceMessage
+	decoder func(*[]*PresenceMessage) interface{}
+}
+
+// Next retrieves the next page of results.
+//
+// See the "Paginated results" section in the package-level documentation.
+func (p *PresencePaginatedResult) Next(ctx context.Context) bool {
+	p.items = nil // avoid mutating already returned items
+	return p.next(ctx, p.decoder(&p.items))
+}
+
+// Items returns the current page of results.
+//
+// See the "Paginated results" section in the package-level documentation.
+func (p *PresencePaginatedResult) Items() []*PresenceMessage {
+	return p.items
+}
+
+// Items returns a convenience iterator for single PresenceHistory, over an underlying
+// paginated iterator.
+//
+// See "Paginated results" section in the package-level documentation.
+func (r PresenceHistoryRequest) Items(ctx context.Context) (*PresencePaginatedItems, error) {
+	var res PresencePaginatedItems
+	var err error
+	res.next, err = res.loadItems(ctx, r.r, func() (interface{}, func() int) {
+		return r.channel.fullPresenceDecoder(&res.items), func() int {
+			return len(res.items)
+		}
+	})
+	return &res, err
+}
+
+// fullPresenceDecoder wraps a destination slice of messages in a decoder value
+// that decodes both the message itself from the transport-level encoding and
+// the data field within from its message-specific encoding.
+func (c *RESTChannel) fullPresenceDecoder(dst *[]*PresenceMessage) interface{} {
+	return &fullPresenceDecoder{dst: dst, c: c}
+}
+
+type fullPresenceDecoder struct {
+	dst *[]*PresenceMessage
+	c   *RESTChannel
+}
+
+func (t *fullPresenceDecoder) UnmarshalJSON(b []byte) error {
+	err := json.Unmarshal(b, &t.dst)
+	if err != nil {
+		return err
+	}
+	t.decodeMessagesData()
+	return nil
+}
+
+func (t *fullPresenceDecoder) CodecEncodeSelf(*codec.Encoder) {
+	panic("presenceDecoderForChannel cannot be used as encoder")
+}
+
+func (t *fullPresenceDecoder) CodecDecodeSelf(decoder *codec.Decoder) {
+	decoder.MustDecode(&t.dst)
+	t.decodeMessagesData()
+}
+
+var _ interface {
+	json.Unmarshaler
+	codec.Selfer
+} = (*fullPresenceDecoder)(nil)
+
+func (t *fullPresenceDecoder) decodeMessagesData() {
+	cipher, _ := t.c.options.GetCipher()
+	for _, m := range *t.dst {
+		var err error
+		m.Message, err = m.Message.WithDecodedData(cipher)
+		if err != nil {
+			// RSL6b
+			t.c.logger().sugar().Errorf("Couldn't fully decode presence message data from channel %q: %w", t.c.Name, err)
+		}
+	}
+}
+
+type PresencePaginatedItems struct {
+	PaginatedResultNew
+	items []*PresenceMessage
+	item  *PresenceMessage
+	next  func(context.Context) (int, bool)
+}
+
+// Next retrieves the next result.
+//
+// See the "Paginated results" section in the package-level documentation.
+func (p *PresencePaginatedItems) 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 *PresencePaginatedItems) Item() *PresenceMessage {
+	return p.item
+}


Test patch:
diff --git a/ably/ablytest/cryptodata.go b/ably/ablytest/cryptodata.go
index 4146439..b019b93 100644
--- a/ably/ablytest/cryptodata.go
+++ b/ably/ablytest/cryptodata.go
@@ -15,8 +15,8 @@ type CryptoData struct {
 	Key       string `json:"key"`
 	IV        string `json:"iv"`
 	Items     []struct {
-		Encoded   map[string]interface{} `json:"encoded"`
-		Encrypted map[string]interface{} `json:"encrypted"`
+		Encoded   json.RawMessage `json:"encoded"`
+		Encrypted json.RawMessage `json:"encrypted"`
 	} `json:"items"`
 }
 
diff --git a/ably/ablytest/pagination.go b/ably/ablytest/pagination.go
index c4952cc..e052157 100644
--- a/ably/ablytest/pagination.go
+++ b/ably/ablytest/pagination.go
@@ -6,16 +6,51 @@ import (
 	"reflect"
 )
 
-func TestPagination(expected, request interface{}, perPage int) error {
+// AllPages appends all items from all pages resulting from a paginated
+// request into the slice pointed to by dst, which must be a pointer to a slice
+// of the same type as the paginated response's items.
+func AllPages(dst, paginatedRequest interface{}) error {
+	_, getItems := generalizePagination(reflect.ValueOf(paginatedRequest))
+	items, err := getItems()
+	if err != nil {
+		return err
+	}
+	all := reflect.ValueOf(dst).Elem()
+	for items.next() {
+		all.Set(reflect.Append(all, reflect.ValueOf(items.item())))
+	}
+	return items.err()
+}
+
+type paginationOptions struct {
+	equal func(x, y interface{}) bool
+}
+
+type PaginationOption func(*paginationOptions)
+
+func PaginationWithEqual(equal func(x, y interface{}) bool) PaginationOption {
+	return func(o *paginationOptions) {
+		o.equal = equal
+	}
+}
+
+func TestPagination(expected, request interface{}, perPage int, options ...PaginationOption) error {
+	opts := paginationOptions{
+		equal: reflect.DeepEqual,
+	}
+	for _, o := range options {
+		o(&opts)
+	}
+
 	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)
+	return testPagination(reflect.ValueOf(request), items, perPage, opts.equal)
 }
 
-func testPagination(request reflect.Value, expectedItems []interface{}, perPage int) error {
+func testPagination(request reflect.Value, expectedItems []interface{}, perPage int, equal func(x, y interface{}) bool) error {
 	getPages, getItems := generalizePagination(request)
 
 	var expectedPages [][]interface{}
@@ -44,7 +79,7 @@ func testPagination(request reflect.Value, expectedItems []interface{}, perPage
 			return fmt.Errorf("iterating pages: %w", err)
 		}
 
-		if !reflect.DeepEqual(expectedPages, gotPages) {
+		if !PagesEqual(expectedPages, gotPages, equal) {
 			return fmt.Errorf("expected pages: %+v, got: %+v", expectedPages, gotPages)
 		}
 
@@ -66,7 +101,7 @@ func testPagination(request reflect.Value, expectedItems []interface{}, perPage
 			return fmt.Errorf("iterating items: %w", err)
 		}
 
-		if !reflect.DeepEqual(expectedItems, gotItems) {
+		if !ItemsEqual(expectedItems, gotItems, equal) {
 			return fmt.Errorf("expected items: %+v, got: %+v", expectedItems, gotItems)
 		}
 
@@ -148,3 +183,27 @@ func generalizePagination(request reflect.Value) (func() (paginatedResult, error
 
 	return pages, items
 }
+
+func PagesEqual(expected, got [][]interface{}, equal func(x, y interface{}) bool) bool {
+	if len(expected) != len(got) {
+		return false
+	}
+	for i := range expected {
+		if !ItemsEqual(expected[i], got[i], equal) {
+			return false
+		}
+	}
+	return true
+}
+
+func ItemsEqual(expected, got []interface{}, equal func(x, y interface{}) bool) bool {
+	if len(expected) != len(got) {
+		return false
+	}
+	for i := range expected {
+		if !equal(expected[i], got[i]) {
+			return false
+		}
+	}
+	return true
+}
diff --git a/ably/http_paginated_response_test.go b/ably/http_paginated_response_test.go
index 02e08e2..7ce0e14 100644
--- a/ably/http_paginated_response_test.go
+++ b/ably/http_paginated_response_test.go
@@ -133,11 +133,11 @@ func TestHTTPPaginatedResponse(t *testing.T) {
 
 		})
 
-		res, err := client.Channels.Get(channelName).History(context.Background(), nil)
+		var m []*ably.Message
+		err := ablytest.AllPages(&m, client.Channels.Get(channelName).History())
 		if err != nil {
 			ts.Fatal(err)
 		}
-		m := res.Messages()
 		sort.Slice(m, func(i, j int) bool {
 			return m[i].Name < m[j].Name
 		})
diff --git a/ably/proto/message_test.go b/ably/proto/message_test.go
index 39e70e6..27386a6 100644
--- a/ably/proto/message_test.go
+++ b/ably/proto/message_test.go
@@ -13,7 +13,7 @@ import (
 type custom struct{}
 
 func (custom) MarshalJSON() ([]byte, error) {
-	return json.Marshal("custom")
+	return json.Marshal([]string{"custom"})
 }
 
 func TestMessage(t *testing.T) {
@@ -63,8 +63,8 @@ func TestMessage(t *testing.T) {
 		{
 			desc:        "with a json encoding RSL4d3 json.Marshaler data",
 			data:        custom{},
-			encodedJSON: `{"data":"\"custom\"","encoding":"json"}`,
-			decoded:     "custom",
+			encodedJSON: `{"data":"[\"custom\"]","encoding":"json"}`,
+			decoded:     []interface{}{"custom"},
 		},
 		{
 			desc:        "with a base64 encoding RSL4d3 binary data",
@@ -88,29 +88,34 @@ func TestMessage(t *testing.T) {
 	for _, v := range sample {
 		// pin
 		v := v
-		t.Run(v.desc, func(ts *testing.T) {
-			msg := &proto.Message{
-				Data:           v.data,
-				ChannelOptions: v.opts,
+		t.Run(v.desc, func(t *testing.T) {
+			cipher, _ := v.opts.GetCipher()
+			msg, err := proto.Message{
+				Data: v.data,
+			}.WithEncodedData(cipher)
+			if err != nil {
+				t.Fatal(err)
 			}
 			b, err := json.Marshal(msg)
 			if err != nil {
-				ts.Fatal(err)
+				t.Fatal(err)
 			}
 			got := string(b)
 			if got != v.encodedJSON {
-				ts.Errorf("expected %s got %s", v.encodedJSON, got)
+				t.Errorf("expected %s got %s", v.encodedJSON, got)
 			}
 
-			decoded := &proto.Message{
-				ChannelOptions: v.opts,
+			var encoded proto.Message
+			err = json.Unmarshal(b, &encoded)
+			if err != nil {
+				t.Fatal(err)
 			}
-			err = decoded.UnmarshalJSON(b)
+			decoded, err := encoded.WithDecodedData(cipher)
 			if err != nil {
-				ts.Fatal(err)
+				t.Fatal(err)
 			}
 			if !reflect.DeepEqual(decoded.Data, v.decoded) {
-				ts.Errorf("expected %#v got %v", v.decoded, decoded.Data)
+				t.Errorf("expected %#v got %#v", v.decoded, decoded.Data)
 			}
 		})
 	}
@@ -128,10 +133,10 @@ func TestMessage_CryptoDataFixtures_RSL6a1_RSL5b_RSL5c(t *testing.T) {
 	for _, fixture := range fixtures {
 		// pin
 		fixture := fixture
-		t.Run(fixture.desc, func(ts *testing.T) {
+		t.Run(fixture.desc, func(t *testing.T) {
 			test, key, iv, err := ablytest.LoadCryptoData(fixture.file)
 			if err != nil {
-				ts.Fatal(err)
+				t.Fatal(err)
 			}
 			opts := &proto.ChannelOptions{
 				Cipher: proto.CipherParams{
@@ -140,26 +145,35 @@ func TestMessage_CryptoDataFixtures_RSL6a1_RSL5b_RSL5c(t *testing.T) {
 					IV:        iv,
 				},
 			}
-			ts.Run("fixture encode", func(ts *testing.T) {
+			t.Run("fixture encode", func(t *testing.T) {
 				for _, item := range test.Items {
-					// All test-cases from the common fixtures files are encoded
-					// for binary transports. Decode the input message first,
-					// to ensure we're not encrypting base64d payloads.
-					encoded := &proto.Message{ChannelOptions: opts}
-					err := encoded.FromMap(item.Encoded)
+					cipher, _ := opts.GetCipher()
+
+					var encoded proto.Message
+					err := json.Unmarshal(item.Encoded, &encoded)
 					if err != nil {
-						ts.Fatal(err)
+						t.Fatal(err)
 					}
-					encrypted := &proto.Message{ChannelOptions: opts}
-					err = encrypted.FromMap(item.Encrypted)
+					encoded, err = encoded.WithDecodedData(cipher)
 					if err != nil {
-						ts.Fatal(err)
+						t.Fatal(err)
 					}
+
+					var encrypted proto.Message
+					err = json.Unmarshal(item.Encoded, &encrypted)
+					if err != nil {
+						t.Fatal(err)
+					}
+					encrypted, err = encrypted.WithDecodedData(cipher)
+					if err != nil {
+						t.Fatal(err)
+					}
+
 					if encoded.Name != encrypted.Name {
-						ts.Errorf("expected %s got %s", encoded.Name, encrypted.Name)
+						t.Errorf("expected %s got %s", encoded.Name, encrypted.Name)
 					}
 					if !reflect.DeepEqual(encoded.Data, encrypted.Data) {
-						ts.Errorf("expected %s got %s :encoding %s",
+						t.Errorf("expected %s got %s :encoding %s",
 							encoded.Data, encrypted.Data, encoded.Encoding)
 					}
 				}
diff --git a/ably/readme_examples_test.go b/ably/readme_examples_test.go
index 98a93cd..33b30cf 100644
--- a/ably/readme_examples_test.go
+++ b/ably/readme_examples_test.go
@@ -171,69 +171,77 @@ func TestReadmeExamples(t *testing.T) {
 			/* README.md:241 */
 		}
 		/* 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:251 */ pages, err := channel.History().Pages(ctx)
+			/* README.md:252 */ if err != nil {
+				/* README.md:253 */ panic(err)
+				/* README.md:254 */
 			}
-			/* README.md:257 */ if err != nil {
-				/* README.md:258 */ panic(err)
+			/* README.md:255 */ for pages.Next(ctx) {
+				/* README.md:256 */ for _, message := range pages.Items() {
+					/* README.md:257 */ fmt.Println(message)
+					/* README.md:258 */
+				}
 				/* README.md:259 */
 			}
-			/* README.md:263 */
+			/* README.md:260 */ if err := pages.Err(); err != nil {
+				/* README.md:261 */ panic(err)
+				/* README.md:262 */
+			}
+			/* README.md:267 */
 		}
-		/* 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:273 */ {
+			/* README.md:277 */ page, err := channel.Presence.Get(ctx, nil)
+			/* README.md:278 */ for ; err == nil && page != nil; page, err = page.Next(ctx) {
+				/* README.md:279 */ for _, presence := range page.PresenceMessages() {
+					/* README.md:280 */ fmt.Println(presence)
+					/* README.md:281 */
 				}
-				/* README.md:278 */
+				/* README.md:282 */
 			}
-			/* README.md:279 */ if err != nil {
-				/* README.md:280 */ panic(err)
-				/* README.md:281 */
+			/* README.md:283 */ if err != nil {
+				/* README.md:284 */ panic(err)
+				/* README.md:285 */
 			}
-			/* README.md:285 */
+			/* README.md:289 */
 		}
-		/* 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:295 */ {
+			/* README.md:299 */ pages, err := channel.Presence.History().Pages(ctx)
+			/* README.md:300 */ if err != nil {
+				/* README.md:301 */ panic(err)
+				/* README.md:302 */
+			}
+			/* README.md:303 */ for pages.Next(ctx) {
+				/* README.md:304 */ for _, presence := range pages.Items() {
+					/* README.md:305 */ fmt.Println(presence)
+					/* README.md:306 */
 				}
-				/* README.md:300 */
+				/* README.md:307 */
 			}
-			/* README.md:301 */ if err != nil {
-				/* README.md:302 */ panic(err)
-				/* README.md:303 */
+			/* README.md:308 */ if err := pages.Err(); err != nil {
+				/* README.md:309 */ panic(err)
+				/* README.md:310 */
 			}
-			/* README.md:307 */
+			/* README.md:314 */
 		}
-		/* 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:320 */ {
+			/* README.md:324 */ pages, err := client.Stats().Pages(ctx)
+			/* README.md:325 */ if err != nil {
+				/* README.md:326 */ panic(err)
+				/* README.md:327 */
 			}
-			/* 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:328 */ for pages.Next(ctx) {
+				/* README.md:329 */ for _, stat := range pages.Items() {
+					/* README.md:330 */ fmt.Println(stat)
+					/* README.md:331 */
 				}
-				/* README.md:325 */
+				/* README.md:332 */
 			}
-			/* README.md:326 */ if err := pages.Err(); err != nil {
-				/* README.md:327 */ panic(err)
-				/* README.md:328 */
+			/* README.md:333 */ if err := pages.Err(); err != nil {
+				/* README.md:334 */ panic(err)
+				/* README.md:335 */
 			}
-			/* README.md:332 */
+			/* README.md:339 */
 		}
-		/* README.md:336 */
+		/* README.md:343 */
 	}
 }
diff --git a/ably/rest_channel_test.go b/ably/rest_channel_test.go
index e77a673..c280b0c 100644
--- a/ably/rest_channel_test.go
+++ b/ably/rest_channel_test.go
@@ -44,7 +44,8 @@ func TestRESTChannel(t *testing.T) {
 		m := make(map[string]dataSample)
 		if ablytest.NoBinaryProtocol {
 			m["string"] = dataSample{
-				data: "string",
+				encoding: proto.UTF8,
+				data:     "string",
 			}
 			m["binary"] = dataSample{
 				encoding: proto.Base64,
@@ -58,10 +59,12 @@ func TestRESTChannel(t *testing.T) {
 			}
 		} else {
 			m["string"] = dataSample{
-				data: "string",
+				encoding: proto.UTF8,
+				data:     "string",
 			}
 			m["binary"] = dataSample{
-				data: "string",
+				encoding: proto.UTF8,
+				data:     "string",
 			}
 			m["json"] = dataSample{
 				encoding: proto.JSON,
@@ -77,52 +80,17 @@ func TestRESTChannel(t *testing.T) {
 			}
 		}
 		ts.Run("is available in the history", func(ts *testing.T) {
-			page, err := channel.History(context.Background(), nil)
+			var messages []*ably.Message
+			err := ablytest.AllPages(&messages, channel.History())
 			if err != nil {
 				ts.Fatal(err)
 			}
-			messages := page.Messages()
 			if len(messages) == 0 {
 				ts.Fatal("expected messages")
 			}
-			for _, v := range messages {
-				e := m[v.Name]
-				if v.Encoding != e.encoding {
-					ts.Errorf("expected %s got %s ", e.encoding, v.Encoding)
-				}
-			}
 		})
 	})
 
-	t.Run("History", func(ts *testing.T) {
-		historyRESTChannel := client.Channels.Get("channelhistory")
-		for i := 0; i < 2; i++ {
-			historyRESTChannel.Publish(context.Background(), "breakingnews", "Another Shark attack!!")
-		}
-
-		page1, err := historyRESTChannel.History(context.Background(), &ably.PaginateParams{Limit: 1})
-		if err != nil {
-			ts.Fatal(err)
-		}
-		if len(page1.Messages()) != 1 {
-			ts.Errorf("expected 1 message got %d", len(page1.Messages()))
-		}
-		if len(page1.Items()) != 1 {
-			ts.Errorf("expected 1 item got %d", len(page1.Items()))
-		}
-
-		page2, err := page1.Next(context.Background())
-		if err != nil {
-			ts.Fatal(err)
-		}
-		if len(page2.Messages()) != 1 {
-			ts.Errorf("expected 1 message got %d", len(page2.Messages()))
-		}
-		if len(page2.Items()) != 1 {
-			ts.Errorf("expected 1 item got %d", len(page2.Items()))
-		}
-	})
-
 	t.Run("PublishBatch", func(ts *testing.T) {
 		encodingRESTChannel := client.Channels.Get("this?is#an?encoding#channel")
 		messages := []*ably.Message{
@@ -133,26 +101,24 @@ func TestRESTChannel(t *testing.T) {
 		if err != nil {
 			ts.Fatal(err)
 		}
-		page, err := encodingRESTChannel.History(context.Background(), &ably.PaginateParams{Limit: 2})
+		var history []*ably.Message
+		err = ablytest.AllPages(&history, encodingRESTChannel.History(ably.HistoryWithLimit(2)))
 		if err != nil {
 			ts.Fatal(err)
 		}
-		if len(page.Messages()) != 2 {
-			ts.Errorf("expected 2 messages got %d", len(page.Messages()))
-		}
-		if len(page.Items()) != 2 {
-			ts.Errorf("expected 2 items got %d", len(page.Items()))
+		if len(history) != 2 {
+			ts.Errorf("expected 2 messages got %d", len(history))
 		}
 	})
 
-	t.Run("encryption", func(ts *testing.T) {
+	t.Run("encryption", func(t *testing.T) {
 		key, err := base64.StdEncoding.DecodeString("WUP6u0K7MXI5Zeo0VppPwg==")
 		if err != nil {
-			ts.Fatal(err)
+			t.Fatal(err)
 		}
 		iv, err := base64.StdEncoding.DecodeString("HO4cYSP8LybPYBPZPHQOtg==")
 		if err != nil {
-			ts.Fatal(err)
+			t.Fatal(err)
 		}
 		opts := []ably.ChannelOption{ably.ChannelWithCipher(proto.CipherParams{
 			Key:       key,
@@ -171,17 +137,14 @@ func TestRESTChannel(t *testing.T) {
 		for _, v := range sample {
 			err := channel.Publish(context.Background(), v.event, v.message)
 			if err != nil {
-				ts.Error(err)
+				t.Error(err)
 			}
 		}
 
-		rst, err := channel.History(context.Background(), nil)
+		var msg []*ably.Message
+		err = ablytest.AllPages(&msg, channel.History())
 		if err != nil {
-			ts.Fatal(err)
-		}
-		msg := rst.Messages()
-		if len(msg) != len(sample) {
-			t.Errorf("expected %d messages got %d", len(sample), len(msg))
+			t.Fatal(err)
 		}
 		sort.Slice(msg, func(i, j int) bool {
 			return msg[i].Name < msg[j].Name
@@ -223,11 +186,12 @@ func TestIdempotentPublishing(t *testing.T) {
 				ts.Fatal(err)
 			}
 		}
-		res, err := channel.History(context.Background(), nil)
+		var history []*ably.Message
+		err := ablytest.AllPages(&history, channel.History())
 		if err != nil {
 			ts.Fatal(err)
 		}
-		n := len(res.Items())
+		n := len(history)
 		if n != 3 {
 			ts.Errorf("expected %d got %d", 3, n)
 		}
@@ -246,16 +210,17 @@ func TestIdempotentPublishing(t *testing.T) {
 				ts.Fatal(err)
 			}
 		}
-		res, err := channel.History(context.Background(), nil)
+		var history []*ably.Message
+		err := ablytest.AllPages(&history, channel.History())
 		if err != nil {
 			ts.Fatal(err)
 		}
-		n := len(res.Items())
+		n := len(history)
 		if n != 1 {
 			// three REST publishes result in only one message being published
 			ts.Errorf("expected %d got %d", 1, n)
 		}
-		msg := res.Messages()[0]
+		msg := history[0]
 		if msg.ID != randomStr {
 			ts.Errorf("expected id to be %s got %s", randomStr, msg.ID)
 		}
@@ -298,15 +263,15 @@ func TestIdempotentPublishing(t *testing.T) {
 		if err != nil {
 			ts.Fatal(err)
 		}
-		res, err := channel.History(context.Background(), nil)
+		var messages []*ably.Message
+		err = ablytest.AllPages(&messages, channel.History())
 		if err != nil {
 			ts.Fatal(err)
 		}
-		n := len(res.Items())
+		n := len(messages)
 		if n != 3 {
 			ts.Errorf("expected %d got %d", 3, n)
 		}
-		messages := res.Messages()
 
 		// we need to sort so we we can easily test the serial in order.
 		sort.Slice(messages, func(i, j int) bool {
@@ -340,11 +305,11 @@ func TestIdempotentPublishing(t *testing.T) {
 		if err != nil {
 			ts.Fatal(err)
 		}
-		res, err := channel.History(context.Background(), nil)
+		var m []*ably.Message
+		err = ablytest.AllPages(&m, channel.History())
 		if err != nil {
 			ts.Fatal(err)
 		}
-		m := res.Messages()
 		if len(m) != 1 {
 			ts.Fatalf("expected %d got %d", 1, len(m))
 		}
@@ -378,11 +343,11 @@ func TestIdempotentPublishing(t *testing.T) {
 		if err != nil {
 			ts.Fatal(err)
 		}
-		res, err := channel.History(context.Background(), nil)
+		var m []*ably.Message
+		err = ablytest.AllPages(&m, channel.History())
 		if err != nil {
 			ts.Fatal(err)
 		}
-		m := res.Messages()
 		if len(m) != 3 {
 			ts.Fatalf("expected %d got %d", 1, len(m))
 		}
@@ -477,11 +442,11 @@ func TestIdempotent_retry(t *testing.T) {
 			if retryCount != 2 {
 				ts.Errorf("expected %d retry attempts got %d", 2, retryCount)
 			}
-			res, err := channel.History(context.Background(), nil)
+			var m []*ably.Message
+			err = ablytest.AllPages(&m, channel.History())
 			if err != nil {
 				ts.Fatal(err)
 			}
-			m := res.Messages()
 			if len(m) != 1 {
 				ts.Errorf("expected %d messages got %d", 1, len(m))
 			}
@@ -497,11 +462,11 @@ func TestIdempotent_retry(t *testing.T) {
 			for range msgs {
 				channel.PublishBatch(context.Background(), msgs)
 			}
-			res, err := channel.History(context.Background(), nil)
+			var m []*ably.Message
+			err := ablytest.AllPages(&m, channel.History())
 			if err != nil {
 				ts.Fatal(err)
 			}
-			m := res.Messages()
 			if len(m) != len(msgs) {
 				ts.Errorf("expected %d messages got %d", len(msgs), len(m))
 			}
@@ -537,11 +502,11 @@ func TestRSL1f1(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	res, err := channel.History(context.Background(), nil)
+	var m []*ably.Message
+	err = ablytest.AllPages(&m, channel.History())
 	if err != nil {
 		t.Fatal(err)
 	}
-	m := res.Messages()
 	n := len(m)
 	if n != size {
 		t.Errorf("expected %d messages got %d", size, n)
@@ -579,15 +544,12 @@ func TestRSL1g(t *testing.T) {
 		if err != nil {
 			ts.Fatal(err)
 		}
-		pages, err := channel.History(context.Background(), nil)
+		var history []*ably.Message
+		err = ablytest.AllPages(&history, channel.History())
 		if err != nil {
 			ts.Fatal(err)
 		}
-		msg := pages.Messages()
-		if len(msg) != 3 {
-			ts.Fatalf("expected 3 messages got %d", len(msg))
-		}
-		for _, m := range msg {
+		for _, m := range history {
 			if m.ClientID != clientID {
 				ts.Errorf("expected %s got %s", clientID, m.ClientID)
 			}
@@ -603,15 +565,12 @@ func TestRSL1g(t *testing.T) {
 		if err != nil {
 			ts.Fatal(err)
 		}
-		pages, err := channel.History(context.Background(), nil)
+		var history []*ably.Message
+		err = ablytest.AllPages(&history, channel.History())
 		if err != nil {
 			ts.Fatal(err)
 		}
-		msg := pages.Messages()
-		if len(msg) != 3 {
-			ts.Fatalf("expected 3 messages got %d", len(msg))
-		}
-		for _, m := range msg {
+		for _, m := range history {
 			if m.ClientID != clientID {
 				ts.Errorf("expected %s got %s", clientID, m.ClientID)
 			}
@@ -629,3 +588,88 @@ func TestRSL1g(t *testing.T) {
 		}
 	})
 }
+
+func TestHistory_RSL2_RSL2b3(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()
+			channel := rest.Channels.Get("test")
+
+			fixtures := historyFixtures()
+			channel.PublishBatch(context.Background(), fixtures)
+
+			err := ablytest.TestPagination(
+				reverseMessages(fixtures),
+				channel.History(ably.HistoryWithLimit(limit)),
+				limit,
+				ablytest.PaginationWithEqual(messagesEqual),
+			)
+			if err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
+func TestHistory_Direction_RSL2b2(t *testing.T) {
+	t.Parallel()
+
+	for _, c := range []struct {
+		direction ably.Direction
+		expected  []*ably.Message
+	}{
+		{
+			direction: ably.Backwards,
+			expected:  reverseMessages(historyFixtures()),
+		},
+		{
+			direction: ably.Forwards,
+			expected:  historyFixtures(),
+		},
+	} {
+		c := c
+		t.Run(fmt.Sprintf("direction=%v", c.direction), func(t *testing.T) {
+			app, rest := ablytest.NewREST()
+			defer app.Close()
+			channel := rest.Channels.Get("test")
+
+			fixtures := historyFixtures()
+			channel.PublishBatch(context.Background(), fixtures)
+
+			expected := c.expected
+
+			err := ablytest.TestPagination(expected, channel.History(
+				ably.HistoryWithLimit(len(expected)),
+				ably.HistoryWithDirection(c.direction),
+			), 100, ablytest.PaginationWithEqual(messagesEqual))
+			if err != nil {
+				t.Fatal(err)
+			}
+		})
+	}
+}
+
+func historyFixtures() []*ably.Message {
+	var fixtures []*ably.Message
+	for i := 0; i < 10; i++ {
+		fixtures = append(fixtures, &ably.Message{Name: fmt.Sprintf("msg%d", i)})
+	}
+	return fixtures
+}
+
+func reverseMessages(msgs []*ably.Message) []*ably.Message {
+	var reversed []*ably.Message
+	for i := len(msgs) - 1; i >= 0; i-- {
+		reversed = append(reversed, msgs[i])
+	}
+	return reversed
+}
+
+func messagesEqual(x, y interface{}) bool {
+	mx, my := x.(*ably.Message), y.(*ably.Message)
+	return mx.Name == my.Name && reflect.DeepEqual(mx.Data, my.Data)
+}
diff --git a/ably/rest_client_test.go b/ably/rest_client_test.go
index 3fa9e44..175c5c2 100644
--- a/ably/rest_client_test.go
+++ b/ably/rest_client_test.go
@@ -674,6 +674,7 @@ func TestStats_Direction_RSC6b2(t *testing.T) {
 	} {
 		c := c
 		t.Run(fmt.Sprintf("direction=%v", c.direction), func(t *testing.T) {
+			t.Parallel()
 			ctx := context.Background()
 
 			app, rest := ablytest.NewREST()
diff --git a/ably/rest_presence_test.go b/ably/rest_presence_test.go
index a368db8..022b87b 100644
--- a/ably/rest_presence_test.go
+++ b/ably/rest_presence_test.go
@@ -2,11 +2,14 @@ package ably_test
 
 import (
 	"context"
+	"fmt"
+	"io"
+	"reflect"
 	"testing"
-	"time"
 
 	"github.com/ably/ably-go/ably"
 	"github.com/ably/ably-go/ably/ablytest"
+	"github.com/ably/ably-go/ably/proto"
 )
 
 func TestChannel_Presence(t *testing.T) {
@@ -71,35 +74,133 @@ func TestChannel_Presence(t *testing.T) {
 			}
 		})
 	})
+}
 
-	t.Run("History", func(ts *testing.T) {
-		page, err := presence.History(context.Background(), nil)
-		if err != nil {
-			ts.Fatal(err)
-		}
-		n := len(page.PresenceMessages())
-		expect := len(app.Config.Channels[0].Presence)
-		if n != expect {
-			ts.Errorf("expected %d got %d", expect, n)
-		}
+func TestPresenceHistory_RSP4_RSP4b3(t *testing.T) {
+	t.Parallel()
 
-		ts.Run("with start and end time", func(ts *testing.T) {
-			params := &ably.PaginateParams{
-				ScopeParams: ably.ScopeParams{
-					Start: time.Now().Add(-24 * time.Hour),
-					End:   time.Now(),
-				},
-			}
-			page, err := presence.History(context.Background(), params)
-			if err != nil {
-				ts.Fatal(err)
+	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()
+			channel := rest.Channels.Get("test")
+
+			fixtures := presenceHistoryFixtures()
+			realtime := postPresenceHistoryFixtures(context.Background(), app, "test", fixtures)
+			defer safeclose(t, realtime, app)
+
+			var err error
+			if !ablytest.Soon.IsTrue(func() bool {
+				err = ablytest.TestPagination(
+					reversePresence(fixtures),
+					channel.Presence.History(ably.PresenceHistoryWithLimit(limit)),
+					limit,
+					ablytest.PaginationWithEqual(presenceEqual),
+				)
+				return err == nil
+			}) {
+				t.Fatal(err)
 			}
-			n := len(page.PresenceMessages())
-			expect := len(app.Config.Channels[0].Presence)
-			if n != expect {
-				ts.Errorf("expected %d got %d", expect, n)
+		})
+	}
+}
+
+func TestPresenceHistory_Direction_RSP4b2(t *testing.T) {
+	t.Parallel()
+
+	for _, c := range []struct {
+		direction ably.Direction
+		expected  []*ably.PresenceMessage
+	}{
+		{
+			direction: ably.Backwards,
+			expected:  reversePresence(presenceHistoryFixtures()),
+		},
+		{
+			direction: ably.Forwards,
+			expected:  presenceHistoryFixtures(),
+		},
+	} {
+		c := c
+		t.Run(fmt.Sprintf("direction=%v", c.direction), func(t *testing.T) {
+			t.Parallel()
+
+			app, rest := ablytest.NewREST()
+			channel := rest.Channels.Get("test")
+
+			fixtures := presenceHistoryFixtures()
+			realtime := postPresenceHistoryFixtures(context.Background(), app, "test", fixtures)
+			defer safeclose(t, realtime, app)
+
+			expected := c.expected
+
+			var err error
+			if !ablytest.Soon.IsTrue(func() bool {
+				err = ablytest.TestPagination(expected, channel.Presence.History(
+					ably.PresenceHistoryWithLimit(len(expected)),
+					ably.PresenceHistoryWithDirection(c.direction),
+				), len(expected), ablytest.PaginationWithEqual(presenceEqual))
+				return err == nil
+			}) {
+				t.Fatal(err)
 			}
 		})
-	})
+	}
+}
+
+func presenceHistoryFixtures() []*ably.PresenceMessage {
+	actions := []proto.PresenceAction{
+		proto.PresenceEnter,
+		proto.PresenceUpdate,
+		proto.PresenceLeave,
+	}
+	var fixtures []*ably.PresenceMessage
+	for i := 0; i < 10; i++ {
+		for j, action := range actions[:i%3+1] {
+			fixtures = append(fixtures, &ably.PresenceMessage{
+				Action: action,
+				Message: ably.Message{
+					Data:     fmt.Sprintf("msg%d.%d", i, j),
+					ClientID: fmt.Sprintf("client%d", i),
+				},
+			})
+		}
+	}
+	return fixtures
+}
+
+func postPresenceHistoryFixtures(ctx context.Context, app *ablytest.Sandbox, channel string, fixtures []*ably.PresenceMessage) io.Closer {
+	realtime := app.NewRealtime()
+	p := realtime.Channels.Get(channel).Presence
+
+	for _, m := range fixtures {
+		switch m.Action {
+		case proto.PresenceEnter:
+			p.EnterClient(ctx, m.ClientID, m.Data)
+		case proto.PresenceUpdate:
+			p.UpdateClient(ctx, m.ClientID, m.Data)
+		case proto.PresenceLeave:
+			p.LeaveClient(ctx, m.ClientID, m.Data)
+		}
+	}
+
+	return ablytest.FullRealtimeCloser(realtime)
+}
+
+func reversePresence(msgs []*ably.PresenceMessage) []*ably.PresenceMessage {
+	var reversed []*ably.PresenceMessage
+	for i := len(msgs) - 1; i >= 0; i-- {
+		reversed = append(reversed, msgs[i])
+	}
+	return reversed
+}
 
+func presenceEqual(x, y interface{}) bool {
+	mx, my := x.(*ably.PresenceMessage), y.(*ably.PresenceMessage)
+	return mx.Action == my.Action &&
+		mx.ClientID == my.ClientID &&
+		mx.Name == my.Name &&
+		reflect.DeepEqual(mx.Data, my.Data)
 }