Repository: ably/ably-go
Pull request #202 (2020-10-14 14:09:01)

v1.2-compliant message and presence publish and subscribe
Missing tests, but can be reviewed.


Patch:
diff --git a/.travis.yml b/.travis.yml
index 45e94e4..fec5bcb 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -10,5 +10,5 @@ env:
 sudo: false
 script:
   - make build
-  - if ! (git status | grep "working tree clean"); then echo "uncommitted generated code; run 'make build' and commit the result"; exit 1; fi
+  - if ! (git status | grep "working tree clean"); then git status; echo "uncommitted generated code; run 'make build' and commit the result"; exit 1; fi
   - make test
diff --git a/ably/doc.go b/ably/doc.go
index 59113f4..e053372 100644
--- a/ably/doc.go
+++ b/ably/doc.go
@@ -30,4 +30,7 @@
 //
 // Calling any of these methods an "off" function inside a handler will only
 // have effect for subsequent events.
+//
+// For messages and presence messages, "on" is called "subscribe" and "off" is
+// called "unsubscribe".
 package ably
diff --git a/ably/error.go b/ably/error.go
index 57a857b..22e401c 100644
--- a/ably/error.go
+++ b/ably/error.go
@@ -119,6 +119,13 @@ func newErrorFromProto(err *proto.ErrorInfo) *ErrorInfo {
 	}
 }
 
+func (e *ErrorInfo) unwrapNil() error {
+	if e == nil {
+		return nil
+	}
+	return e
+}
+
 func code(err error) ErrorCode {
 	if e, ok := err.(*ErrorInfo); ok {
 		return e.Code
diff --git a/ably/options.go b/ably/options.go
index c0b9b7b..47a1398 100644
--- a/ably/options.go
+++ b/ably/options.go
@@ -24,7 +24,7 @@ const (
 )
 
 var defaultOptions = clientOptions{
-	RestHost:                 restHost,
+	RESTHost:                 restHost,
 	HTTPMaxRetryCount:        3,
 	HTTPRequestTimeout:       10 * time.Second,
 	RealtimeHost:             "realtime.ably.io",
@@ -33,7 +33,7 @@ var defaultOptions = clientOptions{
 	DisconnectedRetryTimeout: 15 * time.Second, // TO3l1
 	ChannelRetryTimeout:      15 * time.Second, // TO3l7
 	FallbackRetryTimeout:     10 * time.Minute,
-	IdempotentRestPublishing: false,
+	IdempotentRESTPublishing: false,
 	Port:                     80,
 	TLSPort:                  443,
 	Now:                      time.Now,
@@ -121,7 +121,7 @@ type authOptions struct {
 	// Spec: TO3j11
 	DefaultTokenParams *TokenParams
 
-	// UseTokenAuth makes the Rest and Realtime clients always use token
+	// UseTokenAuth makes the REST and Realtime clients always use token
 	// authentication method.
 	UseTokenAuth bool
 }
@@ -161,7 +161,7 @@ func (opts *authOptions) KeySecret() string {
 type clientOptions struct {
 	authOptions
 
-	RestHost string // optional; overwrite endpoint hostname for REST client
+	RESTHost string // optional; overwrite endpoint hostname for REST client
 
 	FallbackHosts   []string
 	RealtimeHost    string        // optional; overwrite endpoint hostname for Realtime client
@@ -194,7 +194,7 @@ type clientOptions struct {
 
 	// When true idempotent rest publishing will be enabled.
 	// Spec TO3n
-	IdempotentRestPublishing bool
+	IdempotentRESTPublishing bool
 
 	// TimeoutConnect is the time period after which connect request is failed.
 	//
@@ -217,7 +217,7 @@ type clientOptions struct {
 	// If Dial is nil, the default websocket connection is used.
 	Dial func(protocol string, u *url.URL) (proto.Conn, error)
 
-	// HTTPClient specifies the client used for HTTP communication by RestClient.
+	// HTTPClient specifies the client used for HTTP communication by REST.
 	//
 	// If HTTPClient is nil, a client configured with default settings is used.
 	HTTPClient *http.Client
@@ -266,7 +266,7 @@ func (opts *clientOptions) disconnectedRetryTimeout() time.Duration {
 }
 
 func (opts *clientOptions) restURL() string {
-	host := resolveHost(opts.RestHost, opts.Environment, defaultOptions.RestHost)
+	host := resolveHost(opts.RESTHost, opts.Environment, defaultOptions.RESTHost)
 	if opts.NoTLS {
 		port := opts.Port
 		if port == 0 {
@@ -324,8 +324,8 @@ func (opts *clientOptions) protocol() string {
 	return protocolMsgPack
 }
 
-func (opts *clientOptions) idempotentRestPublishing() bool {
-	return opts.IdempotentRestPublishing
+func (opts *clientOptions) idempotentRESTPublishing() bool {
+	return opts.IdempotentRESTPublishing
 }
 
 type ScopeParams struct {
@@ -573,7 +573,7 @@ func (os ClientOptions) QueueMessages(queue bool) ClientOptions {
 
 func (os ClientOptions) RESTHost(host string) ClientOptions {
 	return append(os, func(os *clientOptions) {
-		os.RestHost = host
+		os.RESTHost = host
 	})
 }
 
@@ -639,7 +639,7 @@ func (os ClientOptions) HTTPMaxRetryCount(count int) ClientOptions {
 
 func (os ClientOptions) IdempotentRESTPublishing(idempotent bool) ClientOptions {
 	return append(os, func(os *clientOptions) {
-		os.IdempotentRestPublishing = idempotent
+		os.IdempotentRESTPublishing = idempotent
 	})
 }
 
diff --git a/ably/proto/presence_message.go b/ably/proto/presence_message.go
index 156f6e5..78b423e 100644
--- a/ably/proto/presence_message.go
+++ b/ably/proto/presence_message.go
@@ -6,10 +6,10 @@ import (
 	"github.com/ugorji/go/codec"
 )
 
-type PresenceState int64
+type PresenceAction int64
 
 const (
-	PresenceAbsent PresenceState = iota
+	PresenceAbsent PresenceAction = iota
 	PresencePresent
 	PresenceEnter
 	PresenceLeave
@@ -18,7 +18,7 @@ const (
 
 type PresenceMessage struct {
 	Message
-	State PresenceState `json:"action" codec:"action"`
+	Action PresenceAction `json:"action" codec:"action"`
 }
 
 func (m PresenceMessage) MarshalJSON() ([]byte, error) {
@@ -27,7 +27,7 @@ func (m PresenceMessage) MarshalJSON() ([]byte, error) {
 		return nil, err
 	}
 	ctx := e.ToMap()
-	ctx["action"] = m.State
+	ctx["action"] = m.Action
 	return json.Marshal(ctx)
 }
 
@@ -46,7 +46,7 @@ func (m PresenceMessage) CodecEncodeSelf(encoder *codec.Encoder) {
 		panic(err)
 	}
 	ctx := e.ToMap()
-	ctx["action"] = m.State
+	ctx["action"] = m.Action
 	encoder.MustEncode(ctx)
 }
 
@@ -65,13 +65,13 @@ func (m *PresenceMessage) FromMap(ctx map[string]interface{}) error {
 		return err
 	}
 	if v, ok := ctx["action"]; ok {
-		m.State = PresenceState(coerceInt64(v))
+		m.Action = PresenceAction(coerceInt64(v))
 	}
 	return nil
 }
 
 func (m PresenceMessage) ToMap() map[string]interface{} {
 	ctx := m.Message.ToMap()
-	ctx["action"] = m.State
+	ctx["action"] = m.Action
 	return ctx
 }
diff --git a/ably/proto/protocol_message.go b/ably/proto/protocol_message.go
index 8cbe151..a9db03b 100644
--- a/ably/proto/protocol_message.go
+++ b/ably/proto/protocol_message.go
@@ -107,17 +107,17 @@ func (p *ProtocolMessage) FromMap(ctx map[string]interface{}) {
 	if v, ok := ctx["messages"]; ok {
 		i := v.([]interface{})
 		for _, v := range i {
-			msg := &Message{}
+			var msg Message
 			msg.FromMap(v.(map[string]interface{}))
-			p.Messages = append(p.Messages, msg)
+			p.Messages = append(p.Messages, &msg)
 		}
 	}
 	if v, ok := ctx["presence"]; ok {
 		i := v.([]interface{})
 		for _, v := range i {
-			msg := &PresenceMessage{}
+			var msg PresenceMessage
 			msg.FromMap(v.(map[string]interface{}))
-			p.Presence = append(p.Presence, msg)
+			p.Presence = append(p.Presence, &msg)
 		}
 	}
 	if v, ok := ctx["id"]; ok {
diff --git a/ably/proto/stat.go b/ably/proto/stat.go
index 8a93a97..621262b 100644
--- a/ably/proto/stat.go
+++ b/ably/proto/stat.go
@@ -53,7 +53,7 @@ type MessageTypes struct {
 type MessageTraffic struct {
 	All           MessageTypes `json:"all" codec:"all"`
 	RealTime      MessageTypes `json:"realtime" codec:"realtime"`
-	Rest          MessageTypes `json:"rest" codec:"rest"`
+	REST          MessageTypes `json:"rest" codec:"rest"`
 	Webhook       MessageTypes `json:"webhook" codec:"webhook"`
 	Push          MessageTypes `json:"push" codec:"push"`
 	ExternalQueue MessageTypes `json:"externalQueue" codec:"externalQueue"`
diff --git a/ably/realtime_channel.go b/ably/realtime_channel.go
index c4e91d7..9e98ed0 100644
--- a/ably/realtime_channel.go
+++ b/ably/realtime_channel.go
@@ -103,18 +103,16 @@ func (ch *Channels) All() []*RealtimeChannel {
 	return chans
 }
 
-// Release closes a channel looked up by the name.
-//
-// It is safe to call Release from multiple goroutines - if a channel happened
-// to be already concurrently released, the method is a nop.
-func (ch *Channels) Release(name string) error {
+// Release releases all resources associated with a channel, detaching it first
+// if necessary. See RealtimeChannel.Detach for details.
+func (ch *Channels) Release(ctx context.Context, name string) error {
 	ch.mtx.Lock()
 	defer ch.mtx.Unlock()
 	c, ok := ch.chans[name]
 	if !ok {
 		return nil
 	}
-	err := wait(c.Detach())
+	err := c.Detach(ctx)
 	if err != nil {
 		return err
 	}
@@ -142,9 +140,9 @@ type RealtimeChannel struct {
 	errorReason     *ErrorInfo
 	internalEmitter ChannelEventEmitter
 
-	client *Realtime
-	subs   *subscriptions
-	queue  *msgQueue
+	client         *Realtime
+	messageEmitter *eventEmitter
+	queue          *msgQueue
 }
 
 func newRealtimeChannel(name string, client *Realtime) *RealtimeChannel {
@@ -155,8 +153,8 @@ func newRealtimeChannel(name string, client *Realtime) *RealtimeChannel {
 		state:           ChannelStateInitialized,
 		internalEmitter: ChannelEventEmitter{newEventEmitter(client.logger())},
 
-		client: client,
-		subs:   newSubscriptions(subscriptionMessages, client.logger()),
+		client:         client,
+		messageEmitter: newEventEmitter(client.logger()),
 	}
 	c.Presence = newRealtimePresence(c)
 	c.queue = newMsgQueue(client.Connection)
@@ -175,15 +173,19 @@ func (c *RealtimeChannel) onConnStateChange(change ConnectionStateChange) {
 	}
 }
 
-// Attach initiates attach request, which is being processed on a separate
-// goroutine.
+// Attach attaches the Realtime connection to the channel, after which it starts
+// receiving messages from it.
 //
-// If channel is already attached, this method is a nop.
-// If sending attach message failed, the returned error value is non-nil.
-// If sending attach message succeed, the returned Result value can be used
-// to wait until ack from server is received.
-func (c *RealtimeChannel) Attach() (Result, error) {
-	return c.attach(true)
+// If the context is canceled before the attach operation finishes, the call
+// returns with an error, but the operation carries on in the background and
+// the channel may eventually be attached anyway.
+func (c *RealtimeChannel) Attach(ctx context.Context) error {
+	res, err := c.attach(true)
+	if err != nil {
+		return err
+	}
+	// TODO: Don't ignore context.
+	return res.Wait()
 }
 
 func (c *RealtimeChannel) attach(result bool) (Result, error) {
@@ -226,7 +228,7 @@ func (c *RealtimeChannel) lockAttach(result bool, err error) (Result, error) {
 		return goWaiter(func() error {
 			change := <-changes
 			if change.Current != ConnectionStateConnected {
-				return change.Reason
+				return change.Reason.unwrapNil()
 			}
 
 			res, err := c.mayAttach(result, true)
@@ -258,15 +260,19 @@ func (c *RealtimeChannel) lockAttach(result bool, err error) (Result, error) {
 	return res, nil
 }
 
-// Detach initiates detach request, which is being processed on a separate
-// goroutine.
+// Detach detaches the Realtime connection to the channel, after which it stops
+// receiving messages from it.
 //
-// If channel is already detached, this method is a nop.
-// If sending detach message failed, the returned error value is non-nil.
-// If sending detach message succeed, the returned Result value can be used
-// to wait until ack from server is received.
-func (c *RealtimeChannel) Detach() (Result, error) {
-	return c.detach(true)
+// If the context is canceled before the detach operation finishes, the call
+// returns with an error, but the operation carries on in the background and
+// the channel may eventually be detached anyway.
+func (c *RealtimeChannel) Detach(ctx context.Context) error {
+	res, err := c.detach(true)
+	if err != nil {
+		return err
+	}
+	// TODO: Don't ignore context.
+	return res.Wait()
 }
 
 func (c *RealtimeChannel) detach(result bool) (Result, error) {
@@ -297,30 +303,62 @@ func (c *RealtimeChannel) detach(result bool) (Result, error) {
 	return res, nil
 }
 
-// Subscribe subscribes to a realtime channel, which makes any newly received
-// messages relayed to the returned Subscription value.
+type subscriptionName string
+
+func (subscriptionName) isEmitterEvent() {}
+
+type subscriptionMessage Message
+
+func (*subscriptionMessage) isEmitterData() {}
+
+// Subscribe registers a message handler to be called with each message with the
+// given name received from the channel.
 //
-// If no names are given, returned Subscription will receive all messages.
-// If ch is non-nil and it was already registered to receive messages with different
-// names than the ones given, it will be added to receive also the new ones.
-func (c *RealtimeChannel) Subscribe(names ...string) (*Subscription, error) {
-	if _, err := c.attach(false); err != nil {
+// This implicitly attaches the channel if it's not already attached. If the
+// context is canceled before the attach operation finishes, the call
+// returns with an error, but the operation carries on in the background and
+// the channel may eventually be attached anyway.
+//
+// See package-level documentation on Event Emitter for details about
+// messages dispatch.
+func (c *RealtimeChannel) Subscribe(ctx context.Context, name string, handle func(*Message)) (unsubscribe func(), err error) {
+	res, err := c.attach(true)
+	if err != nil {
 		return nil, err
 	}
-	return c.subs.subscribe(namesToKeys(names)...)
+	// TODO: Don't ignore context.
+	err = res.Wait()
+	if err != nil {
+		return nil, err
+	}
+	return c.messageEmitter.On(subscriptionName(name), func(message emitterData) {
+		handle((*Message)(message.(*subscriptionMessage)))
+	}), nil
 }
 
-// Unsubscribe removes previous Subscription for the given message names.
+// SubscribeAll register a message handler to be called with each message
+// received from the channel.
 //
-// Unsubscribe panics if the given sub was subscribed for presence messages and
-// not for regular channel messages.
+// This implicitly attaches the channel if it's not already attached. If the
+// context is canceled before the attach operation finishes, the call
+// returns with an error, but the operation carries on in the background and
+// the channel may eventually be attached anyway.
 //
-// If sub was already unsubscribed, the method is a nop.
-func (c *RealtimeChannel) Unsubscribe(sub *Subscription, names ...string) {
-	if sub.typ != subscriptionMessages {
-		panic(errInvalidType{typ: sub.typ})
+// See package-level documentation on Event Emitter for details about
+// messages dispatch.
+func (c *RealtimeChannel) SubscribeAll(ctx context.Context, handle func(*Message)) (unsubscribe func(), err error) {
+	res, err := c.attach(true)
+	if err != nil {
+		return nil, err
+	}
+	// TODO: Don't ignore context.
+	err = res.Wait()
+	if err != nil {
+		return nil, err
 	}
-	c.subs.unsubscribe(true, sub, namesToKeys(names)...)
+	return c.messageEmitter.OnAll(func(message emitterData) {
+		handle((*Message)(message.(*subscriptionMessage)))
+	}), nil
 }
 
 type channelStateChanges chan ChannelStateChange
@@ -383,24 +421,28 @@ func (em ChannelEventEmitter) OffAll() {
 	em.emitter.OffAll()
 }
 
-// Publish publishes a message on the channel, which is send on separate
-// goroutine. Publish does not block.
+// Publish publishes a message on the channel.
 //
-// This implicitly attaches the channel if it's not already attached.
-func (c *RealtimeChannel) Publish(name string, data interface{}) (Result, error) {
-	return c.PublishAll([]*proto.Message{{Name: name, Data: data}})
+// This implicitly attaches the channel if it's not already attached. If the
+// context is canceled before the attach operation finishes, the call
+// returns with an error, but the operation carries on in the background and
+// the channel may eventually be attached and the message published anyway.
+func (c *RealtimeChannel) Publish(ctx context.Context, name string, data interface{}) error {
+	return c.PublishBatch(ctx, []*Message{{Name: name, Data: data}})
 }
 
-// PublishAll publishes all given messages on the channel at once.
-// PublishAll does not block.
+// PublishBatch publishes all given messages on the channel at once.
 //
-// This implicitly attaches the channel if it's not already attached.
-func (c *RealtimeChannel) PublishAll(messages []*proto.Message) (Result, error) {
+// This implicitly attaches the channel if it's not already attached. If the
+// context is canceled before the attach operation finishes, the call
+// returns with an error, but the operation carries on in the background and
+// the channel may eventually be attached and the message published anyway.
+func (c *RealtimeChannel) PublishBatch(ctx context.Context, messages []*Message) error {
 	id := c.client.Auth.clientIDForCheck()
 	for _, v := range messages {
 		if v.ClientID != "" && id != wildcardClientID && v.ClientID != id {
 			// Spec RSL1g3,RSL1g4
-			return nil, fmt.Errorf("Unable to publish message containing a clientId (%s) that is incompatible with the library clientId (%s)", v.ClientID, id)
+			return fmt.Errorf("Unable to publish message containing a clientId (%s) that is incompatible with the library clientId (%s)", v.ClientID, id)
 		}
 	}
 	msg := &proto.ProtocolMessage{
@@ -408,7 +450,12 @@ func (c *RealtimeChannel) PublishAll(messages []*proto.Message) (Result, error)
 		Channel:  c.Name,
 		Messages: messages,
 	}
-	return c.send(msg)
+	res, err := c.send(msg)
+	if err != nil {
+		return err
+	}
+	// TODO: Don't ignore context.
+	return res.Wait()
 }
 
 // History gives the channel's message history according to the given parameters.
@@ -445,7 +492,7 @@ func (c *RealtimeChannel) State() ChannelState {
 }
 
 // Reason gives the last error that caused channel transition to failed state.
-func (c *RealtimeChannel) Reason() error {
+func (c *RealtimeChannel) Reason() *ErrorInfo {
 	c.mtx.Lock()
 	defer c.mtx.Unlock()
 	return c.errorReason
@@ -504,7 +551,9 @@ func (c *RealtimeChannel) notify(msg *proto.ProtocolMessage) {
 		c.setState(ChannelStateFailed, newErrorFromProto(msg.Error))
 		c.queue.Fail(newErrorFromProto(msg.Error))
 	case proto.ActionMessage:
-		c.subs.messageEnqueue(msg)
+		for _, msg := range msg.Messages {
+			c.messageEmitter.Emit(subscriptionName(msg.Name), (*subscriptionMessage)(msg))
+		}
 	default:
 	}
 }
@@ -585,8 +634,5 @@ func (c *RealtimeChannel) lockSetState(state ChannelState, err error) error {
 	}
 	c.internalEmitter.emitter.Emit(change.Event, change)
 	c.emitter.Emit(change.Event, change)
-	if c.errorReason == nil {
-		return nil
-	}
-	return c.errorReason
+	return c.errorReason.unwrapNil()
 }
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index d1f6971..f0f5995 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -421,7 +421,7 @@ func (c *Connection) send(msg *proto.ProtocolMessage, listen chan<- error) error
 		return err
 	}
 	c.updateSerial(msg, listen)
-	c.mtx.Unlock()
+	defer c.mtx.Unlock()
 	return c.conn.Send(msg)
 }
 
@@ -716,8 +716,5 @@ func (c *Connection) lockSetState(state ConnectionState, err error) error {
 	}
 	c.internalEmitter.emitter.Emit(change.Event, change)
 	c.emitter.Emit(change.Event, change)
-	if c.errorReason == nil {
-		return nil
-	}
-	return c.errorReason
+	return c.errorReason.unwrapNil()
 }
diff --git a/ably/realtime_presence.go b/ably/realtime_presence.go
index eb7f578..6acaf8f 100644
--- a/ably/realtime_presence.go
+++ b/ably/realtime_presence.go
@@ -1,6 +1,7 @@
 package ably
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"sync"
@@ -20,24 +21,24 @@ const (
 // It allows entering, leaving and updating presence state for the current
 // client or on behalf of other client.
 type RealtimePresence struct {
-	mtx       sync.Mutex
-	data      interface{}
-	serial    string
-	subs      *subscriptions
-	channel   *RealtimeChannel
-	members   map[string]*proto.PresenceMessage
-	stale     map[string]struct{}
-	state     proto.PresenceState
-	syncMtx   sync.Mutex
-	syncState syncState
+	mtx            sync.Mutex
+	data           interface{}
+	serial         string
+	messageEmitter *eventEmitter
+	channel        *RealtimeChannel
+	members        map[string]*proto.PresenceMessage
+	stale          map[string]struct{}
+	state          proto.PresenceAction
+	syncMtx        sync.Mutex
+	syncState      syncState
 }
 
 func newRealtimePresence(channel *RealtimeChannel) *RealtimePresence {
 	pres := &RealtimePresence{
-		subs:      newSubscriptions(subscriptionPresenceMessages, channel.logger()),
-		channel:   channel,
-		members:   make(map[string]*proto.PresenceMessage),
-		syncState: syncInitial,
+		messageEmitter: newEventEmitter(channel.logger()),
+		channel:        channel,
+		members:        make(map[string]*proto.PresenceMessage),
+		syncState:      syncInitial,
 	}
 	// Lock syncMtx to make all callers to Get(true) wait until the presence
 	// is in initial sync state. This is to not make them early return
@@ -130,7 +131,7 @@ func (pres *RealtimePresence) syncEnd() {
 		delete(pres.members, memberKey)
 	}
 	for memberKey, presence := range pres.members {
-		if presence.State == proto.PresenceAbsent {
+		if presence.Action == proto.PresenceAbsent {
 			delete(pres.members, memberKey)
 		}
 	}
@@ -161,11 +162,9 @@ func (pres *RealtimePresence) processIncomingMessage(msg *proto.ProtocolMessage,
 				continue // do not process old message
 			}
 		}
-		switch member.State {
+		switch member.Action {
 		case proto.PresenceUpdate:
-			memberCopy := *member
-			member = &memberCopy
-			member.State = proto.PresencePresent
+			member.Action = proto.PresencePresent
 			fallthrough
 		case proto.PresencePresent:
 			delete(pres.stale, memberKey)
@@ -181,7 +180,22 @@ func (pres *RealtimePresence) processIncomingMessage(msg *proto.ProtocolMessage,
 	pres.mtx.Unlock()
 	msg.Count = len(messages)
 	msg.Presence = messages
-	pres.subs.presenceEnqueue(msg)
+	for _, msg := range msg.Presence {
+		var action PresenceAction
+		switch msg.Action {
+		case proto.PresenceAbsent:
+			action = PresenceActionAbsent
+		case proto.PresencePresent:
+			action = PresenceActionPresent
+		case proto.PresenceEnter:
+			action = PresenceActionEnter
+		case proto.PresenceLeave:
+			action = PresenceActionLeave
+		case proto.PresenceUpdate:
+			action = PresenceActionUpdate
+		}
+		pres.messageEmitter.Emit(action, (*subscriptionPresenceMessage)(msg))
+	}
 }
 
 // Get returns a list of current members on the channel.
@@ -204,22 +218,79 @@ func (pres *RealtimePresence) Get(wait bool) ([]*proto.PresenceMessage, error) {
 	return members, nil
 }
 
-// Subscribe subscribes to presence events on the associated channel.
+// A PresenceAction is a kind of action involving presence in a channel.
+type PresenceAction struct {
+	name string
+}
+
+var (
+	PresenceActionAbsent  PresenceAction = PresenceAction{name: "ABSENT"}
+	PresenceActionPresent PresenceAction = PresenceAction{name: "PRESENT"}
+	PresenceActionEnter   PresenceAction = PresenceAction{name: "ENTER"}
+	PresenceActionLeave   PresenceAction = PresenceAction{name: "LEAVE"}
+	PresenceActionUpdate  PresenceAction = PresenceAction{name: "UPDATE"}
+)
+
+func (e PresenceAction) String() string {
+	return e.name
+}
+
+func (PresenceAction) isEmitterEvent() {}
+
+type PresenceMessage = proto.PresenceMessage
+
+type subscriptionPresenceMessage PresenceMessage
+
+func (*subscriptionPresenceMessage) isEmitterData() {}
+
+// Subscribe registers a presence message handler to be called with each
+// presence message with the given action received from the channel.
 //
-// If the channel is not attached, Subscribe implicitly attaches it.
-// If no presence states are given, Subscribe subscribes to all of them.
-func (pres *RealtimePresence) Subscribe(states ...proto.PresenceState) (*Subscription, error) {
-	if _, err := pres.channel.attach(false); err != nil {
+// This implicitly attaches the channel if it's not already attached. If the
+// context is canceled before the attach operation finishes, the call
+// returns with an error, but the operation carries on in the background and
+// the channel may eventually be attached anyway.
+//
+// See package-level documentation on Event Emitter for details about
+// messages dispatch.
+func (pres *RealtimePresence) Subscribe(ctx context.Context, action PresenceAction, handle func(*PresenceMessage)) (unsubscribe func(), err error) {
+	res, err := pres.channel.attach(true)
+	if err != nil {
+		return nil, err
+	}
+	// TODO: Don't ignore context.
+	err = res.Wait()
+	if err != nil {
 		return nil, err
 	}
-	return pres.subs.subscribe(statesToKeys(states)...)
+	return pres.messageEmitter.On(action, func(message emitterData) {
+		handle((*PresenceMessage)(message.(*subscriptionPresenceMessage)))
+	}), nil
 }
 
-// Unsubscribe removes previous Subscription for the given presence states.
+// Subscribe registers a presence message handler to be called with each
+// presence message received from the channel.
+//
+// This implicitly attaches the channel if it's not already attached. If the
+// context is canceled before the attach operation finishes, the call
+// returns with an error, but the operation carries on in the background and
+// the channel may eventually be attached anyway.
 //
-// If sub was already unsubscribed, the method is a nop.
-func (pres *RealtimePresence) Unsubscribe(sub *Subscription, states ...proto.PresenceState) {
-	pres.subs.unsubscribe(true, sub, statesToKeys(states)...)
+// See package-level documentation on Event Emitter for details about
+// messages dispatch.
+func (pres *RealtimePresence) SubscribeAll(ctx context.Context, handle func(*PresenceMessage)) (unsubscribe func(), err error) {
+	res, err := pres.channel.attach(true)
+	if err != nil {
+		return nil, err
+	}
+	// TODO: Don't ignore context.
+	err = res.Wait()
+	if err != nil {
+		return nil, err
+	}
+	return pres.messageEmitter.OnAll(func(message emitterData) {
+		handle((*PresenceMessage)(message.(*subscriptionPresenceMessage)))
+	}), nil
 }
 
 // Enter announces presence of the current client with an enter message
@@ -261,12 +332,12 @@ func (pres *RealtimePresence) EnterClient(clientID string, data interface{}) (Re
 	pres.data = data
 	pres.state = proto.PresenceEnter
 	pres.mtx.Unlock()
-	msg := &proto.PresenceMessage{
-		State: proto.PresenceEnter,
+	msg := proto.PresenceMessage{
+		Action: proto.PresenceEnter,
 	}
 	msg.Data = data
 	msg.ClientID = clientID
-	return pres.send(msg)
+	return pres.send(&msg)
 }
 
 func nonnil(a, b interface{}) interface{} {
@@ -289,12 +360,12 @@ func (pres *RealtimePresence) UpdateClient(clientID string, data interface{}) (R
 	}
 	pres.data = data
 	pres.mtx.Unlock()
-	msg := &proto.PresenceMessage{
-		State: proto.PresenceUpdate,
+	msg := proto.PresenceMessage{
+		Action: proto.PresenceUpdate,
 	}
 	msg.ClientID = clientID
 	msg.Data = data
-	return pres.send(msg)
+	return pres.send(&msg)
 }
 
 // LeaveClient announces the given clientID leave the associated channel altogether
@@ -310,12 +381,12 @@ func (pres *RealtimePresence) LeaveClient(clientID string, data interface{}) (Re
 	}
 	pres.mtx.Unlock()
 
-	msg := &proto.PresenceMessage{
-		State: proto.PresenceLeave,
+	msg := proto.PresenceMessage{
+		Action: proto.PresenceLeave,
 	}
 	msg.ClientID = clientID
 	msg.Data = data
-	return pres.send(msg)
+	return pres.send(&msg)
 }
 
 func (pres *RealtimePresence) auth() *Auth {
diff --git a/ably/rest_channel.go b/ably/rest_channel.go
index 2a37713..a493960 100644
--- a/ably/rest_channel.go
+++ b/ably/rest_channel.go
@@ -1,7 +1,9 @@
 package ably
 
 import (
+	"context"
 	"fmt"
+	"net/url"
 	"strings"
 
 	"github.com/ably/ably-go/ably/internal/ablyutil"
@@ -24,45 +26,76 @@ var encodeURIComponent = strings.NewReplacer(
 	"#", "%23",
 )
 
-type RestChannel struct {
+// RESTChannel is the interface for REST API operations on a channel.
+type RESTChannel struct {
 	Name     string
-	Presence *RestPresence
+	Presence *RESTPresence
 
 	client  *REST
 	baseURL string
 	options *proto.ChannelOptions
 }
 
-func newRestChannel(name string, client *REST) *RestChannel {
-	c := &RestChannel{
+func newRESTChannel(name string, client *REST) *RESTChannel {
+	c := &RESTChannel{
 		Name:    name,
 		client:  client,
 		baseURL: "/channels/" + encodeURIComponent.Replace(name),
 	}
-	c.Presence = &RestPresence{
+	c.Presence = &RESTPresence{
 		client:  client,
 		channel: c,
 	}
 	return c
 }
 
-func (c *RestChannel) Publish(name string, data interface{}) error {
-	messages := []*proto.Message{
+// Publish publishes a message on the channel.
+func (c *RESTChannel) Publish(ctx context.Context, name string, data interface{}) error {
+	return c.PublishBatch(ctx, []*Message{
 		{Name: name, Data: data},
+	})
+}
+
+// Message is what Ably channels send and receive.
+type Message = proto.Message
+
+// PublishBatch publishes multiple messages in a batch.
+func (c *RESTChannel) PublishBatch(ctx context.Context, messages []*Message) error {
+	return c.PublishBatchWithOptions(ctx, messages)
+}
+
+// PublishBatchOptions is an optional parameter for
+// RESTChannel.PublishBatchWithOptions.
+type PublishBatchOption func(*publishBatchOptions)
+
+type publishBatchOptions struct {
+	params map[string]string
+}
+
+// Params adds query parameters to the resulting HTTP request to the REST API.
+func PublishBatchWithParams(params map[string]string) PublishBatchOption {
+	return func(options *publishBatchOptions) {
+		options.params = params
 	}
-	return c.PublishAll(messages)
 }
 
-// PublishAll sends multiple messages in the same http call.
-// This is the more efficient way of transmitting a batch of messages
-// using the Rest API.
-func (c *RestChannel) PublishAll(messages []*proto.Message) error {
+// 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
 		}
 	}
-	useIdempotent := c.client.opts.idempotentRestPublishing()
+	useIdempotent := c.client.opts.idempotentRESTPublishing()
 	if useIdempotent {
 		switch len(messages) {
 		case 1:
@@ -93,7 +126,15 @@ func (c *RestChannel) PublishAll(messages []*proto.Message) error {
 			}
 		}
 	}
-	res, err := c.client.post(c.baseURL+"/messages", messages, nil)
+	var query string
+	if params := publishOpts.params; len(params) > 0 {
+		queryParams := url.Values{}
+		for k, v := range params {
+			queryParams.Set(k, v)
+		}
+		query = "?" + queryParams.Encode()
+	}
+	res, err := c.client.post(c.baseURL+"/messages"+query, messages, nil)
 	if err != nil {
 		return err
 	}
@@ -103,7 +144,7 @@ func (c *RestChannel) PublishAll(messages []*proto.Message) error {
 // 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(params *PaginateParams) (*PaginatedResult, error) {
+func (c *RESTChannel) History(params *PaginateParams) (*PaginatedResult, error) {
 	path := c.baseURL + "/history"
 	rst, err := newPaginatedResult(c.options, paginatedRequest{typ: msgType, path: path, params: params, query: query(c.client.get), logger: c.logger(), respCheck: checkValidHTTPResponse})
 	if err != nil {
@@ -112,6 +153,6 @@ func (c *RestChannel) History(params *PaginateParams) (*PaginatedResult, error)
 	return rst, nil
 }
 
-func (c *RestChannel) logger() *LoggerOptions {
+func (c *RESTChannel) logger() *LoggerOptions {
 	return c.client.logger()
 }
diff --git a/ably/rest_client.go b/ably/rest_client.go
index 3e80e5a..aa575bc 100644
--- a/ably/rest_client.go
+++ b/ably/rest_client.go
@@ -36,10 +36,10 @@ func query(fn func(string, interface{}) (*http.Response, error)) queryFunc {
 	}
 }
 
-// RestChannels provides an API for managing collection of RestChannel. This is
+// RESTChannels provides an API for managing collection of RESTChannel. This is
 // safe for concurrent use.
-type RestChannels struct {
-	cache  map[string]*RestChannel
+type RESTChannels struct {
+	cache  map[string]*RESTChannel
 	mu     sync.RWMutex
 	client *REST
 }
@@ -47,12 +47,12 @@ type RestChannels struct {
 // Range iterates over the channels calling fn on every iteration. If fn returns
 // false then the iteration is stopped.
 //
-// This uses locking to take a snapshot of the underlying RestChannel map before
+// This uses locking to take a snapshot of the underlying RESTChannel map before
 // iteration to avoid any deadlock, meaning any modification (like creating new
-// RestChannel, or removing one) that occurs during iteration will not have any
+// RESTChannel, or removing one) that occurs during iteration will not have any
 // effect to the values passed to the fn.
-func (c *RestChannels) Range(fn func(name string, channel *RestChannel) bool) {
-	clone := make(map[string]*RestChannel)
+func (c *RESTChannels) Range(fn func(name string, channel *RESTChannel) bool) {
+	clone := make(map[string]*RESTChannel)
 	c.mu.RLock()
 	for k, v := range c.cache {
 		clone[k] = v
@@ -67,7 +67,7 @@ func (c *RestChannels) Range(fn func(name string, channel *RestChannel) bool) {
 }
 
 // Exists returns true if the channel by the given name exists.
-func (c *RestChannels) Exists(name string) bool {
+func (c *RESTChannels) Exists(name string) bool {
 	c.mu.RLock()
 	_, ok := c.cache[name]
 	c.mu.RUnlock()
@@ -79,7 +79,7 @@ func (c *RestChannels) Exists(name string) bool {
 // You can optionally pass ChannelOptions, if the channel exists it will
 // updated with the options and when it doesn't a new channel will be created
 // with the given options.
-func (c *RestChannels) Get(name string, options ...ChannelOption) *RestChannel {
+func (c *RESTChannels) Get(name string, options ...ChannelOption) *RESTChannel {
 	var o channelOptions
 	for _, set := range options {
 		set(&o)
@@ -87,7 +87,7 @@ func (c *RestChannels) Get(name string, options ...ChannelOption) *RestChannel {
 	return c.get(name, (*proto.ChannelOptions)(&o))
 }
 
-func (c *RestChannels) get(name string, opts *proto.ChannelOptions) *RestChannel {
+func (c *RESTChannels) get(name string, opts *proto.ChannelOptions) *RESTChannel {
 	c.mu.RLock()
 	v, ok := c.cache[name]
 	c.mu.RUnlock()
@@ -97,7 +97,7 @@ func (c *RestChannels) get(name string, opts *proto.ChannelOptions) *RestChannel
 		}
 		return v
 	}
-	v = newRestChannel(name, c.client)
+	v = newRESTChannel(name, c.client)
 	v.options = opts
 	c.mu.Lock()
 	c.cache[name] = v
@@ -106,14 +106,14 @@ func (c *RestChannels) get(name string, opts *proto.ChannelOptions) *RestChannel
 }
 
 // Release deletes the channel from the cache.
-func (c *RestChannels) Release(ch *RestChannel) {
+func (c *RESTChannels) Release(ch *RESTChannel) {
 	c.mu.Lock()
 	delete(c.cache, ch.Name)
 	c.mu.Unlock()
 }
 
 // Len returns the number of channels stored.
-func (c *RestChannels) Len() (size int) {
+func (c *RESTChannels) Len() (size int) {
 	c.mu.RLock()
 	size = len(c.cache)
 	c.mu.RUnlock()
@@ -122,7 +122,7 @@ func (c *RestChannels) Len() (size int) {
 
 type REST struct {
 	Auth                *Auth
-	Channels            *RestChannels
+	Channels            *RESTChannels
 	opts                *clientOptions
 	successFallbackHost *fallbackCache
 }
@@ -137,8 +137,8 @@ func NewREST(options ClientOptions) (*REST, error) {
 		return nil, err
 	}
 	c.Auth = auth
-	c.Channels = &RestChannels{
-		cache:  make(map[string]*RestChannel),
+	c.Channels = &RESTChannels{
+		cache:  make(map[string]*RESTChannel),
 		client: c,
 	}
 	return c, nil
@@ -345,7 +345,7 @@ func (c *REST) doWithHandle(r *request, handle func(*http.Response, interface{})
 		log.Error("RestClient: error handling response: ", err)
 		if e, ok := err.(*ErrorInfo); ok {
 			if canFallBack(e.StatusCode) &&
-				(strings.HasPrefix(req.URL.Host, defaultOptions.RestHost) ||
+				(strings.HasPrefix(req.URL.Host, defaultOptions.RESTHost) ||
 					c.opts.FallbackHosts != nil) {
 				fallback := defaultFallbackHosts()
 				if c.opts.FallbackHosts != nil {
diff --git a/ably/rest_presence.go b/ably/rest_presence.go
index aa73f38..8f4e566 100644
--- a/ably/rest_presence.go
+++ b/ably/rest_presence.go
@@ -1,14 +1,14 @@
 package ably
 
-type RestPresence struct {
+type RESTPresence struct {
 	client  *REST
-	channel *RestChannel
+	channel *RESTChannel
 }
 
 // Get gives the channel's presence messages according to the given parameters.
 // The returned result can be inspected for the presence messages via
 // the PresenceMessages() method.
-func (p *RestPresence) Get(params *PaginateParams) (*PaginatedResult, error) {
+func (p *RESTPresence) Get(params *PaginateParams) (*PaginatedResult, error) {
 	path := p.channel.baseURL + "/presence"
 	return newPaginatedResult(nil, paginatedRequest{typ: presMsgType, path: path, params: params, query: query(p.client.get), logger: p.logger(), respCheck: checkValidHTTPResponse})
 }
@@ -16,11 +16,11 @@ func (p *RestPresence) Get(params *PaginateParams) (*PaginatedResult, error) {
 // 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(params *PaginateParams) (*PaginatedResult, error) {
+func (p *RESTPresence) History(params *PaginateParams) (*PaginatedResult, error) {
 	path := p.channel.baseURL + "/presence/history"
 	return newPaginatedResult(nil, paginatedRequest{typ: presMsgType, path: path, params: params, query: query(p.client.get), logger: p.logger(), respCheck: checkValidHTTPResponse})
 }
 
-func (p *RestPresence) logger() *LoggerOptions {
+func (p *RESTPresence) logger() *LoggerOptions {
 	return p.client.logger()
 }
diff --git a/ably/state.go b/ably/state.go
index 5924e78..43abec2 100644
--- a/ably/state.go
+++ b/ably/state.go
@@ -39,6 +39,8 @@ var (
 	errSuspended      = newErrorf(80002, "Connection unavailable")
 	errFailed         = newErrorf(80000, "Connection failed")
 	errNeverConnected = newErrorf(80002, "Unable to establish connection")
+
+	errSerialSkipped = newErrorf(ErrInternalError, "Serial for message was skipped by acknowledgement")
 )
 
 var connStateErrors = map[ConnectionState]ErrorInfo{
@@ -51,7 +53,7 @@ var connStateErrors = map[ConnectionState]ErrorInfo{
 func connStateError(state ConnectionState, err error) *ErrorInfo {
 	// Set default error information associated with the target state.
 	e, ok := err.(*ErrorInfo)
-	if ok {
+	if ok && e != nil {
 		return e
 	}
 	if e, ok := connStateErrors[state]; ok {
@@ -70,7 +72,7 @@ var channelStateErrors = map[ChannelState]ErrorInfo{
 func channelStateError(state ChannelState, err error) *ErrorInfo {
 	// Set default error information associated with the target state.
 	e, ok := err.(*ErrorInfo)
-	if ok {
+	if ok && e != nil {
 		return e
 	}
 	if e, ok := channelStateErrors[state]; ok {
@@ -128,7 +130,7 @@ func (q *pendingEmitter) Enqueue(serial int64, ch chan<- error) {
 	}
 }
 
-func (q *pendingEmitter) Ack(serial int64, count int, err error) {
+func (q *pendingEmitter) Ack(serial int64, count int, errInfo *ErrorInfo) {
 	if q.Len() == 0 {
 		return
 	}
@@ -144,8 +146,9 @@ func (q *pendingEmitter) Ack(serial int64, count int, err error) {
 		nack = i + 1
 		ack = min(i+1+count, q.Len())
 	}
-	if err == nil {
-		err = newError(50000, err)
+	err := errInfo.unwrapNil()
+	if err == nil && nack > 0 {
+		err = errSerialSkipped
 	}
 	for _, sch := range q.queue[:nack] {
 		q.logger.Printf(LogVerbose, "received NACK for message serial %d", sch.serial)
@@ -158,7 +161,7 @@ func (q *pendingEmitter) Ack(serial int64, count int, err error) {
 	q.queue = q.queue[ack:]
 }
 
-func (q *pendingEmitter) Nack(serial int64, count int, err error) {
+func (q *pendingEmitter) Nack(serial int64, count int, errInfo *ErrorInfo) {
 	if q.Len() == 0 {
 		return
 	}
@@ -171,9 +174,7 @@ func (q *pendingEmitter) Nack(serial int64, count int, err error) {
 	default:
 		nack = min(i+1+count, q.Len())
 	}
-	if err == nil {
-		err = newError(50000, err)
-	}
+	err := errInfo.unwrapNil()
 	for _, sch := range q.queue[:nack] {
 		q.logger.Printf(LogVerbose, "received NACK for message serial %d", sch.serial)
 		sch.ch <- err
diff --git a/ably/subscription.go b/ably/subscription.go
deleted file mode 100644
index 9a07f80..0000000
--- a/ably/subscription.go
+++ /dev/null
@@ -1,269 +0,0 @@
-package ably
-
-import (
-	"reflect"
-	"sync"
-
-	"github.com/ably/ably-go/ably/proto"
-)
-
-var (
-	subscriptionMessages         = reflect.TypeOf((chan *proto.Message)(nil))
-	subscriptionPresenceMessages = reflect.TypeOf((chan *proto.PresenceMessage)(nil))
-)
-
-// Subscription queues messages received from a realtime channel.
-type Subscription struct {
-	typ         reflect.Type
-	mtx         sync.Mutex
-	channel     interface{}
-	sleep       chan struct{}
-	queue       []interface{}
-	unsubscribe func(*Subscription)
-	stopped     bool
-	logger      *LoggerOptions
-}
-
-func newSubscription(typ reflect.Type, unsubscribe func(*Subscription), log *LoggerOptions) *Subscription {
-	sub := &Subscription{
-		typ:         typ,
-		channel:     reflect.MakeChan(typ, 0).Interface(),
-		sleep:       make(chan struct{}, 1),
-		unsubscribe: unsubscribe,
-		logger:      log,
-	}
-	go sub.loop()
-	return sub
-}
-
-// MessageChannel gives a channel on which the messages are delivered.
-// It panics when sub was not subscribed to receive channel's messages.
-func (sub *Subscription) MessageChannel() <-chan *proto.Message {
-	if sub.typ != subscriptionMessages {
-		panic(errInvalidType{typ: sub.typ})
-	}
-	ch := sub.channel.(chan *proto.Message)
-	return ch
-}
-
-// PresenceChannel gives a channel on which the presence messages are delivered.
-// It panics when sub was not subscribed to receive channel's presence messages.
-func (sub *Subscription) PresenceChannel() <-chan *proto.PresenceMessage {
-	if sub.typ != subscriptionPresenceMessages {
-		panic(errInvalidType{typ: sub.typ})
-	}
-	ch := sub.channel.(chan *proto.PresenceMessage)
-	return ch
-}
-
-// Close unsubscribes from the realtime channel the sub was previously subscribed.
-// It closes the chan returned by C method.
-func (sub *Subscription) Close() error {
-	return sub.close(true)
-}
-
-func (sub *Subscription) drain() {
-	v := reflect.ValueOf(sub.channel)
-	for {
-		if _, ok := v.TryRecv(); !ok {
-			v.Close()
-			return
-		}
-	}
-}
-
-func (sub *Subscription) close(unsubscribe bool) error {
-	if unsubscribe {
-		sub.unsubscribe(sub)
-	}
-	sub.mtx.Lock()
-	if sub.stopped {
-		sub.mtx.Unlock()
-		return nil
-	}
-	sub.stopped = true
-	sub.queue = nil
-	close(sub.sleep)
-	sub.mtx.Unlock()
-	sub.drain() // drain sub.channel to stop loop goroutine.
-	return nil
-}
-
-// Len gives a number of messages currently queued.
-func (sub *Subscription) Len() int {
-	sub.mtx.Lock()
-	defer sub.mtx.Unlock()
-	return len(sub.queue)
-}
-
-func (sub *Subscription) enqueue(msg interface{}) {
-	sub.mtx.Lock()
-	defer sub.mtx.Unlock()
-	if sub.stopped {
-		return
-	}
-	sleeping := len(sub.queue) == 0
-	sub.queue = append(sub.queue, msg)
-	if sleeping {
-		sub.sleep <- struct{}{}
-	}
-}
-
-func (sub *Subscription) pop() (msg interface{}, n int) {
-	sub.mtx.Lock()
-	defer sub.mtx.Unlock()
-	if n = len(sub.queue); n == 0 {
-		return nil, 0
-	}
-	msg, sub.queue = sub.queue[0], sub.queue[1:]
-	return msg, n
-}
-
-func (sub *Subscription) loop() {
-	channel := reflect.ValueOf(sub.channel)
-	for range sub.sleep {
-		for msg, n := sub.pop(); n != 0; msg, n = sub.pop() {
-			channel.Send(reflect.ValueOf(msg))
-		}
-	}
-}
-
-var (
-	subsAll     struct{}
-	subsAllKeys = []interface{}{subsAll}
-)
-
-func namesToKeys(names []string) []interface{} {
-	if len(names) == 0 {
-		return nil
-	}
-	keys := make([]interface{}, 0, len(names))
-	for _, name := range names {
-		// Ignore empty names.
-		if name != "" {
-			keys = append(keys, name)
-		}
-	}
-	return keys
-}
-
-func statesToKeys(states []proto.PresenceState) []interface{} {
-	if len(states) == 0 {
-		return nil
-	}
-	keys := make([]interface{}, 0, len(states))
-	for _, state := range states {
-		keys = append(keys, state)
-	}
-	return keys
-}
-
-type subscriptions struct {
-	typ    reflect.Type
-	mtx    sync.Mutex
-	all    map[interface{}]map[*Subscription]struct{}
-	logger *LoggerOptions
-}
-
-func newSubscriptions(typ reflect.Type, log *LoggerOptions) *subscriptions {
-	return &subscriptions{
-		typ:    typ,
-		all:    make(map[interface{}]map[*Subscription]struct{}),
-		logger: log,
-	}
-}
-
-func (subs *subscriptions) close() {
-	for _, subs := range subs.all {
-		for sub := range subs {
-			// Stop is idempotent, no need to keep track which sub was already
-			// stopped.
-			sub.close(false)
-		}
-	}
-	// Unsubscribe all channels by creating new sub map for future use.
-	subs.all = make(map[interface{}]map[*Subscription]struct{})
-}
-
-func (subs *subscriptions) subscribe(keys ...interface{}) (*Subscription, error) {
-	unsubscribe := func(sub *Subscription) { subs.unsubscribe(false, sub, keys...) }
-	sub := newSubscription(subs.typ, unsubscribe, subs.logger)
-	if len(keys) == 0 {
-		keys = subsAllKeys
-	}
-	subs.mtx.Lock()
-	for _, key := range keys {
-		all, ok := subs.all[key]
-		if !ok {
-			all = make(map[*Subscription]struct{})
-			subs.all[key] = all
-		}
-		all[sub] = struct{}{}
-	}
-	subs.mtx.Unlock()
-	return sub, nil
-}
-
-func (subs *subscriptions) unsubscribe(stop bool, sub *Subscription, keys ...interface{}) {
-	if sub.typ != subs.typ {
-		panic(errInvalidType{typ: sub.typ})
-	}
-	if len(keys) == 0 {
-		keys = subsAllKeys
-	}
-	subs.mtx.Lock()
-	for _, key := range keys {
-		delete(subs.all[key], sub)
-		if len(subs.all[key]) == 0 {
-			delete(subs.all, key)
-		}
-	}
-	// Don't try to stop when we got here from the (*Subscription).Close method.
-	if stop {
-		count := 0
-		for _, all := range subs.all {
-			if _, ok := all[sub]; ok {
-				count++
-			}
-		}
-		// Stop subscription if it no longer listens for messages.
-		if count == 0 {
-			sub.close(false)
-		}
-	}
-	subs.mtx.Unlock()
-}
-
-func (subs *subscriptions) messageEnqueue(msg *proto.ProtocolMessage) {
-	subs.mtx.Lock()
-	for _, msg := range msg.Messages {
-		if subs, ok := subs.all[subsAll]; ok {
-			for sub := range subs {
-				sub.enqueue(msg)
-			}
-		}
-		if subs, ok := subs.all[msg.Name]; ok {
-			for sub := range subs {
-				sub.enqueue(msg)
-			}
-		}
-	}
-	subs.mtx.Unlock()
-}
-
-func (subs *subscriptions) presenceEnqueue(msg *proto.ProtocolMessage) {
-	subs.mtx.Lock()
-	for _, msg := range msg.Presence {
-		if subs, ok := subs.all[subsAll]; ok {
-			for sub := range subs {
-				sub.enqueue(msg)
-			}
-		}
-		if subs, ok := subs.all[msg.State]; ok {
-			for sub := range subs {
-				sub.enqueue(msg)
-			}
-		}
-	}
-	subs.mtx.Unlock()
-}


Test patch:
diff --git a/ably/ablytest/ablytest.go b/ably/ablytest/ablytest.go
index 4c9ab53..1a9e7f3 100644
--- a/ably/ablytest/ablytest.go
+++ b/ably/ablytest/ablytest.go
@@ -1,6 +1,7 @@
 package ablytest
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -92,3 +93,35 @@ func reflectContains(s, sub reflect.Value) bool {
 	}
 	return false
 }
+
+type MessageChannel chan *ably.Message
+
+func (ch MessageChannel) Receive(m *ably.Message) {
+	ch <- m
+}
+
+func ReceiveMessages(channel *ably.RealtimeChannel, name string) (messages <-chan *ably.Message, unsubscribe func(), err error) {
+	ch := make(MessageChannel, 100)
+	if name == "" {
+		unsubscribe, err = channel.SubscribeAll(context.Background(), ch.Receive)
+	} else {
+		unsubscribe, err = channel.Subscribe(context.Background(), name, ch.Receive)
+	}
+	return ch, unsubscribe, err
+}
+
+type PresenceChannel chan *ably.PresenceMessage
+
+func (ch PresenceChannel) Receive(m *ably.PresenceMessage) {
+	ch <- m
+}
+
+func ReceivePresenceMessages(channel *ably.RealtimeChannel, action *ably.PresenceAction) (messages <-chan *ably.PresenceMessage, unsubscribe func(), err error) {
+	ch := make(PresenceChannel, 100)
+	if action == nil {
+		unsubscribe, err = channel.Presence.SubscribeAll(context.Background(), ch.Receive)
+	} else {
+		unsubscribe, err = channel.Presence.Subscribe(context.Background(), *action, ch.Receive)
+	}
+	return ch, unsubscribe, err
+}
diff --git a/ably/ablytest/recorders.go b/ably/ablytest/recorders.go
index 7a21b54..bd406e3 100644
--- a/ably/ablytest/recorders.go
+++ b/ably/ablytest/recorders.go
@@ -514,7 +514,11 @@ func (c realtimeIOCloser) Close() error {
 		ably.ConnectionStateClosed,
 		ably.ConnectionStateFailed:
 
-		return c.c.Connection.ErrorReason()
+		err := c.c.Connection.ErrorReason()
+		if err != nil {
+			return err
+		}
+		return nil
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), Timeout)
diff --git a/ably/ablytest/resultgroup.go b/ably/ablytest/resultgroup.go
index 3946111..e52e57c 100644
--- a/ably/ablytest/resultgroup.go
+++ b/ably/ablytest/resultgroup.go
@@ -65,6 +65,10 @@ func (rg *ResultGroup) Add(res ably.Result, err error) {
 	}()
 }
 
+func (rg *ResultGroup) GoAdd(f ResultFunc) {
+	rg.Add(f.Go(), nil)
+}
+
 func (rg *ResultGroup) Wait() error {
 	if rg.err != nil {
 		return rg.err
@@ -133,3 +137,13 @@ type ResultFunc func() error
 func (f ResultFunc) Wait() error {
 	return f()
 }
+
+func (f ResultFunc) Go() ably.Result {
+	err := make(chan error, 1)
+	go func() {
+		err <- f()
+	}()
+	return ResultFunc(func() error {
+		return <-err
+	})
+}
diff --git a/ably/auth_test.go b/ably/auth_test.go
index de72731..6b1eebc 100644
--- a/ably/auth_test.go
+++ b/ably/auth_test.go
@@ -2,6 +2,7 @@ package ably_test
 
 import (
 	"bytes"
+	"context"
 	"encoding/base64"
 	"errors"
 	"fmt"
@@ -569,18 +570,22 @@ func TestAuth_RequestToken_PublishClientID(t *testing.T) {
 			t.Errorf("%d: want ClientID to be %q; got %s", i, cas.clientID, id)
 			continue
 		}
-		channel := client.Channels.GetAndAttach("publish")
-		sub, err := channel.Subscribe("test")
+		channel := client.Channels.Get("publish")
+		if err := channel.Attach(context.Background()); err != nil {
+			t.Fatal(err)
+		}
+		messages, unsub, err := ablytest.ReceiveMessages(channel, "test")
+		defer unsub()
 		if err != nil {
-			t.Errorf("%d: Subscribe()=%v", i, err)
+			t.Errorf("%d:.Subscribe(context.Background())=%v", i, err)
 			continue
 		}
-		msg := []*proto.Message{{
+		msg := []*ably.Message{{
 			ClientID: cas.publishAs,
 			Name:     "test",
 			Data:     "payload",
 		}}
-		err = ablytest.Wait(channel.PublishAll(msg))
+		err = channel.PublishBatch(context.Background(), msg)
 		if cas.rejected {
 			if err == nil {
 				t.Errorf("%d: expected message to be rejected %#v", i, cas)
@@ -588,11 +593,11 @@ func TestAuth_RequestToken_PublishClientID(t *testing.T) {
 			continue
 		}
 		if err != nil {
-			t.Errorf("%d: PublishAll()=%v", i, err)
+			t.Errorf("%d: PublishBatch()=%v", i, err)
 			continue
 		}
 		select {
-		case msg := <-sub.MessageChannel():
+		case msg := <-messages:
 			if msg.ClientID != cas.publishAs {
 				t.Errorf("%d: want ClientID=%q; got %q", i, cas.publishAs, msg.ClientID)
 			}
@@ -792,7 +797,7 @@ func TestAuth_RealtimeAccessToken(t *testing.T) {
 	if err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
 		t.Fatalf("Connect()=%v", err)
 	}
-	if err := ablytest.Wait(client.Channels.Get("test").Publish("name", "value")); err != nil {
+	if err := client.Channels.Get("test").Publish(context.Background(), "name", "value"); err != nil {
 		t.Fatalf("Publish()=%v", err)
 	}
 	if clientID := client.Auth.ClientID(); clientID != explicitClientID {
diff --git a/ably/export_test.go b/ably/export_test.go
index 7b1c0de..d3238c1 100644
--- a/ably/export_test.go
+++ b/ably/export_test.go
@@ -40,16 +40,6 @@ func UnwrapErrorCode(err error) ErrorCode {
 	return code(err)
 }
 
-// GetAndAttach is a helper method, which returns attached channel or panics if
-// the attaching failed.
-func (ch *Channels) GetAndAttach(name string) *RealtimeChannel {
-	channel := ch.Get(name)
-	if err := wait(channel.Attach()); err != nil {
-		panic(`attach to "` + name + `" failed: ` + err.Error())
-	}
-	return channel
-}
-
 func (a *Auth) Timestamp(query bool) (time.Time, error) {
 	return a.timestamp(query)
 }
diff --git a/ably/proto/presence_message_test.go b/ably/proto/presence_message_test.go
index b533131..ed0bd3b 100644
--- a/ably/proto/presence_message_test.go
+++ b/ably/proto/presence_message_test.go
@@ -10,7 +10,7 @@ import (
 )
 
 func TestPresenceMessage(t *testing.T) {
-	actions := []proto.PresenceState{
+	actions := []proto.PresenceAction{
 		proto.PresenceAbsent,
 		proto.PresencePresent,
 		proto.PresenceEnter,
@@ -25,7 +25,7 @@ func TestPresenceMessage(t *testing.T) {
 			Message: proto.Message{
 				ID: id,
 			},
-			State: a,
+			Action: a,
 		}
 
 		t.Run("json", func(ts *testing.T) {
@@ -41,8 +41,8 @@ func TestPresenceMessage(t *testing.T) {
 			if msg.ID != id {
 				ts.Errorf("expected id to be %s got %s", id, msg.ID)
 			}
-			if msg.State != a {
-				ts.Errorf("expected action to be %d got %d", a, msg.State)
+			if msg.Action != a {
+				ts.Errorf("expected action to be %d got %d", a, msg.Action)
 			}
 		})
 		t.Run("msgpack", func(ts *testing.T) {
@@ -58,8 +58,8 @@ func TestPresenceMessage(t *testing.T) {
 			if msg.ID != id {
 				ts.Errorf("expected id to be %s got %s", id, msg.ID)
 			}
-			if msg.State != a {
-				ts.Errorf("expected action to be %d got %d", a, msg.State)
+			if msg.Action != a {
+				ts.Errorf("expected action to be %d got %d", a, msg.Action)
 			}
 		})
 	}
diff --git a/ably/realtime_channel_spec_test.go b/ably/realtime_channel_spec_test.go
index 66108d5..852b6ed 100644
--- a/ably/realtime_channel_spec_test.go
+++ b/ably/realtime_channel_spec_test.go
@@ -34,7 +34,7 @@ func TestRealtimeChannel_RTL2_ChannelEventForStateChange(t *testing.T) {
 			changes <- change
 		})
 
-		_, err := channel.Attach()
+		err := channel.Attach(context.Background())
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -74,7 +74,7 @@ func TestRealtimeChannel_RTL2_ChannelEventForStateChange(t *testing.T) {
 			changes <- change
 		})
 
-		_, err := channel.Detach()
+		err := channel.Detach(context.Background())
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -101,7 +101,7 @@ func TestRealtimeChannel_RTL2_ChannelEventForStateChange(t *testing.T) {
 			changes <- change
 		})
 
-		_, err := channel.Detach()
+		err := channel.Detach(context.Background())
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -142,7 +142,7 @@ func attachAndWait(t *testing.T, channel *ably.RealtimeChannel) {
 		defer off()
 	}
 
-	_, err := channel.Attach()
+	err := channel.Attach(context.Background())
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -200,10 +200,9 @@ func TestRealtimeChannel_RTL13_HandleDetached(t *testing.T) {
 
 		channel = c.Channels.Get("test")
 
-		_, err = channel.Attach()
-		if err != nil {
-			t.Fatal(err)
-		}
+		ctx, cancel := context.WithCancel(context.Background())
+		go channel.Attach(ctx)
+		defer cancel()
 
 		ablytest.Instantly.Recv(t, nil, out, t.Fatalf) // Consume ATTACH
 
diff --git a/ably/realtime_channel_test.go b/ably/realtime_channel_test.go
index 8c45cf6..5470180 100644
--- a/ably/realtime_channel_test.go
+++ b/ably/realtime_channel_test.go
@@ -1,6 +1,7 @@
 package ably_test
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"net/url"
@@ -41,7 +42,7 @@ func TestRealtimeChannel_Publish(t *testing.T) {
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
 	channel := client.Channels.Get("test")
-	if err := ablytest.Wait(channel.Publish("hello", "world")); err != nil {
+	if err := channel.Publish(context.Background(), "hello", "world"); err != nil {
 		t.Fatalf("Publish()=%v", err)
 	}
 }
@@ -56,44 +57,44 @@ func TestRealtimeChannel_Subscribe(t *testing.T) {
 	channel1 := client1.Channels.Get("test")
 	channel2 := client2.Channels.Get("test")
 
-	if err := ablytest.Wait(channel1.Attach()); err != nil {
+	if err := channel1.Attach(context.Background()); err != nil {
 		t.Fatalf("client1: Attach()=%v", err)
 	}
-	if err := ablytest.Wait(channel2.Attach()); err != nil {
+	if err := channel2.Attach(context.Background()); err != nil {
 		t.Fatalf("client2: Attach()=%v", err)
 	}
 
-	sub1, err := channel1.Subscribe()
+	sub1, unsub1, err := ablytest.ReceiveMessages(channel1, "")
 	if err != nil {
-		t.Fatalf("client1: Subscribe()=%v", err)
+		t.Fatalf("client1:.Subscribe(context.Background())=%v", err)
 	}
-	defer sub1.Close()
+	defer unsub1()
 
-	sub2, err := channel2.Subscribe()
+	sub2, unsub2, err := ablytest.ReceiveMessages(channel2, "")
 	if err != nil {
-		t.Fatalf("client2: Subscribe()=%v", err)
+		t.Fatalf("client2:.Subscribe(context.Background())=%v", err)
 	}
-	defer sub2.Close()
+	defer unsub2()
 
-	if err := ablytest.Wait(channel1.Publish("hello", "client1")); err != nil {
+	if err := channel1.Publish(context.Background(), "hello", "client1"); err != nil {
 		t.Fatalf("client1: Publish()=%v", err)
 	}
-	if err := ablytest.Wait(channel2.Publish("hello", "client2")); err != nil {
+	if err := channel2.Publish(context.Background(), "hello", "client2"); err != nil {
 		t.Fatalf("client2: Publish()=%v", err)
 	}
 
 	timeout := 15 * time.Second
 
-	if err := expectMsg(sub1.MessageChannel(), "hello", "client1", timeout, true); err != nil {
+	if err := expectMsg(sub1, "hello", "client1", timeout, true); err != nil {
 		t.Fatal(err)
 	}
-	if err := expectMsg(sub1.MessageChannel(), "hello", "client2", timeout, true); err != nil {
+	if err := expectMsg(sub1, "hello", "client2", timeout, true); err != nil {
 		t.Fatal(err)
 	}
-	if err := expectMsg(sub2.MessageChannel(), "hello", "client1", timeout, true); err != nil {
+	if err := expectMsg(sub2, "hello", "client1", timeout, true); err != nil {
 		t.Fatal(err)
 	}
-	if err := expectMsg(sub2.MessageChannel(), "hello", "client2", timeout, false); err != nil {
+	if err := expectMsg(sub2, "hello", "client2", timeout, false); err != nil {
 		t.Fatal(err)
 	}
 }
@@ -104,17 +105,17 @@ func TestRealtimeChannel_Detach(t *testing.T) {
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
 	channel := client.Channels.Get("test")
-	sub, err := channel.Subscribe()
+	sub, unsub, err := ablytest.ReceiveMessages(channel, "")
 	if err != nil {
-		t.Fatalf("channel.Subscribe()=%v", err)
+		t.Fatalf("ablytest.ReceiveMessages(channel)=%v", err)
 	}
-	defer sub.Close()
-	if err := ablytest.Wait(channel.Publish("hello", "world")); err != nil {
+	defer unsub()
+	if err := channel.Publish(context.Background(), "hello", "world"); err != nil {
 		t.Fatalf("channel.Publish()=%v", err)
 	}
 	done := make(chan error)
 	go func() {
-		msg, ok := <-sub.MessageChannel()
+		msg, ok := <-sub
 		if !ok {
 			done <- errors.New("did not receive published message")
 		}
@@ -126,8 +127,8 @@ func TestRealtimeChannel_Detach(t *testing.T) {
 	if state := channel.State(); state != ably.ChannelStateAttached {
 		t.Fatalf("want state=%v; got %v", ably.ChannelStateAttached, state)
 	}
-	if err := ablytest.Wait(channel.Detach()); err != nil {
-		t.Fatalf("channel.Detach()=%v", err)
+	if err := channel.Detach(context.Background()); err != nil {
+		t.Fatalf("channel.Detach(context.Background())=%v", err)
 	}
 	if err := ablytest.FullRealtimeCloser(client).Close(); err != nil {
 		t.Fatalf("ablytest.FullRealtimeCloser(client).Close()=%v", err)
@@ -177,11 +178,8 @@ func TestRealtimeChannel_AttachWhileDisconnected(t *testing.T) {
 	channel.On(ably.ChannelEventAttached, attached.Receive)
 
 	res := make(chan ably.Result)
-	var attachErr error
 	go func() {
-		var r ably.Result
-		r, attachErr = channel.Attach()
-		res <- r
+		res <- ablytest.ResultFunc.Go(func() error { return channel.Attach(context.Background()) })
 	}()
 	ablytest.Before(1*time.Second).NoRecv(t, nil, attached, t.Fatalf)
 
@@ -195,8 +193,4 @@ func TestRealtimeChannel_AttachWhileDisconnected(t *testing.T) {
 	if err := ablytest.Wait(<-res, nil); err != nil {
 		t.Fatal(err)
 	}
-
-	if attachErr != nil {
-		t.Fatal(attachErr)
-	}
 }
diff --git a/ably/realtime_client_test.go b/ably/realtime_client_test.go
index d6b01c9..fc0d342 100644
--- a/ably/realtime_client_test.go
+++ b/ably/realtime_client_test.go
@@ -1,6 +1,7 @@
 package ably_test
 
 import (
+	"context"
 	"fmt"
 	"sync"
 	"testing"
@@ -110,8 +111,8 @@ func TestRealtime_multiple(t *testing.T) {
 			rg.Add(ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected), nil)
 			for j := 0; j < 10; j++ {
 				channel := c.Channels.Get(fmt.Sprintf("client/%d/channel/%d", i, j))
-				rg.Add(channel.Attach())
-				rg.Add(channel.Attach())
+				rg.GoAdd(func() error { return channel.Attach(context.Background()) })
+				rg.GoAdd(func() error { return channel.Attach(context.Background()) })
 				rg.Add(channel.Presence.Enter(""))
 			}
 			if err := rg.Wait(); err != nil {
@@ -119,7 +120,10 @@ func TestRealtime_multiple(t *testing.T) {
 			}
 			for j := 0; j < 25; j++ {
 				channel := c.Channels.Get(fmt.Sprintf("client/%d/channel/%d", i, j))
-				rg.Add(channel.Publish(fmt.Sprintf("event/%d/%d", i, j), fmt.Sprintf("data/%d/%d", i, j)))
+				event, data := fmt.Sprintf("event/%d/%d", i, j), fmt.Sprintf("data/%d/%d", i, j)
+				rg.GoAdd(func() error {
+					return channel.Publish(context.Background(), event, data)
+				})
 			}
 			if err := rg.Wait(); err != nil {
 				all.Add(nil, err)
@@ -127,8 +131,8 @@ func TestRealtime_multiple(t *testing.T) {
 			for j := 0; j < 10; j++ {
 				channel := c.Channels.Get(fmt.Sprintf("client/%d/channel/%d", i, j))
 				rg.Add(channel.Presence.Leave(""))
-				rg.Add(channel.Detach())
-				rg.Add(channel.Detach())
+				rg.GoAdd(func() error { return channel.Detach(context.Background()) })
+				rg.GoAdd(func() error { return channel.Detach(context.Background()) })
 			}
 			if err := rg.Wait(); err != nil {
 				all.Add(nil, err)
diff --git a/ably/realtime_conn_spec_test.go b/ably/realtime_conn_spec_test.go
index a47d2b5..191bf48 100644
--- a/ably/realtime_conn_spec_test.go
+++ b/ably/realtime_conn_spec_test.go
@@ -193,14 +193,15 @@ func TestRealtimeConn_RTN15a_ReconnectOnEOF(t *testing.T) {
 
 	channel := client.Channels.Get("channel")
 
-	if err := ablytest.Wait(channel.Attach()); err != nil {
+	if err := channel.Attach(context.Background()); err != nil {
 		t.Fatal(err)
 	}
 
-	sub, err := channel.Subscribe()
+	sub, unsub, err := ablytest.ReceiveMessages(channel, "")
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer unsub()
 
 	stateChanges := make(chan ably.ConnectionStateChange, 16)
 	client.Connection.OnAll(func(c ably.ConnectionStateChange) {
@@ -228,7 +229,7 @@ func TestRealtimeConn_RTN15a_ReconnectOnEOF(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	err = rest.Channels.Get("channel").Publish("name", "data")
+	err = rest.Channels.Get("channel").Publish(context.Background(), "name", "data")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -254,7 +255,7 @@ func TestRealtimeConn_RTN15a_ReconnectOnEOF(t *testing.T) {
 	}
 
 	select {
-	case msg := <-sub.MessageChannel():
+	case msg := <-sub:
 		if expected, got := "data", msg.Data; expected != got {
 			t.Fatalf("expected message with data %v, got %v", expected, got)
 		}
@@ -329,7 +330,7 @@ func TestRealtimeConn_RTN15b(t *testing.T) {
 
 	channel := client.Channels.Get("channel")
 
-	if err := ablytest.Wait(channel.Attach()); err != nil {
+	if err := channel.Attach(context.Background()); err != nil {
 		t.Fatal(err)
 	}
 
@@ -360,7 +361,7 @@ func TestRealtimeConn_RTN15b(t *testing.T) {
 		t.Fatal(err)
 	}
 	goOn := <-gotDial
-	err = rest.Channels.Get("channel").Publish("name", "data")
+	err = rest.Channels.Get("channel").Publish(context.Background(), "name", "data")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -458,14 +459,15 @@ func TestRealtimeConn_RTN15c1(t *testing.T) {
 	}
 
 	channel := client.Channels.Get("channel")
-	if err := ablytest.Wait(channel.Attach()); err != nil {
+	if err := channel.Attach(context.Background()); err != nil {
 		t.Fatal(err)
 	}
 
-	sub, err := channel.Subscribe()
+	sub, unsub, err := ablytest.ReceiveMessages(channel, "")
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer unsub()
 
 	chanStateChanges := make(ably.ChannelStateChanges)
 	off := channel.OnAll(chanStateChanges.Receive)
@@ -493,7 +495,7 @@ func TestRealtimeConn_RTN15c1(t *testing.T) {
 		t.Fatal(err)
 	}
 	goOn := <-gotDial
-	err = rest.Channels.Get("channel").Publish("name", "data")
+	err = rest.Channels.Get("channel").Publish(context.Background(), "name", "data")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -508,7 +510,7 @@ func TestRealtimeConn_RTN15c1(t *testing.T) {
 		t.Fatalf("expected transition to %v, got %v", expected, got)
 	}
 	select {
-	case msg := <-sub.MessageChannel():
+	case msg := <-sub:
 		if expected, got := "data", msg.Data; expected != got {
 			t.Fatalf("expected message with data %v, got %v", expected, got)
 		}
@@ -576,17 +578,18 @@ func TestRealtimeConn_RTN15c2(t *testing.T) {
 	}
 
 	channel := client.Channels.Get("channel")
-	if err := ablytest.Wait(channel.Attach()); err != nil {
+	if err := channel.Attach(context.Background()); err != nil {
 		t.Fatal(err)
 	}
 	chanStateChanges := make(ably.ChannelStateChanges)
 	off := channel.OnAll(chanStateChanges.Receive)
 	defer off()
 
-	sub, err := channel.Subscribe()
+	sub, unsub, err := ablytest.ReceiveMessages(channel, "")
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer unsub()
 
 	stateChanges := make(chan ably.ConnectionStateChange, 16)
 	client.Connection.OnAll(func(c ably.ConnectionStateChange) {
@@ -610,7 +613,7 @@ func TestRealtimeConn_RTN15c2(t *testing.T) {
 		t.Fatal(err)
 	}
 	goOn := <-gotDial
-	err = rest.Channels.Get("channel").Publish("name", "data")
+	err = rest.Channels.Get("channel").Publish(context.Background(), "name", "data")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -626,7 +629,7 @@ func TestRealtimeConn_RTN15c2(t *testing.T) {
 	}
 
 	select {
-	case msg := <-sub.MessageChannel():
+	case msg := <-sub:
 		if expected, got := "data", msg.Data; expected != got {
 			t.Fatalf("expected message with data %v, got %v", expected, got)
 		}
@@ -704,7 +707,7 @@ func TestRealtimeConn_RTN15c3_attached(t *testing.T) {
 	}
 
 	channel := client.Channels.Get("channel")
-	if err := ablytest.Wait(channel.Attach()); err != nil {
+	if err := channel.Attach(context.Background()); err != nil {
 		t.Fatal(err)
 	}
 	chanStateChanges := make(ably.ChannelStateChanges, 18)
@@ -733,7 +736,7 @@ func TestRealtimeConn_RTN15c3_attached(t *testing.T) {
 		t.Fatal(err)
 	}
 	goOn := <-gotDial
-	err = rest.Channels.Get("channel").Publish("name", "data")
+	err = rest.Channels.Get("channel").Publish(context.Background(), "name", "data")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -815,9 +818,17 @@ func TestRealtimeConn_RTN15c3_attaching(t *testing.T) {
 	}
 
 	channel := client.Channels.Get("channel")
-	if _, err := channel.Attach(); err != nil {
-		t.Fatal(err)
-	}
+	attaching := make(ably.ChannelStateChanges, 1)
+	off := channel.On(ably.ChannelEventAttaching, attaching.Receive)
+	defer off()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		channel.Attach(ctx)
+	}()
+
+	ablytest.Soon.Recv(t, nil, attaching, t.Fatalf)
 
 	stateChanges := make(chan ably.ConnectionStateChange, 16)
 	client.Connection.OnAll(func(c ably.ConnectionStateChange) {
@@ -841,7 +852,7 @@ func TestRealtimeConn_RTN15c3_attaching(t *testing.T) {
 		t.Fatal(err)
 	}
 	goOn := <-gotDial
-	err = rest.Channels.Get("channel").Publish("name", "data")
+	err = rest.Channels.Get("channel").Publish(ctx, "name", "data")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -913,7 +924,7 @@ func TestRealtimeConn_RTN15c4(t *testing.T) {
 	}
 
 	channel := client.Channels.Get("channel")
-	if err := ablytest.Wait(channel.Attach()); err != nil {
+	if err := channel.Attach(context.Background()); err != nil {
 		t.Fatal(err)
 	}
 	chanStateChanges := make(ably.ChannelStateChanges, 1)
@@ -942,7 +953,7 @@ func TestRealtimeConn_RTN15c4(t *testing.T) {
 		t.Fatal(err)
 	}
 	goOn := <-gotDial
-	err = rest.Channels.Get("channel").Publish("name", "data")
+	err = rest.Channels.Get("channel").Publish(context.Background(), "name", "data")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -1001,15 +1012,16 @@ func TestRealtimeConn_RTN15d_MessageRecovery(t *testing.T) {
 	}
 
 	channel := client.Channels.Get("test")
-	err := ablytest.Wait(channel.Attach())
+	err := channel.Attach(context.Background())
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	sub, err := channel.Subscribe("test")
+	sub, unsub, err := ablytest.ReceiveMessages(channel, "test")
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer unsub()
 
 	if err := ablytest.ConnWaiter(client, func() {
 		doEOF <- struct{}{}
@@ -1026,7 +1038,7 @@ func TestRealtimeConn_RTN15d_MessageRecovery(t *testing.T) {
 		t.Fatal(err)
 	}
 	for i := 0; i < 3; i++ {
-		err := rest.Channels.Get("test").Publish("test", fmt.Sprintf("msg %d", i))
+		err := rest.Channels.Get("test").Publish(context.Background(), "test", fmt.Sprintf("msg %d", i))
 		if err != nil {
 			t.Fatalf("%d: %v", i, err)
 		}
@@ -1044,8 +1056,8 @@ func TestRealtimeConn_RTN15d_MessageRecovery(t *testing.T) {
 	for i := 0; i < 3; i++ {
 		fatalf := ablytest.FmtFunc(t.Fatalf).Wrap(t, "%d: %s", i)
 
-		var msg *proto.Message
-		ablytest.Soon.Recv(t, &msg, sub.MessageChannel(), fatalf)
+		var msg *ably.Message
+		ablytest.Soon.Recv(t, &msg, sub, fatalf)
 
 		if expected, got := fmt.Sprintf("msg %d", i), msg.Data; expected != got {
 			fatalf("expected %v, got %v", expected, got)
@@ -1153,19 +1165,13 @@ func TestRealtimeConn_RTN15g_NewConnectionOnStateLost(t *testing.T) {
 	// Get channels to ATTACHING, ATTACHED and DETACHED. (TODO: SUSPENDED)
 
 	attaching := c.Channels.Get("attaching")
-	_, err = attaching.Attach()
-	if err != nil {
-		t.Fatal(err)
-	}
+	_ = ablytest.ResultFunc.Go(func() error { return attaching.Attach(context.Background()) })
 	if msg := <-out; msg.Action != proto.ActionAttach {
 		t.Fatalf("expected ATTACH, got %v", msg)
 	}
 
 	attached := c.Channels.Get("attached")
-	attachWaiter, err := attached.Attach()
-	if err != nil {
-		t.Fatal(err)
-	}
+	attachWaiter := ablytest.ResultFunc.Go(func() error { return attached.Attach(context.Background()) })
 	if msg := <-out; msg.Action != proto.ActionAttach {
 		t.Fatalf("expected ATTACH, got %v", msg)
 	}
@@ -1176,10 +1182,7 @@ func TestRealtimeConn_RTN15g_NewConnectionOnStateLost(t *testing.T) {
 	ablytest.Wait(attachWaiter, err)
 
 	detached := c.Channels.Get("detached")
-	attachWaiter, err = detached.Attach()
-	if err != nil {
-		t.Fatal(err)
-	}
+	attachWaiter = ablytest.ResultFunc.Go(func() error { return detached.Attach(context.Background()) })
 	if msg := <-out; msg.Action != proto.ActionAttach {
 		t.Fatalf("expected ATTACH, got %v", msg)
 	}
@@ -1188,10 +1191,7 @@ func TestRealtimeConn_RTN15g_NewConnectionOnStateLost(t *testing.T) {
 		Channel: "detached",
 	}
 	ablytest.Wait(attachWaiter, err)
-	detachWaiter, err := detached.Detach()
-	if err != nil {
-		t.Fatal(err)
-	}
+	detachWaiter := ablytest.ResultFunc.Go(func() error { return detached.Detach(context.Background()) })
 	if msg := <-out; msg.Action != proto.ActionDetach {
 		t.Fatalf("expected DETACH, got %v", msg)
 	}
@@ -1547,10 +1547,7 @@ func TestRealtimeConn_RTN15i_OnErrorWhenConnected(t *testing.T) {
 	// Get a channel to ATTACHED.
 
 	channel := c.Channels.Get("test")
-	attachWaiter, err := channel.Attach()
-	if err != nil {
-		t.Fatal(err)
-	}
+	attachWaiter := ablytest.ResultFunc.Go(func() error { return channel.Attach(context.Background()) })
 
 	_ = <-out // consume ATTACH
 
@@ -1611,10 +1608,10 @@ func TestRealtimeConn_RTN16(t *testing.T) {
 		t.Fatal(err)
 	}
 	channel := c.Channels.Get("channel")
-	if err := ablytest.Wait(channel.Attach()); err != nil {
+	if err := channel.Attach(context.Background()); err != nil {
 		t.Fatal(err)
 	}
-	if err := ablytest.Wait(channel.Publish("name", "data")); err != nil {
+	if err := channel.Publish(context.Background(), "name", "data"); err != nil {
 		t.Fatal(err)
 	}
 	prevMsgSerial := c.Connection.MsgSerial()
diff --git a/ably/realtime_presence_test.go b/ably/realtime_presence_test.go
index c7e9eb5..8977004 100644
--- a/ably/realtime_presence_test.go
+++ b/ably/realtime_presence_test.go
@@ -1,6 +1,7 @@
 package ably_test
 
 import (
+	"context"
 	"fmt"
 	"reflect"
 	"strconv"
@@ -9,10 +10,9 @@ import (
 
 	"github.com/ably/ably-go/ably"
 	"github.com/ably/ably-go/ably/ablytest"
-	"github.com/ably/ably-go/ably/proto"
 )
 
-func contains(members []*proto.PresenceMessage, clients ...string) error {
+func contains(members []*ably.PresenceMessage, clients ...string) error {
 	lookup := make(map[string]struct{}, len(members))
 	for _, member := range members {
 		lookup[member.ClientID] = struct{}{}
@@ -45,7 +45,10 @@ func TestRealtimePresence_Sync(t *testing.T) {
 	app, client := ablytest.NewRealtime(nil)
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
-	members, err := client.Channels.GetAndAttach("persisted:presence_fixtures").Presence.Get(true)
+	members, err := client.Channels.Get("persisted:presence_fixtures").Presence.Get(true)
+	if err != nil {
+		t.Fatal(err)
+	}
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -62,14 +65,20 @@ func TestRealtimePresence_Sync250(t *testing.T) {
 	client3 := app.NewRealtime(nil)
 	defer safeclose(t, ablytest.FullRealtimeCloser(client2), ablytest.FullRealtimeCloser(client3))
 
-	channel1 := client1.Channels.GetAndAttach("sync250")
-	channel2 := client2.Channels.GetAndAttach("sync250")
+	channel1 := client1.Channels.Get("sync250")
+	if err := channel1.Attach(context.Background()); err != nil {
+		t.Fatal(err)
+	}
+	channel2 := client2.Channels.Get("sync250")
+	if err := channel2.Attach(context.Background()); err != nil {
+		t.Fatal(err)
+	}
 
-	sub2, err := channel2.Presence.Subscribe()
+	sub2, unsub2, err := ablytest.ReceivePresenceMessages(channel2, nil)
 	if err != nil {
-		t.Fatalf("channel2.Presence.Subscribe()=%v", err)
+		t.Fatalf("channel2.ablytest.ReceiveMessages(Presence)=%v", err)
 	}
-	defer safeclose(t, sub2)
+	defer unsub2()
 
 	var rg ablytest.ResultGroup
 	var clients = generateClients(250)
@@ -79,12 +88,12 @@ func TestRealtimePresence_Sync250(t *testing.T) {
 	if err := rg.Wait(); err != nil {
 		t.Fatalf("rg.Wait()=%v", err)
 	}
-	members2 := make([]*proto.PresenceMessage, 250)
+	members2 := make([]*ably.PresenceMessage, 250)
 	tout := time.After(250 * ablytest.Timeout)
 
 	for i := range members2 {
 		select {
-		case msg := <-sub2.PresenceChannel():
+		case msg := <-sub2:
 			members2[i] = msg
 		case <-tout:
 			t.Fatalf("waiting for presence messages timed out after %v", 250*ablytest.Timeout)
@@ -94,7 +103,7 @@ func TestRealtimePresence_Sync250(t *testing.T) {
 	if err = contains(members2, clients...); err != nil {
 		t.Fatalf("members2: %v", err)
 	}
-	members3, err := client3.Channels.GetAndAttach("sync250").Presence.Get(true)
+	members3, err := client3.Channels.Get("sync250").Presence.Get(true)
 	if err != nil {
 		t.Fatal(err)
 	}
diff --git a/ably/rest_channel_test.go b/ably/rest_channel_test.go
index dbf1b86..b8c6c35 100644
--- a/ably/rest_channel_test.go
+++ b/ably/rest_channel_test.go
@@ -1,6 +1,7 @@
 package ably_test
 
 import (
+	"context"
 	"crypto/tls"
 	"encoding/base64"
 	"fmt"
@@ -21,7 +22,7 @@ import (
 	"github.com/ably/ably-go/ably/proto"
 )
 
-func TestRestChannel(t *testing.T) {
+func TestRESTChannel(t *testing.T) {
 	t.Parallel()
 	app, err := ablytest.NewSandbox(nil)
 	if err != nil {
@@ -70,7 +71,7 @@ func TestRestChannel(t *testing.T) {
 			}
 		}
 		for k, v := range m {
-			err := channel.Publish(k, v.data)
+			err := channel.Publish(context.Background(), k, v.data)
 			if err != nil {
 				ts.Fatal(err)
 			}
@@ -94,12 +95,12 @@ func TestRestChannel(t *testing.T) {
 	})
 
 	t.Run("History", func(ts *testing.T) {
-		historyRestChannel := client.Channels.Get("channelhistory")
+		historyRESTChannel := client.Channels.Get("channelhistory")
 		for i := 0; i < 2; i++ {
-			historyRestChannel.Publish("breakingnews", "Another Shark attack!!")
+			historyRESTChannel.Publish(context.Background(), "breakingnews", "Another Shark attack!!")
 		}
 
-		page1, err := historyRestChannel.History(&ably.PaginateParams{Limit: 1})
+		page1, err := historyRESTChannel.History(&ably.PaginateParams{Limit: 1})
 		if err != nil {
 			ts.Fatal(err)
 		}
@@ -122,17 +123,17 @@ func TestRestChannel(t *testing.T) {
 		}
 	})
 
-	t.Run("PublishAll", func(ts *testing.T) {
-		encodingRestChannel := client.Channels.Get("this?is#an?encoding#channel")
-		messages := []*proto.Message{
+	t.Run("PublishBatch", func(ts *testing.T) {
+		encodingRESTChannel := client.Channels.Get("this?is#an?encoding#channel")
+		messages := []*ably.Message{
 			{Name: "send", Data: "test data 1"},
 			{Name: "send", Data: "test data 2"},
 		}
-		err := encodingRestChannel.PublishAll(messages)
+		err := encodingRESTChannel.PublishBatch(context.Background(), messages)
 		if err != nil {
 			ts.Fatal(err)
 		}
-		page, err := encodingRestChannel.History(&ably.PaginateParams{Limit: 2})
+		page, err := encodingRESTChannel.History(&ably.PaginateParams{Limit: 2})
 		if err != nil {
 			ts.Fatal(err)
 		}
@@ -168,7 +169,7 @@ func TestRestChannel(t *testing.T) {
 			{"publish_1", "second message"},
 		}
 		for _, v := range sample {
-			err := channel.Publish(v.event, v.message)
+			err := channel.Publish(context.Background(), v.event, v.message)
 			if err != nil {
 				ts.Error(err)
 			}
@@ -217,7 +218,7 @@ func TestIdempotentPublishing(t *testing.T) {
 	t.Run("when ID is not included (#RSL1k2)", func(ts *testing.T) {
 		channel := client.Channels.Get("idempotent_test_1")
 		for range make([]struct{}, 3) {
-			err := channel.Publish("", randomStr)
+			err := channel.Publish(context.Background(), "", randomStr)
 			if err != nil {
 				ts.Fatal(err)
 			}
@@ -235,7 +236,7 @@ func TestIdempotentPublishing(t *testing.T) {
 	t.Run("when ID is included (#RSL1k2, #RSL1k5)", func(ts *testing.T) {
 		channel := client.Channels.Get("idempotent_test_2")
 		for range make([]struct{}, 3) {
-			err := channel.PublishAll([]*proto.Message{
+			err := channel.PublishBatch(context.Background(), []*ably.Message{
 				{
 					ID:   randomStr,
 					Data: randomStr,
@@ -262,7 +263,7 @@ func TestIdempotentPublishing(t *testing.T) {
 
 	t.Run("multiple messages in one publish operation (#RSL1k3)", func(ts *testing.T) {
 		channel := client.Channels.Get("idempotent_test_3")
-		err := channel.PublishAll([]*proto.Message{
+		err := channel.PublishBatch(context.Background(), []*ably.Message{
 			{
 				ID:   randomStr,
 				Data: randomStr,
@@ -287,13 +288,13 @@ func TestIdempotentPublishing(t *testing.T) {
 
 	t.Run("multiple messages in one publish operation with IDs following the required format described in RSL1k1 (#RSL1k3)", func(ts *testing.T) {
 		channel := client.Channels.Get("idempotent_test_4")
-		var m []*proto.Message
+		var m []*ably.Message
 		for i := 0; i < 3; i++ {
-			m = append(m, &proto.Message{
+			m = append(m, &ably.Message{
 				ID: fmt.Sprintf("%s:%d", randomStr, i),
 			})
 		}
-		err := channel.PublishAll(m)
+		err := channel.PublishBatch(context.Background(), m)
 		if err != nil {
 			ts.Fatal(err)
 		}
@@ -335,7 +336,7 @@ func TestIdempotentPublishing(t *testing.T) {
 
 	t.Run("the ID is populated with a random ID and serial 0 from this lib (#RSL1k1)", func(ts *testing.T) {
 		channel := client.Channels.Get("idempotent_test_5")
-		err := channel.Publish("event", "")
+		err := channel.Publish(context.Background(), "event", "")
 		if err != nil {
 			ts.Fatal(err)
 		}
@@ -369,7 +370,7 @@ func TestIdempotentPublishing(t *testing.T) {
 	t.Run("publishing a batch of messages", func(ts *testing.T) {
 		channel := client.Channels.Get("idempotent_test_6")
 		name := "event"
-		err := channel.PublishAll([]*proto.Message{
+		err := channel.PublishBatch(context.Background(), []*ably.Message{
 			{Name: name},
 			{Name: name},
 			{Name: name},
@@ -468,7 +469,7 @@ func TestIdempotent_retry(t *testing.T) {
 
 		ts.Run("two REST publish retries result in only one message being published'", func(ts *testing.T) {
 			channel := client.Channels.Get("idempotent_test_fallback")
-			err = channel.Publish("", randomStr)
+			err = channel.Publish(context.Background(), "", randomStr)
 			if err != nil {
 				ts.Error(err)
 			}
@@ -487,13 +488,13 @@ func TestIdempotent_retry(t *testing.T) {
 		ts.Run("or multiple messages in one publish operation'", func(ts *testing.T) {
 			retryCount = 0
 			channel := client.Channels.Get("idempotent_test_fallback_1")
-			msgs := []*proto.Message{
+			msgs := []*ably.Message{
 				{Data: randomStr},
 				{Data: randomStr},
 				{Data: randomStr},
 			}
 			for range msgs {
-				channel.PublishAll(msgs)
+				channel.PublishBatch(context.Background(), msgs)
 			}
 			res, err := channel.History(nil)
 			if err != nil {
@@ -523,15 +524,15 @@ func TestRSL1f1(t *testing.T) {
 	}
 	channel := client.Channels.Get("RSL1f")
 	id := "any_client_id"
-	var msgs []*proto.Message
+	var msgs []*ably.Message
 	size := 10
 	for i := 0; i < size; i++ {
-		msgs = append(msgs, &proto.Message{
+		msgs = append(msgs, &ably.Message{
 			ClientID: id,
 			Data:     fmt.Sprint(i),
 		})
 	}
-	err = channel.PublishAll(msgs)
+	err = channel.PublishBatch(context.Background(), msgs)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -568,7 +569,7 @@ func TestRSL1g(t *testing.T) {
 	}
 	t.Run("RSL1g1b", func(ts *testing.T) {
 		channel := client.Channels.Get("RSL1g1b")
-		err := channel.PublishAll([]*proto.Message{
+		err := channel.PublishBatch(context.Background(), []*ably.Message{
 			{Name: "some 1"},
 			{Name: "some 2"},
 			{Name: "some 3"},
@@ -592,7 +593,7 @@ func TestRSL1g(t *testing.T) {
 	})
 	t.Run("RSL1g2", func(ts *testing.T) {
 		channel := client.Channels.Get("RSL1g2")
-		err := channel.PublishAll([]*proto.Message{
+		err := channel.PublishBatch(context.Background(), []*ably.Message{
 			{Name: "1", ClientID: clientID},
 			{Name: "2", ClientID: clientID},
 			{Name: "3", ClientID: clientID},
@@ -616,7 +617,7 @@ func TestRSL1g(t *testing.T) {
 	})
 	t.Run("RSL1g3", func(ts *testing.T) {
 		channel := client.Channels.Get("RSL1g3")
-		err := channel.PublishAll([]*proto.Message{
+		err := channel.PublishBatch(context.Background(), []*ably.Message{
 			{Name: "1", ClientID: clientID},
 			{Name: "2", ClientID: "other client"},
 			{Name: "3", ClientID: clientID},
diff --git a/ably/rest_client_test.go b/ably/rest_client_test.go
index d72c396..48f2b89 100644
--- a/ably/rest_client_test.go
+++ b/ably/rest_client_test.go
@@ -1,6 +1,7 @@
 package ably_test
 
 import (
+	"context"
 	"crypto/tls"
 	"encoding/json"
 	"errors"
@@ -61,7 +62,7 @@ func TestRestClient(t *testing.T) {
 			if err != nil {
 				ts.Fatal(err)
 			}
-			err = client.Channels.Get("test").Publish("ping", "pong")
+			err = client.Channels.Get("test").Publish(context.Background(), "ping", "pong")
 			if err != nil {
 				ts.Fatal(err)
 			}
@@ -95,7 +96,7 @@ func TestRestClient(t *testing.T) {
 			if err != nil {
 				ts.Fatal(err)
 			}
-			err = client.Channels.Get("test").Publish("ping", "pong")
+			err = client.Channels.Get("test").Publish(context.Background(), "ping", "pong")
 			if err != nil {
 				ts.Fatal(err)
 			}
@@ -291,7 +292,7 @@ func TestRest_hostfallback(t *testing.T) {
 		if err != nil {
 			ts.Fatal(err)
 		}
-		err = client.Channels.Get("test").Publish("ping", "pong")
+		err = client.Channels.Get("test").Publish(context.Background(), "ping", "pong")
 		if err == nil {
 			ts.Error("expected an error")
 		}
@@ -456,7 +457,7 @@ func TestRest_rememberHostFallback(t *testing.T) {
 			ts.Fatal(err)
 		}
 		channel := client.Channels.Get("remember_fallback_host")
-		err = channel.Publish("ping", "pong")
+		err = channel.Publish(context.Background(), "ping", "pong")
 		if err != nil {
 			ts.Fatal(err)
 		}
@@ -467,7 +468,7 @@ func TestRest_rememberHostFallback(t *testing.T) {
 		retryCount = 0
 
 		// the same cached host is used again
-		err = channel.Publish("pong", "ping")
+		err = channel.Publish(context.Background(), "pong", "ping")
 		if err != nil {
 			ts.Fatal(err)
 		}
@@ -480,7 +481,7 @@ func TestRest_rememberHostFallback(t *testing.T) {
 		}
 	})
 }
-func TestRestChannels_RSN1(t *testing.T) {
+func TestRESTChannels_RSN1(t *testing.T) {
 	app, err := ablytest.NewSandbox(nil)
 	if err != nil {
 		t.Fatal(err)
@@ -524,7 +525,7 @@ func TestRestChannels_RSN1(t *testing.T) {
 		for _, v := range sample {
 			client.Channels.Get(v.name)
 		}
-		client.Channels.Range(func(name string, _ *ably.RestChannel) bool {
+		client.Channels.Range(func(name string, _ *ably.RESTChannel) bool {
 			n := client.Channels.Get(name + "_range")
 			return client.Channels.Exists(n.Name)
 		})
@@ -551,7 +552,7 @@ func TestFixConnLeak_ISSUE89(t *testing.T) {
 	}
 	channel := client.Channels.Get("issue89")
 	for i := 0; i < 10; i++ {
-		err := channel.Publish(fmt.Sprintf("msg_%d", i), fmt.Sprint(i))
+		err := channel.Publish(context.Background(), fmt.Sprintf("msg_%d", i), fmt.Sprint(i))
 		if err != nil {
 			t.Error(err)
 		}
diff --git a/ably/state_test.go b/ably/state_test.go
index 22b9afd..59d6b1d 100644
--- a/ably/state_test.go
+++ b/ably/state_test.go
@@ -6,6 +6,7 @@ import (
 )
 
 var errNotEmitted = errors.New("not emitted")
+var errNack = NewErrorInfo(123123, errors.New("nack"))
 
 func chans(n int) []chan error {
 	ch := make([]chan error, n)
@@ -28,9 +29,9 @@ func receive(ch ...chan error) []error {
 	return errs
 }
 
-func emit(serial int64, count int, fn func(*pendingEmitter, int64, int, error)) func(*pendingEmitter) {
+func emit(serial int64, count int, fn func(*pendingEmitter, int64, int, *ErrorInfo), err *ErrorInfo) func(*pendingEmitter) {
 	return func(q *pendingEmitter) {
-		fn(q, serial, count, nil)
+		fn(q, serial, count, err)
 	}
 }
 
@@ -75,34 +76,34 @@ func TestQueuedEmitter(t *testing.T) {
 	}{{ // 5 pending messages, ack first two
 		serial: []int64{1, 2, 3, 4, 5},
 		ack:    []int64{1, 2},
-		emit:   emit(1, 2, (*pendingEmitter).Ack),
+		emit:   emit(1, 2, (*pendingEmitter).Ack, nil),
 	}, { // 5 pending messages, ack for second and third, first should got nack
 		serial: []int64{1, 2, 3, 4, 5},
 		ack:    []int64{2, 3},
 		nack:   []int64{1},
-		emit:   emit(2, 2, (*pendingEmitter).Ack),
+		emit:   emit(2, 2, (*pendingEmitter).Ack, nil),
 	}, { // 5 pending messages, 1 and 2 get nack, rest ack
 		serial: []int64{5, 2, 4, 1, 3},
 		ack:    []int64{3, 4, 5},
 		nack:   []int64{1, 2},
-		emit:   emit(3, 3, (*pendingEmitter).Ack),
+		emit:   emit(3, 3, (*pendingEmitter).Ack, nil),
 	}, { // 10 pending messages, first 5 get nacks, next 4 ack
 		serial: []int64{9, 4, 2, 6, 10, 3, 1, 7, 5, 8},
 		ack:    []int64{6, 7, 8, 9},
 		nack:   []int64{1, 2, 3, 4, 5},
-		emit:   emit(6, 4, (*pendingEmitter).Ack),
+		emit:   emit(6, 4, (*pendingEmitter).Ack, nil),
 	}, { // 5 pending messages, nack first 3
 		serial: []int64{3, 5, 1, 4, 2},
 		nack:   []int64{1, 2, 3},
-		emit:   emit(1, 3, (*pendingEmitter).Nack),
+		emit:   emit(1, 3, (*pendingEmitter).Nack, errNack),
 	}, { // 5 pending messages, nack all
 		serial: []int64{2, 1, 3, 5, 4},
 		nack:   []int64{1, 2, 3, 4, 5},
-		emit:   emit(1, 5, (*pendingEmitter).Nack),
+		emit:   emit(1, 5, (*pendingEmitter).Nack, errNack),
 	}, { // 5 pending messages, nack all
 		serial: []int64{4, 5, 2, 3, 1},
 		nack:   []int64{1, 2, 3, 4, 5},
-		emit:   emit(2, 10, (*pendingEmitter).Nack),
+		emit:   emit(2, 10, (*pendingEmitter).Nack, errNack),
 	}}
 	for _, cas := range cases {
 		testQueuedEmitter(t, cas.serial, cas.ack, cas.nack, cas.emit)