Repository: ably/ably-go
Pull request #190 (2020-08-24 09:23:12)

RTL2: EventEmitter for channel events; remove old State
[RTL2](https://docs.ably.io/client-lib-development-guide/features/#RTL2). This also removes the old State machinery, including the temporary workarounds for bridging the two.


Patch:
diff --git a/ably/options.go b/ably/options.go
index 5ce7ccc..a7ba6ee 100644
--- a/ably/options.go
+++ b/ably/options.go
@@ -212,11 +212,6 @@ type clientOptions struct {
 	// If Dial is nil, the default websocket connection is used.
 	Dial func(protocol string, u *url.URL) (proto.Conn, error)
 
-	// Listener if set, will be automatically registered with On method for every
-	// realtime connection and realtime channel created by realtime client.
-	// The listener will receive events for all state transitions.
-	Listener chan<- State
-
 	// HTTPClient specifies the client used for HTTP communication by RestClient.
 	//
 	// If HTTPClient is nil, the http.DefaultClient is used.
diff --git a/ably/realtime_channel.go b/ably/realtime_channel.go
index de362d5..79f179d 100644
--- a/ably/realtime_channel.go
+++ b/ably/realtime_channel.go
@@ -110,61 +110,67 @@ func (ch *Channels) All() []*RealtimeChannel {
 func (ch *Channels) Release(name string) error {
 	ch.mtx.Lock()
 	defer ch.mtx.Unlock()
-	if c, ok := ch.chans[name]; ok {
-		delete(ch.chans, name)
-		return c.Close()
+	c, ok := ch.chans[name]
+	if !ok {
+		return nil
 	}
+	err := wait(c.Detach())
+	if err != nil {
+		return err
+	}
+	delete(ch.chans, name)
 	return nil
 }
 
-func (ch *Channels) broadcastConnStateChange(state State) {
+func (ch *Channels) broadcastConnStateChange(change ConnectionStateChange) {
 	ch.mtx.Lock()
 	defer ch.mtx.Unlock()
 	for _, c := range ch.chans {
-		c.onConnState(state)
+		c.onConnStateChange(change)
 	}
 }
 
 // RealtimeChannel represents a single named message channel.
 type RealtimeChannel struct {
+	mtx sync.Mutex
+
+	ChannelEventEmitter
 	Name     string            // name used to create the channel
 	Presence *RealtimePresence //
 
+	state           ChannelState
+	errorReason     *ErrorInfo
+	internalEmitter ChannelEventEmitter
+
 	client *Realtime
-	state  *stateEmitter
 	subs   *subscriptions
 	queue  *msgQueue
-	listen chan State
 }
 
 func newRealtimeChannel(name string, client *Realtime) *RealtimeChannel {
 	c := &RealtimeChannel{
-		Name:   name,
+		ChannelEventEmitter: ChannelEventEmitter{newEventEmitter(client.logger())},
+		Name:                name,
+
+		state:           ChannelStateInitialized,
+		internalEmitter: ChannelEventEmitter{newEventEmitter(client.logger())},
+
 		client: client,
-		state:  newStateEmitter(StateChan, StateChanInitialized, name, client.logger()),
 		subs:   newSubscriptions(subscriptionMessages, client.logger()),
-		listen: make(chan State, 1),
 	}
 	c.Presence = newRealtimePresence(c)
 	c.queue = newMsgQueue(client.Connection)
-	if c.opts().Listener != nil {
-		c.On(c.opts().Listener)
-	}
 	return c
 }
 
-func (c *RealtimeChannel) onConnState(state State) {
-	c.state.Lock()
+func (c *RealtimeChannel) onConnStateChange(change ConnectionStateChange) {
+	c.mtx.Lock()
 	active := c.isActive()
-	c.state.Unlock()
-	switch state.State {
-	case StateConnFailed:
+	c.mtx.Unlock()
+	switch change.Current {
+	case ConnectionStateFailed:
 		if active {
-			c.state.syncSet(StateChanFailed, state.Err)
-		}
-	case StateConnClosed:
-		if active {
-			c.state.syncSet(StateChanClosed, state.Err)
+			c.setState(ChannelStateFailed, change.Reason)
 		}
 	}
 }
@@ -180,20 +186,13 @@ func (c *RealtimeChannel) Attach() (Result, error) {
 	return c.attach(true)
 }
 
-var attachResultStates = []StateEnum{
-	StateChanAttached, // expected state
-	StateChanClosing,
-	StateChanClosed,
-	StateChanFailed,
-}
-
 func (c *RealtimeChannel) attach(result bool) (Result, error) {
 	return c.mayAttach(result, true)
 }
 
 func (c *RealtimeChannel) mayAttach(result, checkActive bool) (Result, error) {
-	c.state.Lock()
-	defer c.state.Unlock()
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
 	if checkActive {
 		if c.isActive() {
 			return nopResult, nil
@@ -204,20 +203,20 @@ func (c *RealtimeChannel) mayAttach(result, checkActive bool) (Result, error) {
 
 func (c *RealtimeChannel) lockAttach(result bool, err error) (Result, error) {
 	if !c.client.Connection.lockIsActive() {
-		return nil, c.state.set(StateChanFailed, errAttach)
+		return nil, c.lockSetState(ChannelStateFailed, errAttach)
 	}
-	c.state.set(StateChanAttaching, err)
+	c.lockSetState(ChannelStateAttaching, err)
 	var res Result
 	if result {
-		res = c.state.listenResult(attachResultStates...)
+		res = c.internalEmitter.listenResult(ChannelStateAttached, ChannelStateFailed)
 	}
 	msg := &proto.ProtocolMessage{
 		Action:  proto.ActionAttach,
-		Channel: c.state.channel,
+		Channel: c.Name,
 	}
 	err = c.client.Connection.send(msg, nil)
 	if err != nil {
-		return nil, c.state.set(StateChanFailed, err)
+		return nil, c.lockSetState(ChannelStateFailed, err)
 	}
 	return res, nil
 }
@@ -233,55 +232,34 @@ func (c *RealtimeChannel) Detach() (Result, error) {
 	return c.detach(true)
 }
 
-var detachResultStates = []StateEnum{
-	StateChanDetached, // expected state
-	StateChanClosing,
-	StateChanClosed,
-	StateChanFailed,
-}
-
 func (c *RealtimeChannel) detach(result bool) (Result, error) {
-	c.state.Lock()
-	defer c.state.Unlock()
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
 	switch {
-	case c.state.current == StateChanFailed:
-		return nil, stateError(StateChanFailed, errDetach)
+	case c.state == ChannelStateFailed:
+		return nil, channelStateError(ChannelStateFailed, errDetach)
 	case !c.isActive():
 		return nopResult, nil
 	}
 	if !c.client.Connection.lockIsActive() {
-		return nil, c.state.set(StateChanFailed, errDetach)
+		return nil, c.lockSetState(ChannelStateFailed, errDetach)
 	}
-	c.state.set(StateChanDetaching, nil)
+	c.lockSetState(ChannelStateDetaching, nil)
 	var res Result
 	if result {
-		res = c.state.listenResult(detachResultStates...)
+		res = c.internalEmitter.listenResult(ChannelStateDetached, ChannelStateFailed)
 	}
 	msg := &proto.ProtocolMessage{
 		Action:  proto.ActionDetach,
-		Channel: c.state.channel,
+		Channel: c.Name,
 	}
 	err := c.client.Connection.send(msg, nil)
 	if err != nil {
-		return nil, c.state.set(StateChanFailed, err)
+		return nil, c.lockSetState(ChannelStateFailed, err)
 	}
 	return res, nil
 }
 
-// Closes initiates closing sequence for the channel; it waits until the
-// operation is complete.
-//
-// If connection is already closed, this method is a nop.
-// If sending close message succeeds, it closes and unsubscribes all channels.
-func (c *RealtimeChannel) Close() error {
-	err := wait(c.Detach())
-	c.subs.close()
-	if err != nil {
-		return c.state.syncSet(StateChanClosed, err)
-	}
-	return nil
-}
-
 // Subscribe subscribes to a realtime channel, which makes any newly received
 // messages relayed to the returned Subscription value.
 //
@@ -308,23 +286,64 @@ func (c *RealtimeChannel) Unsubscribe(sub *Subscription, names ...string) {
 	c.subs.unsubscribe(true, sub, namesToKeys(names)...)
 }
 
-// On relays request channel states to c; on state transition
-// connection will not block sending to c - the caller must ensure the incoming
-// values are read at proper pace or the c is sufficiently buffered.
+type channelStateChanges chan ChannelStateChange
+
+func (c channelStateChanges) Receive(change ChannelStateChange) {
+	c <- change
+}
+
+type ChannelEventEmitter struct {
+	emitter *eventEmitter
+}
+
+// On registers an event handler for connection events of a specific kind.
+//
+// See package-level documentation on Event Emitter for details.
+func (em ChannelEventEmitter) On(e ChannelEvent, handle func(ChannelStateChange)) (off func()) {
+	return em.emitter.On(e, func(change emitterData) {
+		handle(change.(ChannelStateChange))
+	})
+}
+
+// OnAll registers an event handler for all connection events.
+//
+// See package-level documentation on Event Emitter for details.
+func (em ChannelEventEmitter) OnAll(handle func(ChannelStateChange)) (off func()) {
+	return em.emitter.OnAll(func(change emitterData) {
+		handle(change.(ChannelStateChange))
+	})
+}
+
+// Once registers an one-off event handler for connection events of a specific kind.
 //
-// If no states are given, c is registered for all of them.
-// If c is nil, the method panics.
-// If c is already registered, its state set is expanded.
-func (c *RealtimeChannel) On(ch chan<- State, states ...StateEnum) {
-	c.state.on(ch, states...)
+// See package-level documentation on Event Emitter for details.
+func (em ChannelEventEmitter) Once(e ChannelEvent, handle func(ChannelStateChange)) (off func()) {
+	return em.emitter.Once(e, func(change emitterData) {
+		handle(change.(ChannelStateChange))
+	})
+}
+
+// OnceAll registers an one-off event handler for all connection events.
+//
+// See package-level documentation on Event Emitter for details.
+func (em ChannelEventEmitter) OnceAll(handle func(ChannelStateChange)) (off func()) {
+	return em.emitter.OnceAll(func(change emitterData) {
+		handle(change.(ChannelStateChange))
+	})
 }
 
-// Off removes c from listening on the given channel state transitions.
+// Off deregisters event handlers for connection events of a specific kind.
 //
-// If no states are given, c is removed for all of the connection's states.
-// If c is nil, the method panics.
-func (c *RealtimeChannel) Off(ch chan<- State, states ...StateEnum) {
-	c.state.off(ch, states...)
+// See package-level documentation on Event Emitter for details.
+func (em ChannelEventEmitter) Off(e ChannelEvent) {
+	em.emitter.Off(e)
+}
+
+// Off deregisters all event handlers.
+//
+// See package-level documentation on Event Emitter for details.
+func (em ChannelEventEmitter) OffAll() {
+	em.emitter.OffAll()
 }
 
 // Publish publishes a message on the channel, which is send on separate
@@ -349,7 +368,7 @@ func (c *RealtimeChannel) PublishAll(messages []*proto.Message) (Result, error)
 	}
 	msg := &proto.ProtocolMessage{
 		Action:   proto.ActionMessage,
-		Channel:  c.state.channel,
+		Channel:  c.Name,
 		Messages: messages,
 	}
 	return c.send(msg)
@@ -368,10 +387,10 @@ func (c *RealtimeChannel) send(msg *proto.ProtocolMessage) (Result, error) {
 	}
 	res, listen := newErrResult()
 	switch c.State() {
-	case StateChanInitialized, StateChanAttaching:
+	case ChannelStateInitialized, ChannelStateAttaching:
 		c.queue.Enqueue(msg, listen)
 		return res, nil
-	case StateChanAttached:
+	case ChannelStateAttached:
 	default:
 		return nil, &ErrorInfo{Code: 90001}
 	}
@@ -381,43 +400,43 @@ func (c *RealtimeChannel) send(msg *proto.ProtocolMessage) (Result, error) {
 	return res, nil
 }
 
-// State gives current state of the channel.
-func (c *RealtimeChannel) State() StateEnum {
-	c.state.Lock()
-	defer c.state.Unlock()
-	return c.state.current
+// State gives the current state of the channel.
+func (c *RealtimeChannel) State() ChannelState {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	return c.state
 }
 
 // Reason gives the last error that caused channel transition to failed state.
 func (c *RealtimeChannel) Reason() error {
-	c.state.Lock()
-	defer c.state.Unlock()
-	return c.state.err
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	return c.errorReason
 }
 
 func (c *RealtimeChannel) notify(msg *proto.ProtocolMessage) {
 	switch msg.Action {
 	case proto.ActionAttached:
 		c.Presence.onAttach(msg)
-		c.state.syncSet(StateChanAttached, nil)
+		c.setState(ChannelStateAttached, nil)
 		c.queue.Flush()
 	case proto.ActionDetached:
-		c.state.Lock()
+		c.mtx.Lock()
 
 		err := error(newErrorProto(msg.Error))
-		switch c.state.current {
-		case StateChanDetaching:
-			c.state.set(StateChanDetached, err)
-			c.state.Unlock()
+		switch c.state {
+		case ChannelStateDetaching:
+			c.lockSetState(ChannelStateDetached, err)
+			c.mtx.Unlock()
 			return
-		case StateChanAttached: // TODO: Also SUSPENDED; RTL13a
+		case ChannelStateAttached: // TODO: Also SUSPENDED; RTL13a
 			var res Result
 			res, err = c.lockAttach(true, err)
 			if err != nil {
 				break
 			}
 
-			c.state.Unlock()
+			c.mtx.Unlock()
 			go func() {
 				// We need to wait in another goroutine to allow more messages
 				// to reach the connection.
@@ -426,16 +445,16 @@ func (c *RealtimeChannel) notify(msg *proto.ProtocolMessage) {
 				if err == nil {
 					return
 				}
-				c.state.Lock()
+				c.mtx.Lock()
 
 				c.lockStartRetryAttachLoop(err)
 			}()
 			return
 		case
-			StateChanAttaching,
-			StateChanDetached: // TODO: Should be SUSPENDED
+			ChannelStateAttaching,
+			ChannelStateDetached: // TODO: Should be SUSPENDED
 		default:
-			c.state.Unlock()
+			c.mtx.Unlock()
 			return
 		}
 
@@ -445,7 +464,7 @@ func (c *RealtimeChannel) notify(msg *proto.ProtocolMessage) {
 	case proto.ActionPresence:
 		c.Presence.processIncomingMessage(msg, "")
 	case proto.ActionError:
-		c.state.syncSet(StateChanFailed, newErrorProto(msg.Error))
+		c.setState(ChannelStateFailed, newErrorProto(msg.Error))
 		c.queue.Fail(newErrorProto(msg.Error))
 	case proto.ActionMessage:
 		c.subs.messageEnqueue(msg)
@@ -455,22 +474,21 @@ func (c *RealtimeChannel) notify(msg *proto.ProtocolMessage) {
 
 func (c *RealtimeChannel) lockStartRetryAttachLoop(err error) {
 	// TODO: Move to SUSPENDED; move it to DETACHED for now.
-	c.state.set(StateChanDetached, err)
-	c.state.Unlock()
+	c.lockSetState(ChannelStateDetached, err)
 
-	go func() {
-		// TODO: The listener should be set before unlocking the state.
-		// Will do once we have an EventEmitter.
-		stateChange := make(chan State, 10)
-		c.state.on(stateChange)
-		defer c.state.off(stateChange)
+	stateChange := make(channelStateChanges, 1)
+	off := c.internalEmitter.OnceAll(stateChange.Receive)
 
+	c.mtx.Unlock()
+
+	go func() {
+		defer off()
 		for !c.retryAttach(stateChange) {
 		}
 	}()
 }
 
-func (c *RealtimeChannel) retryAttach(stateChange chan State) (done bool) {
+func (c *RealtimeChannel) retryAttach(stateChange channelStateChanges) (done bool) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
@@ -491,12 +509,12 @@ func (c *RealtimeChannel) retryAttach(stateChange chan State) (done bool) {
 		return true
 	}
 	// TODO: Move to SUSPENDED; move it to DETACHED for now.
-	c.state.syncSet(StateChanDetached, err)
+	c.setState(ChannelStateDetached, err)
 	return false
 }
 
 func (c *RealtimeChannel) isActive() bool {
-	return c.state.current == StateChanAttaching || c.state.current == StateChanAttached
+	return c.state == ChannelStateAttaching || c.state == ChannelStateAttached
 }
 
 func (c *RealtimeChannel) opts() *clientOptions {
@@ -506,3 +524,32 @@ func (c *RealtimeChannel) opts() *clientOptions {
 func (c *RealtimeChannel) logger() *LoggerOptions {
 	return c.client.logger()
 }
+
+func (c *RealtimeChannel) setState(state ChannelState, err error) error {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	return c.lockSetState(state, err)
+}
+
+func (c *RealtimeChannel) lockSetState(state ChannelState, err error) error {
+	previous := c.state
+	changed := c.state != state
+	c.state = state
+	c.errorReason = channelStateError(state, err)
+	change := ChannelStateChange{
+		Current:  c.state,
+		Previous: previous,
+		Reason:   c.errorReason,
+	}
+	if !changed {
+		change.Event = ChannelEventUpdated
+	} else {
+		change.Event = ChannelEvent(change.Current)
+	}
+	c.internalEmitter.emitter.Emit(change.Event, change)
+	c.emitter.Emit(change.Event, change)
+	if c.errorReason == nil {
+		return nil
+	}
+	return c.errorReason
+}
diff --git a/ably/realtime_client.go b/ably/realtime_client.go
index 04d5fbd..f30cdc2 100644
--- a/ably/realtime_client.go
+++ b/ably/realtime_client.go
@@ -32,7 +32,9 @@ func NewRealtime(options ClientOptions) (*Realtime, error) {
 		c.onChannelMsg,
 		c.onReconnected,
 		c.onReconnectionFailed,
-		c.onConnStateChange,
+	})
+	conn.internalEmitter.OnAll(func(change ConnectionStateChange) {
+		c.Channels.broadcastConnStateChange(change)
 	})
 	if err != nil {
 		return nil, err
@@ -75,7 +77,7 @@ func (c *Realtime) onReconnected(err *proto.ErrorInfo, isNewID bool) {
 	for _, ch := range c.Channels.All() {
 		switch ch.State() {
 		// TODO: SUSPENDED
-		case StateChanAttaching, StateChanAttached:
+		case ChannelStateAttaching, ChannelStateAttached:
 			ch.mayAttach(false, false)
 		}
 	}
@@ -83,7 +85,7 @@ func (c *Realtime) onReconnected(err *proto.ErrorInfo, isNewID bool) {
 
 func (c *Realtime) onReconnectionFailed(err *proto.ErrorInfo) {
 	for _, ch := range c.Channels.All() {
-		ch.state.syncSet(StateChanFailed, newErrorProto(err))
+		ch.setState(ChannelStateFailed, newErrorProto(err))
 	}
 }
 
@@ -91,11 +93,6 @@ func isTokenError(err *proto.ErrorInfo) bool {
 	return err.StatusCode == http.StatusUnauthorized && (40140 <= err.Code && err.Code < 40150)
 }
 
-func (c *Realtime) onConnStateChange(state State) {
-	// TODO: Replace with EventEmitter https://github.com/ably/ably-go/pull/144
-	c.Channels.broadcastConnStateChange(state)
-}
-
 func (c *Realtime) opts() *clientOptions {
 	return c.rest.opts
 }
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index db4800a..5b563e6 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -6,6 +6,7 @@ import (
 	"net/url"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/ably/ably-go/ably/internal/ablyutil"
@@ -31,6 +32,14 @@ const (
 // Connection represents a single connection Realtime instantiates for
 // communication with Ably servers.
 type Connection struct {
+	mtx sync.Mutex
+
+	ConnectionEventEmitter
+
+	state           ConnectionState
+	errorReason     *ErrorInfo
+	internalEmitter ConnectionEventEmitter
+
 	id        string
 	key       string
 	serial    int64
@@ -38,7 +47,6 @@ type Connection struct {
 	err       error
 	conn      proto.Conn
 	opts      *clientOptions
-	state     *stateEmitter
 	pending   pendingEmitter
 	queue     *msgQueue
 	auth      *Auth
@@ -63,21 +71,21 @@ type connCallbacks struct {
 	// onReconnectionFailed is called when we get a FAILED response from a
 	// reconnection request.
 	onReconnectionFailed func(*proto.ErrorInfo)
-	onStateChange        func(State)
 }
 
 func newConn(opts *clientOptions, auth *Auth, callbacks connCallbacks) (*Connection, error) {
 	c := &Connection{
+
+		ConnectionEventEmitter: ConnectionEventEmitter{newEventEmitter(auth.logger())},
+		state:                  ConnectionStateInitialized,
+		internalEmitter:        ConnectionEventEmitter{newEventEmitter(auth.logger())},
+
 		opts:      opts,
-		state:     newStateEmitter(StateConn, StateConnInitialized, "", auth.logger()),
 		pending:   newPendingEmitter(auth.logger()),
 		auth:      auth,
 		callbacks: callbacks,
 	}
 	c.queue = newMsgQueue(c)
-	if opts.Listener != nil {
-		c.onState(opts.Listener)
-	}
 	if !opts.NoConnect {
 		if _, err := c.connect(false); err != nil {
 			return nil, err
@@ -100,9 +108,9 @@ func (c *Connection) dial(proto string, u *url.URL) (proto.Conn, error) {
 // Connect attempts to move the connection to the CONNECTED state, if it
 // can and if it isn't already.
 func (c *Connection) Connect() {
-	c.state.Lock()
+	c.mtx.Lock()
 	isActive := c.isActive()
-	c.state.Unlock()
+	c.mtx.Unlock()
 	if isActive {
 		return
 	}
@@ -115,21 +123,15 @@ func (c *Connection) Close() {
 	c.close()
 }
 
-var connectResultStates = []StateEnum{
-	StateConnConnected, // expected state
-	StateConnFailed,
-	StateConnDisconnected,
-}
-
 func (c *Connection) connect(result bool) (Result, error) {
-	c.state.Lock()
+	c.mtx.Lock()
 	mode := c.getMode()
-	c.state.Unlock()
+	c.mtx.Unlock()
 	return c.connectWith(result, mode)
 }
 
 func (c *Connection) reconnect(lastActivityAt time.Time, connDetails *proto.ConnectionDetails, result bool) (Result, error) {
-	c.state.Lock()
+	c.mtx.Lock()
 
 	var mode connectionMode
 	if connDetails != nil && c.opts.Now().Sub(lastActivityAt) >= time.Duration(connDetails.ConnectionStateTTL+connDetails.MaxIdleInterval) {
@@ -143,7 +145,7 @@ func (c *Connection) reconnect(lastActivityAt time.Time, connDetails *proto.Conn
 		mode = c.getMode()
 	}
 
-	c.state.Unlock()
+	c.mtx.Unlock()
 
 	r, err := c.connectWith(result, mode)
 	if err != nil {
@@ -152,9 +154,9 @@ func (c *Connection) reconnect(lastActivityAt time.Time, connDetails *proto.Conn
 	// We have successfully dialed reconnection request. We need to set this so
 	// when the next message arrives it will be treated as the response to
 	// reconnection request.
-	c.state.Lock()
+	c.mtx.Lock()
 	c.reconnecting = true
-	c.state.Unlock()
+	c.mtx.Unlock()
 
 	return r, nil
 }
@@ -207,28 +209,32 @@ func (c *Connection) params(mode connectionMode) (url.Values, error) {
 }
 
 func (c *Connection) connectWith(result bool, mode connectionMode) (Result, error) {
-	c.state.Lock()
-	defer c.state.Unlock()
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
 	if !c.isActive() {
-		c.setState(StateConnConnecting, nil)
+		c.lockSetState(ConnectionStateConnecting, nil)
 	}
 	u, err := url.Parse(c.opts.realtimeURL())
 	if err != nil {
-		return nil, c.setState(StateConnFailed, err)
+		return nil, c.lockSetState(ConnectionStateFailed, err)
 	}
 	var res Result
 	if result {
-		res = c.state.listenResult(connectResultStates...)
+		res = c.internalEmitter.listenResult(
+			ConnectionStateConnected, // expected state
+			ConnectionStateFailed,
+			ConnectionStateDisconnected,
+		)
 	}
 	query, err := c.params(mode)
 	if err != nil {
-		return nil, c.setState(StateConnFailed, err)
+		return nil, c.lockSetState(ConnectionStateFailed, err)
 	}
 	u.RawQuery = query.Encode()
 	proto := c.opts.protocol()
 	conn, err := c.dial(proto, u)
 	if err != nil {
-		return nil, c.setState(StateConnFailed, err)
+		return nil, c.lockSetState(ConnectionStateFailed, err)
 	}
 	if c.logger().Is(LogVerbose) {
 		c.setConn(verboseConn{conn: conn, logger: c.logger()})
@@ -239,28 +245,22 @@ func (c *Connection) connectWith(result bool, mode connectionMode) (Result, erro
 	return res, nil
 }
 
-var closeResultStates = []StateEnum{
-	StateConnClosed, // expected state
-	StateConnFailed,
-	StateConnDisconnected,
-}
-
 func (c *Connection) close() {
-	c.state.Lock()
+	c.mtx.Lock()
 	c.key, c.id = "", "" //(RTN16c)
-	defer c.state.Unlock()
-	switch c.state.current {
+	defer c.mtx.Unlock()
+	switch c.state {
 	case
-		StateConnClosing,
-		StateConnClosed,
-		StateConnInitialized,
-		StateConnFailed,
-		StateConnDisconnected:
+		ConnectionStateClosing,
+		ConnectionStateClosed,
+		ConnectionStateInitialized,
+		ConnectionStateFailed,
+		ConnectionStateDisconnected:
 
 		return
 	}
 
-	c.state.set(StateConnClosing, nil)
+	c.lockSetState(ConnectionStateClosing, nil)
 	msg := &proto.ProtocolMessage{Action: proto.ActionClose}
 	c.updateSerial(msg, nil)
 
@@ -275,10 +275,10 @@ func (c *Connection) close() {
 
 // ID gives unique ID string obtained from Ably upon successful connection.
 // The ID may change due to reconnection and recovery; on every received
-// StateConnConnected event previously obtained ID is no longer valid.
+// ConnectionStateConnected event previously obtained ID is no longer valid.
 func (c *Connection) ID() string {
-	c.state.Lock()
-	defer c.state.Unlock()
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
 	return c.id
 }
 
@@ -286,8 +286,8 @@ func (c *Connection) ID() string {
 // The key may change due to reconnection and recovery; on every received
 // StatConnConnected event previously obtained Key is no longer valid.
 func (c *Connection) Key() string {
-	c.state.Lock()
-	defer c.state.Unlock()
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
 	return c.key
 }
 
@@ -295,22 +295,22 @@ func (c *Connection) Key() string {
 // for ping request and pong response.
 //
 // Ping returns non-nil error without any attempt of communication with Ably
-// if the connection state is StateConnClosed or StateConnFailed.
+// if the connection state is ConnectionStateClosed or ConnectionStateFailed.
 func (c *Connection) Ping() (ping, pong time.Duration, err error) {
 	return 0, 0, errors.New("TODO")
 }
 
 // ErrorReason gives last known error that caused connection transit to
-// StateConnFailed state.
+// ConnectionStateFailed state.
 func (c *Connection) ErrorReason() *ErrorInfo {
-	c.state.Lock()
-	defer c.state.Unlock()
-	return c.state.err
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	return c.errorReason
 }
 
 func (c *Connection) RecoveryKey() string {
-	c.state.Lock()
-	defer c.state.Unlock()
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
 	if c.key == "" {
 		return ""
 	}
@@ -320,34 +320,33 @@ func (c *Connection) RecoveryKey() string {
 // Serial gives serial number of a message received most recently. Last known
 // serial number is used when recovering connection state.
 func (c *Connection) Serial() int64 {
-	c.state.Lock()
-	defer c.state.Unlock()
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
 	return c.serial
 }
 
 // State returns current state of the connection.
 func (c *Connection) State() ConnectionState {
-	c.state.Lock()
-	defer c.state.Unlock()
-	return mapOldToNewConnState(c.state.current)
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	return c.state
 }
 
-// onState relays request connection states to the given channel; onState state transition
-// connection will not block sending to c - the caller must ensure the incoming
-// values are read at proper pace or the c is sufficiently buffered.
-//
-// If no states are given, c is registered for all of them.
-// If c is nil, the method panics.
-// If c is already registered, its state set is expanded.
-func (c *Connection) onState(ch chan<- State, states ...StateEnum) {
-	c.state.on(ch, states...)
+type connStateChanges chan ConnectionStateChange
+
+func (c connStateChanges) Receive(change ConnectionStateChange) {
+	c <- change
+}
+
+type ConnectionEventEmitter struct {
+	emitter *eventEmitter
 }
 
 // On registers an event handler for connection events of a specific kind.
 //
 // See package-level documentation on Event Emitter for details.
-func (c *Connection) On(e ConnectionEvent, handle func(ConnectionStateChange)) (off func()) {
-	return c.state.eventEmitter.On(e, func(change emitterData) {
+func (em ConnectionEventEmitter) On(e ConnectionEvent, handle func(ConnectionStateChange)) (off func()) {
+	return em.emitter.On(e, func(change emitterData) {
 		handle(change.(ConnectionStateChange))
 	})
 }
@@ -355,8 +354,8 @@ func (c *Connection) On(e ConnectionEvent, handle func(ConnectionStateChange)) (
 // OnAll registers an event handler for all connection events.
 //
 // See package-level documentation on Event Emitter for details.
-func (c *Connection) OnAll(handle func(ConnectionStateChange)) (off func()) {
-	return c.state.eventEmitter.OnAll(func(change emitterData) {
+func (em ConnectionEventEmitter) OnAll(handle func(ConnectionStateChange)) (off func()) {
+	return em.emitter.OnAll(func(change emitterData) {
 		handle(change.(ConnectionStateChange))
 	})
 }
@@ -364,8 +363,8 @@ func (c *Connection) OnAll(handle func(ConnectionStateChange)) (off func()) {
 // Once registers an one-off event handler for connection events of a specific kind.
 //
 // See package-level documentation on Event Emitter for details.
-func (c *Connection) Once(e ConnectionEvent, handle func(ConnectionStateChange)) (off func()) {
-	return c.state.eventEmitter.On(e, func(change emitterData) {
+func (em ConnectionEventEmitter) Once(e ConnectionEvent, handle func(ConnectionStateChange)) (off func()) {
+	return em.emitter.Once(e, func(change emitterData) {
 		handle(change.(ConnectionStateChange))
 	})
 }
@@ -373,33 +372,24 @@ func (c *Connection) Once(e ConnectionEvent, handle func(ConnectionStateChange))
 // OnceAll registers an one-off event handler for all connection events.
 //
 // See package-level documentation on Event Emitter for details.
-func (c *Connection) OnceAll(handle func(ConnectionStateChange)) (off func()) {
-	return c.state.eventEmitter.OnceAll(func(change emitterData) {
+func (em ConnectionEventEmitter) OnceAll(handle func(ConnectionStateChange)) (off func()) {
+	return em.emitter.OnceAll(func(change emitterData) {
 		handle(change.(ConnectionStateChange))
 	})
 }
 
-// offState removes c from listening on the given connection state transitions.
-//
-// If no states are given, c is removed for all of the connection's states.
-// If c is nil, the method panics.
-// If c was not registered or is already removed, the method is a nop.
-func (c *Connection) offState(ch chan<- State, states ...StateEnum) {
-	c.state.off(ch, states...)
-}
-
 // Off deregisters event handlers for connection events of a specific kind.
 //
 // See package-level documentation on Event Emitter for details.
-func (c *Connection) Off(e ConnectionEvent) {
-	c.state.eventEmitter.Off(e)
+func (em ConnectionEventEmitter) Off(e ConnectionEvent) {
+	em.emitter.Off(e)
 }
 
 // Off deregisters all event handlers.
 //
 // See package-level documentation on Event Emitter for details.
-func (c *Connection) OffAll() {
-	c.state.eventEmitter.OffAll()
+func (em ConnectionEventEmitter) OffAll() {
+	em.emitter.OffAll()
 }
 
 func (c *Connection) updateSerial(msg *proto.ProtocolMessage, listen chan<- error) {
@@ -412,26 +402,26 @@ func (c *Connection) updateSerial(msg *proto.ProtocolMessage, listen chan<- erro
 }
 
 func (c *Connection) send(msg *proto.ProtocolMessage, listen chan<- error) error {
-	c.state.Lock()
-	switch state := c.state.current; state {
-	case StateConnInitialized, StateConnConnecting, StateConnDisconnected:
-		c.state.Unlock()
+	c.mtx.Lock()
+	switch state := c.state; state {
+	case ConnectionStateInitialized, ConnectionStateConnecting, ConnectionStateDisconnected:
+		c.mtx.Unlock()
 		if c.opts.NoQueueing {
-			return stateError(state, errQueueing)
+			return connStateError(state, errQueueing)
 		}
 		c.queue.Enqueue(msg, listen)
 		return nil
-	case StateConnConnected:
+	case ConnectionStateConnected:
 	default:
-		c.state.Unlock()
-		return stateError(state, nil)
+		c.mtx.Unlock()
+		return connStateError(state, nil)
 	}
 	if err := c.verifyAndUpdateMessages(msg); err != nil {
-		c.state.Unlock()
+		c.mtx.Unlock()
 		return err
 	}
 	c.updateSerial(msg, listen)
-	c.state.Unlock()
+	c.mtx.Unlock()
 	return c.conn.Send(msg)
 }
 
@@ -472,18 +462,18 @@ func (c *Connection) verifyAndUpdateMessages(msg *proto.ProtocolMessage) (err er
 }
 
 func (c *Connection) isActive() bool {
-	return c.state.current == StateConnConnecting || c.state.current == StateConnConnected
+	return c.state == ConnectionStateConnecting || c.state == ConnectionStateConnected
 }
 
 func (c *Connection) lockCanReceiveMessages() bool {
-	c.state.Lock()
-	defer c.state.Unlock()
-	return c.state.current == StateConnConnecting || c.state.current == StateConnConnected || c.state.current == StateConnClosing
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	return c.state == ConnectionStateConnecting || c.state == ConnectionStateConnected || c.state == ConnectionStateClosing
 }
 
 func (c *Connection) lockIsActive() bool {
-	c.state.Lock()
-	defer c.state.Unlock()
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
 	return c.isActive()
 }
 
@@ -512,58 +502,58 @@ func (c *Connection) eventloop() {
 		}
 		msg, err := c.conn.Receive(deadline)
 		if err != nil {
-			c.state.Lock()
-			if c.state.current == StateConnClosed {
-				c.state.Unlock()
+			c.mtx.Lock()
+			if c.state == ConnectionStateClosed {
+				c.mtx.Unlock()
 				return
 			}
 
 			// RTN23a
-			c.setState(StateConnDisconnected, err)
-			c.state.Unlock()
+			c.lockSetState(ConnectionStateDisconnected, err)
+			c.mtx.Unlock()
 			c.reconnect(lastActivityAt, connDetails, false)
 			return
 		}
 		lastActivityAt = c.opts.Now()
 		if msg.ConnectionSerial != 0 {
-			c.state.Lock()
+			c.mtx.Lock()
 			c.setSerial(msg.ConnectionSerial)
-			c.state.Unlock()
+			c.mtx.Unlock()
 		}
 		switch msg.Action {
 		case proto.ActionHeartbeat:
 		case proto.ActionAck:
-			c.state.Lock()
+			c.mtx.Lock()
 			c.pending.Ack(msg.MsgSerial, msg.Count, newErrorProto(msg.Error))
 			c.setSerial(c.serial + 1)
-			c.state.Unlock()
+			c.mtx.Unlock()
 		case proto.ActionNack:
-			c.state.Lock()
+			c.mtx.Lock()
 			c.pending.Nack(msg.MsgSerial, msg.Count, newErrorProto(msg.Error))
-			c.state.Unlock()
+			c.mtx.Unlock()
 		case proto.ActionError:
 			if msg.Channel != "" {
 				c.callbacks.onChannelMsg(msg)
 				break
 			}
 
-			c.state.Lock()
+			c.mtx.Lock()
 			reauthorizing := c.reauthorizing
 			c.reauthorizing = false
 			if isTokenError(msg.Error) {
 				if reauthorizing {
 					c.lockedReauthorizationFailed(newErrorProto(msg.Error))
-					c.state.Unlock()
+					c.mtx.Unlock()
 					return
 				} else {
 					// TODO: RTN14b; may reuse c.reauthorize from RTN15h2.
 				}
 			}
-			c.state.Unlock()
+			c.mtx.Unlock()
 
 			c.failedConnSideEffects(msg.Error)
 		case proto.ActionConnected:
-			c.state.Lock()
+			c.mtx.Lock()
 			// we need to get this before we set c.key so as to be sure if we were
 			// resuming or recovering the connection.
 			mode := c.getMode()
@@ -592,24 +582,24 @@ func (c *Connection) eventloop() {
 				c.msgSerial = msgSerial
 			}
 			c.setSerial(-1)
-			c.state.Unlock()
+			c.mtx.Unlock()
 			if reconnecting {
 				// (RTN15c1) (RTN15c2)
-				c.state.Lock()
-				c.setState(StateConnConnected, newErrorProto(msg.Error))
-				c.state.Unlock()
+				c.mtx.Lock()
+				c.lockSetState(ConnectionStateConnected, newErrorProto(msg.Error))
+				c.mtx.Unlock()
 				if previousID != msg.ConnectionID {
 					// (RTN15c3)
 					// we are calling this outside of locks to avoid deadlock because in the
 					// RealtimeClient client where this callback is implemented we do some ops
-					// with this Conn where we re acquire Conn.state.Lock again.
+					// with this Conn where we re acquire Conn.Lock again.
 					c.callbacks.onReconnected(msg.Error, true)
 				}
 			} else {
 				// preserve old behavior.
-				c.state.Lock()
-				c.setState(StateConnConnected, nil)
-				c.state.Unlock()
+				c.mtx.Lock()
+				c.lockSetState(ConnectionStateConnected, nil)
+				c.mtx.Unlock()
 			}
 			c.queue.Flush()
 		case proto.ActionDisconnected:
@@ -630,10 +620,10 @@ func (c *Connection) eventloop() {
 			c.reauthorize(lastActivityAt, connDetails)
 			return
 		case proto.ActionClosed:
-			c.state.Lock()
+			c.mtx.Lock()
 			c.id, c.key = "", "" //(RTN16c)
-			c.setState(StateConnClosed, nil)
-			c.state.Unlock()
+			c.lockSetState(ConnectionStateClosed, nil)
+			c.mtx.Unlock()
 			if c.conn != nil {
 				c.conn.Close()
 			}
@@ -643,27 +633,15 @@ func (c *Connection) eventloop() {
 	}
 }
 
-func (c *Connection) setState(state StateEnum, err error) error {
-	// TODO: Tempporary hack to fix https://github.com/ably/ably-go/issues/68.
-	//
-	// The proper way of propagating state changes is through the new
-	// EventEmitter at https://github.com/ably/ably-go/pull/144.
-	ch := make(chan State, 1)
-	c.state.once(ch)
-	go func() { c.callbacks.onStateChange(<-ch) }()
-
-	return c.state.set(state, err)
-}
-
 func (c *Connection) failedConnSideEffects(err *proto.ErrorInfo) {
-	c.state.Lock()
+	c.mtx.Lock()
 	if c.reconnecting {
 		c.reconnecting = false
 		c.reauthorizing = false
 		c.callbacks.onReconnectionFailed(err)
 	}
-	c.setState(StateConnFailed, newErrorProto(err))
-	c.state.Unlock()
+	c.lockSetState(ConnectionStateFailed, newErrorProto(err))
+	c.mtx.Unlock()
 	c.queue.Fail(newErrorProto(err))
 	if c.conn != nil {
 		c.conn.Close()
@@ -671,23 +649,23 @@ func (c *Connection) failedConnSideEffects(err *proto.ErrorInfo) {
 }
 
 func (c *Connection) reauthorize(lastActivityAt time.Time, connDetails *proto.ConnectionDetails) {
-	c.state.Lock()
+	c.mtx.Lock()
 	_, err := c.auth.reauthorize()
 	if err != nil {
 		c.lockedReauthorizationFailed(err)
-		c.state.Unlock()
+		c.mtx.Unlock()
 		return
 	}
 
 	// The reauthorize above will have set the new token in c.auth, so
 	// reconnecting will use the new token.
 	c.reauthorizing = true
-	c.state.Unlock()
+	c.mtx.Unlock()
 	c.reconnect(lastActivityAt, connDetails, false)
 }
 
 func (c *Connection) lockedReauthorizationFailed(err error) {
-	c.setState(StateConnDisconnected, err)
+	c.lockSetState(ConnectionStateDisconnected, err)
 	// TODO: RTN14d
 }
 
@@ -714,3 +692,32 @@ func (vc verboseConn) Close() error {
 	vc.logger.Printf(LogVerbose, "Realtime Connection: closed")
 	return vc.conn.Close()
 }
+
+func (c *Connection) setState(state ConnectionState, err error) error {
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
+	return c.lockSetState(state, err)
+}
+
+func (c *Connection) lockSetState(state ConnectionState, err error) error {
+	previous := c.state
+	changed := c.state != state
+	c.state = state
+	c.errorReason = connStateError(state, err)
+	change := ConnectionStateChange{
+		Current:  c.state,
+		Previous: previous,
+		Reason:   c.errorReason,
+	}
+	if !changed {
+		change.Event = ConnectionEventUpdated
+	} else {
+		change.Event = ConnectionEvent(change.Current)
+	}
+	c.internalEmitter.emitter.Emit(change.Event, change)
+	c.emitter.Emit(change.Event, change)
+	if c.errorReason == nil {
+		return nil
+	}
+	return c.errorReason
+}
diff --git a/ably/realtime_presence.go b/ably/realtime_presence.go
index f39bd29..eb7f578 100644
--- a/ably/realtime_presence.go
+++ b/ably/realtime_presence.go
@@ -48,7 +48,7 @@ func newRealtimePresence(channel *RealtimeChannel) *RealtimePresence {
 
 func (pres *RealtimePresence) verifyChanState() error {
 	switch state := pres.channel.State(); state {
-	case StateChanDetached, StateChanDetaching, StateChanClosing, StateChanClosed, StateChanFailed:
+	case ChannelStateDetached, ChannelStateDetaching, ChannelStateFailed:
 		return newError(91001, fmt.Errorf("unable to enter presence channel (invalid channel state: %s)", state.String()))
 	default:
 		return nil
@@ -64,7 +64,7 @@ func (pres *RealtimePresence) send(msg *proto.PresenceMessage) (Result, error) {
 	}
 	protomsg := &proto.ProtocolMessage{
 		Action:   proto.ActionPresence,
-		Channel:  pres.channel.state.channel,
+		Channel:  pres.channel.Name,
 		Presence: []*proto.PresenceMessage{msg},
 	}
 	return pres.channel.send(protomsg)
diff --git a/ably/state.go b/ably/state.go
index 88bf910..de18f79 100644
--- a/ably/state.go
+++ b/ably/state.go
@@ -8,71 +8,6 @@ import (
 	"github.com/ably/ably-go/ably/proto"
 )
 
-// StateType specifies which group of states is relevant in given context;
-// either:
-//
-//   - StateConn* group describing Conn states
-//   - StateChan* group describing RealtimeChannel states
-//
-type StateType int
-
-const (
-	StateConn StateType = 1 + iota
-	StateChan
-)
-
-// Strings implements the fmt.Stringer interface.
-func (st StateType) String() string {
-	switch st {
-	case StateConn:
-		return "connection"
-	case StateChan:
-		return "channel"
-	default:
-		return "invalid"
-	}
-}
-
-// Contains returns true when the state belongs to the given type.
-func (st StateType) Contains(state StateEnum) bool {
-	return stateMasks[st]&state == state
-}
-
-// StateEnum is an enumeration type for connection and channel states.
-type StateEnum int
-
-// String implements the fmt.Stringer interface.
-func (sc StateEnum) String() string {
-	if s, ok := stateText[sc]; ok {
-		return s
-	}
-	return "invalid"
-}
-
-// StateConn describes states of realtime connection.
-const (
-	StateConnInitialized StateEnum = 1 << iota
-	StateConnConnecting
-	StateConnConnected
-	StateConnDisconnected
-	StateConnSuspended
-	StateConnClosing
-	StateConnClosed
-	StateConnFailed
-)
-
-// StateChan describes states of realtime channel.
-const (
-	StateChanInitialized StateEnum = 1 << (iota + 8)
-	StateChanAttaching
-	StateChanAttached
-	StateChanDetaching
-	StateChanDetached
-	StateChanClosing
-	StateChanClosed
-	StateChanFailed
-)
-
 // Result awaits completion of asynchronous operation.
 type Result interface {
 	// Wait blocks until asynchronous operation is completed. Upon its completion,
@@ -88,58 +23,6 @@ func wait(res Result, err error) error {
 	return res.Wait()
 }
 
-var stateText = map[StateEnum]string{
-	StateConnInitialized:  "ably.StateConnInitialized",
-	StateConnConnecting:   "ably.StateConnConnecting",
-	StateConnConnected:    "ably.StateConnConnected",
-	StateConnDisconnected: "ably.StateConnDisconnected",
-	StateConnSuspended:    "ably.StateConnSuspended",
-	StateConnClosing:      "ably.StateConnClosing",
-	StateConnClosed:       "ably.StateConnClosed",
-	StateConnFailed:       "ably.StateConnFailed",
-	StateChanInitialized:  "ably.StateChanInitialized",
-	StateChanAttaching:    "ably.StateChanAttaching",
-	StateChanAttached:     "ably.StateChanAttached",
-	StateChanDetaching:    "ably.StateChanDetaching",
-	StateChanDetached:     "ably.StateChanDetached",
-	StateChanClosing:      "ably.StateChanClosing",
-	StateChanClosed:       "ably.StateChanClosed",
-	StateChanFailed:       "ably.StateChanFailed",
-}
-
-// stateAll lists all valid connection and channel state values.
-var stateAll = map[StateType][]StateEnum{
-	StateConn: {
-		StateConnInitialized,
-		StateConnConnecting,
-		StateConnConnected,
-		StateConnDisconnected,
-		StateConnSuspended,
-		StateConnClosing,
-		StateConnClosed,
-		StateConnFailed,
-	},
-	StateChan: {
-		StateChanInitialized,
-		StateChanAttaching,
-		StateChanAttached,
-		StateChanDetaching,
-		StateChanClosed,
-		StateChanDetached,
-		StateChanFailed,
-	},
-}
-
-// stateMasks is used for testing connection and channel state values.
-var stateMasks = map[StateType]StateEnum{
-	StateConn: StateConnInitialized | StateConnConnecting | StateConnConnected |
-		StateConnDisconnected | StateConnSuspended | StateConnClosing | StateConnClosed |
-		StateConnFailed,
-	StateChan: StateChanInitialized | StateChanAttaching | StateChanAttached |
-		StateChanDetaching | StateChanDetached | StateChanClosing | StateChanClosed |
-		StateChanFailed,
-}
-
 var (
 	errDisconnected   = newErrorf(80003, "Connection temporarily unavailable")
 	errSuspended      = newErrorf(80002, "Connection unavailable")
@@ -147,21 +30,20 @@ var (
 	errNeverConnected = newErrorf(80002, "Unable to establish connection")
 )
 
-var stateErrors = map[StateEnum]ErrorInfo{
-	StateConnInitialized:  *errNeverConnected,
-	StateConnDisconnected: *errDisconnected,
-	StateConnFailed:       *errFailed,
-	StateConnSuspended:    *errSuspended,
-	StateChanFailed:       *errFailed,
+var connStateErrors = map[ConnectionState]ErrorInfo{
+	ConnectionStateInitialized:  *errNeverConnected,
+	ConnectionStateDisconnected: *errDisconnected,
+	ConnectionStateFailed:       *errFailed,
+	ConnectionStateSuspended:    *errSuspended,
 }
 
-func stateError(state StateEnum, err error) *ErrorInfo {
+func connStateError(state ConnectionState, err error) *ErrorInfo {
 	// Set default error information associated with the target state.
 	e, ok := err.(*ErrorInfo)
 	if ok {
 		return e
 	}
-	if e, ok := stateErrors[state]; ok {
+	if e, ok := connStateErrors[state]; ok {
 		if err != nil {
 			e.err = err
 		}
@@ -170,166 +52,23 @@ func stateError(state StateEnum, err error) *ErrorInfo {
 	return newError(0, err)
 }
 
-// State describes a single state transition of either realtime connection or channel
-// that occurred due to some external condition (dropped connection, retried etc.).
-//
-// Each realtime connection and channel maintains its state to ensure high availability
-// and resilience, which is inherently asynchronous. In order to listen to transition
-// between states for both realtime connection and realtime channel user may provide
-// a channel, which will get notified with single State value for each transition
-// than takes place.
-type State struct {
-	Channel string     // channel name or empty if Type is StateConn
-	Err     *ErrorInfo // eventual error value associated with transition
-	State   StateEnum  // state which connection or channel has transitioned to
-	Type    StateType  // whether transition happened on connection or channel
-}
-
-type stateEmitter struct {
-	sync.Mutex
-	channel   string
-	listeners map[StateEnum]map[chan<- State]struct{}
-	onetime   map[StateEnum]map[chan<- State]struct{}
-	err       *ErrorInfo
-	current   StateEnum
-	typ       StateType
-	logger    *LoggerOptions
-
-	eventEmitter *eventEmitter
-}
-
-func newStateEmitter(typ StateType, startState StateEnum, channel string, log *LoggerOptions) *stateEmitter {
-	if !typ.Contains(startState) {
-		panic(`invalid start state: "` + startState.String() + `"`)
-	}
-	return &stateEmitter{
-		channel:   channel,
-		listeners: make(map[StateEnum]map[chan<- State]struct{}),
-		onetime:   make(map[StateEnum]map[chan<- State]struct{}),
-		current:   startState,
-		typ:       typ,
-		logger:    log,
-
-		eventEmitter: newEventEmitter(log),
-	}
+var channelStateErrors = map[ChannelState]ErrorInfo{
+	ChannelStateFailed: *errFailed,
 }
 
-func (s *stateEmitter) set(state StateEnum, err error) error {
-	previous := s.current
-	changed := s.current != state
-	s.current = state
-	s.err = stateError(state, err)
-	if changed || err != nil {
-		s.emit(State{
-			Channel: s.channel,
-			Err:     s.err,
-			State:   s.current,
-			Type:    s.typ,
-		})
-	}
-
-	if StateConn.Contains(state) {
-		previous := mapOldToNewConnState(previous)
-		change := ConnectionStateChange{
-			Current:  mapOldToNewConnState(s.current),
-			Previous: previous,
-			Reason:   s.err,
-		}
-		if !changed {
-			change.Event = ConnectionEventUpdated
-		} else {
-			change.Event = ConnectionEvent(change.Current)
-		}
-		s.eventEmitter.Emit(change.Event, change)
-	}
-
-	return s.err
-}
-
-func (s *stateEmitter) emit(st State) {
-	for ch := range s.listeners[st.State] {
-		select {
-		case ch <- st:
-		default:
-			s.logger.Printf(LogWarning, "dropping %s due to slow receiver", st)
-		}
-	}
-	onetime := s.onetime[st.State]
-	if len(onetime) != 0 {
-		delete(s.onetime, st.State)
-		for ch := range onetime {
-			select {
-			case ch <- st:
-			default:
-				s.logger.Printf(LogWarning, "dropping %s due to slow receiver", st)
-			}
-			for _, l := range s.onetime {
-				delete(l, ch)
-			}
-		}
-	}
-}
-
-func (s *stateEmitter) syncSet(state StateEnum, err error) error {
-	s.Lock()
-	defer s.Unlock()
-	return s.set(state, err)
-}
-
-func (s *stateEmitter) once(ch chan<- State, states ...StateEnum) {
-	if len(states) == 0 {
-		states = stateAll[s.typ]
-	}
-	for _, state := range states {
-		l, ok := s.onetime[state]
-		if !ok {
-			l = make(map[chan<- State]struct{})
-			s.onetime[state] = l
-		}
-		l[ch] = struct{}{}
-	}
-}
-
-func (s *stateEmitter) on(ch chan<- State, states ...StateEnum) {
-	if ch == nil {
-		panic(fmt.Sprintf("ably: %s On using nil channel", s.typ))
-	}
-	if len(states) == 0 {
-		states = stateAll[s.typ]
-	}
-	s.Lock()
-	for _, state := range states {
-		if !s.typ.Contains(state) {
-			panic(fmt.Sprintf("ably: %s On using invalid state value: %s", s.typ, state.String()))
-		}
-		l, ok := s.listeners[state]
-		if !ok {
-			l = make(map[chan<- State]struct{})
-			s.listeners[state] = l
-		}
-		l[ch] = struct{}{}
-	}
-	s.Unlock()
-}
-
-func (s *stateEmitter) off(ch chan<- State, states ...StateEnum) {
-	if ch == nil {
-		panic(fmt.Sprintf("ably: %s Off using nil channel", s.typ))
-	}
-	if len(states) == 0 {
-		states = stateAll[s.typ]
+func channelStateError(state ChannelState, err error) *ErrorInfo {
+	// Set default error information associated with the target state.
+	e, ok := err.(*ErrorInfo)
+	if ok {
+		return e
 	}
-	s.Lock()
-	for _, state := range states {
-		if !s.typ.Contains(state) {
-			panic(fmt.Sprintf("ably: %s Off using invalid state value: %s", s.typ, state.String()))
-		}
-		delete(s.listeners[state], ch)
-		if len(s.listeners[state]) == 0 {
-			delete(s.listeners, state)
+	if e, ok := channelStateErrors[state]; ok {
+		if err != nil {
+			e.err = err
 		}
+		err = &e
 	}
-	s.Unlock()
+	return newError(0, err)
 }
 
 // queuedEmitter emits confirmation events triggered by ACK or NACK messages.
@@ -507,44 +246,67 @@ func (res *errResult) Wait() error {
 	return res.err
 }
 
-type stateResult struct {
-	err      error
-	listen   <-chan State
-	expected StateEnum
-}
+type resultFunc func() error
 
-func newResult(expected StateEnum) (Result, chan<- State) {
-	listen := make(chan State, 1)
-	res := &stateResult{listen: listen, expected: expected}
-	return res, listen
+func (f resultFunc) Wait() error {
+	return f()
 }
 
-func (s *stateEmitter) listenResult(states ...StateEnum) Result {
-	res, listen := newResult(states[0])
-	s.once(listen, states...)
-	return res
-}
+func (e ChannelEventEmitter) listenResult(expected ChannelState, failed ...ChannelState) Result {
+	changes := make(channelStateChanges, 1)
 
-// Wait implements the Result interface.
-func (res *stateResult) Wait() error {
-	if res == nil {
-		return nil
+	var offs []func()
+	offs = append(offs, e.Once(ChannelEvent(expected), changes.Receive))
+	for _, ev := range failed {
+		offs = append(offs, e.Once(ChannelEvent(ev), changes.Receive))
 	}
-	if res.listen != nil {
-		switch state := <-res.listen; {
-		case state.State == res.expected:
-		case state.Err != nil:
-			res.err = state.Err
-		default:
-			code := 50001
-			if state.Type == StateConn {
-				code = 50002
+
+	return resultFunc(func() error {
+		defer func() {
+			for _, off := range offs {
+				off()
 			}
-			res.err = newError(code, fmt.Errorf("failed %s state: %s", state.Type, state.State))
+		}()
+
+		switch change := <-changes; {
+		case change.Current == expected:
+		case change.Reason != nil:
+			return change.Reason
+		default:
+			code := ErrInternalChannelError
+			return newError(code, fmt.Errorf("failed channel change: %s", change.Current))
 		}
-		res.listen = nil
+
+		return nil
+	})
+}
+func (e ConnectionEventEmitter) listenResult(expected ConnectionState, failed ...ConnectionState) Result {
+	changes := make(connStateChanges, 1)
+
+	var offs []func()
+	offs = append(offs, e.Once(ConnectionEvent(expected), changes.Receive))
+	for _, ev := range failed {
+		offs = append(offs, e.Once(ConnectionEvent(ev), changes.Receive))
 	}
-	return res.err
+
+	return resultFunc(func() error {
+		defer func() {
+			for _, off := range offs {
+				off()
+			}
+		}()
+
+		switch change := <-changes; {
+		case change.Current == expected:
+		case change.Reason != nil:
+			return change.Reason
+		default:
+			code := ErrInternalConnectionError
+			return newError(code, fmt.Errorf("failed connection change: %s", change.Current))
+		}
+
+		return nil
+	})
 }
 
 // A ConnectionState identifies the state of an Ably realtime connection.
@@ -606,25 +368,63 @@ type ConnectionStateChange struct {
 
 func (ConnectionStateChange) isEmitterData() {}
 
-func mapOldToNewConnState(old StateEnum) ConnectionState {
-	switch old {
-	case StateConnInitialized:
-		return ConnectionStateInitialized
-	case StateConnConnecting:
-		return ConnectionStateConnecting
-	case StateConnConnected:
-		return ConnectionStateConnected
-	case StateConnDisconnected:
-		return ConnectionStateDisconnected
-	case StateConnSuspended:
-		return ConnectionStateSuspended
-	case StateConnClosing:
-		return ConnectionStateClosing
-	case StateConnClosed:
-		return ConnectionStateClosed
-	case StateConnFailed:
-		return ConnectionStateFailed
-	default:
-		panic(fmt.Errorf("unexpected StateEnum: %v", old))
-	}
+// A ChannelState identifies the state of an Ably realtime channel.
+type ChannelState struct {
+	name string
+}
+
+var (
+	ChannelStateInitialized = ChannelState{name: "INITIALIZED"}
+	ChannelStateAttaching   = ChannelState{name: "ATTACHING"}
+	ChannelStateAttached    = ChannelState{name: "ATTACHED"}
+	ChannelStateDetaching   = ChannelState{name: "DETACHING"}
+	ChannelStateDetached    = ChannelState{name: "DETACHED"}
+	ChannelStateSuspended   = ChannelState{name: "SUSPENDED"}
+	ChannelStateFailed      = ChannelState{name: "FAILED"}
+)
+
+func (e ChannelState) String() string {
+	return e.name
+}
+
+// A ChannelEvent identifies an event in the lifetime of an Ably realtime
+// channel.
+type ChannelEvent struct {
+	name string
+}
+
+func (ChannelEvent) isEmitterEvent() {}
+
+var (
+	ChannelEventInitialized = ChannelEvent(ChannelStateInitialized)
+	ChannelEventAttaching   = ChannelEvent(ChannelStateAttaching)
+	ChannelEventAttached    = ChannelEvent(ChannelStateAttached)
+	ChannelEventDetaching   = ChannelEvent(ChannelStateDetaching)
+	ChannelEventDetached    = ChannelEvent(ChannelStateDetached)
+	ChannelEventSuspended   = ChannelEvent(ChannelStateSuspended)
+	ChannelEventFailed      = ChannelEvent(ChannelStateFailed)
+	ChannelEventUpdated     = ChannelEvent{name: "UPDATED"}
+)
+
+func (e ChannelEvent) String() string {
+	return e.name
+}
+
+// A ChannelStateChange is the data associated with a ChannelEvent.
+//
+// If the Event is a ChannelEventUpdated, Current and Previous are the
+// the same. Otherwise, the event is a state transition from Previous to
+// Current.
+type ChannelStateChange struct {
+	Current  ChannelState
+	Event    ChannelEvent
+	Previous ChannelState
+	// Reason, if any, is an error that caused the state change.
+	Reason *ErrorInfo
+	// Resumed is set to true for Attached and Update events when channel state
+	// has been maintainted without interruption in the server, so there has
+	// been no loss of message continuity.
+	Resumed bool
 }
+
+func (ChannelStateChange) isEmitterData() {}


Test patch:
diff --git a/ably/ablytest/ablytest.go b/ably/ablytest/ablytest.go
index 5ed19e9..4c9ab53 100644
--- a/ably/ablytest/ablytest.go
+++ b/ably/ablytest/ablytest.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"reflect"
 	"strconv"
 	"time"
 
@@ -75,3 +76,19 @@ func protocol(opts ably.ClientOptions) string {
 	}
 	return "application/json"
 }
+
+func Contains(s, sub interface{}) bool {
+	return reflectContains(reflect.ValueOf(s), reflect.ValueOf(sub))
+}
+
+func reflectContains(s, sub reflect.Value) bool {
+	for i := 0; i+sub.Len() <= s.Len(); i++ {
+		if reflect.DeepEqual(
+			s.Slice(i, i+sub.Len()).Interface(),
+			sub.Interface(),
+		) {
+			return true
+		}
+	}
+	return false
+}
diff --git a/ably/ablytest/recorders.go b/ably/ablytest/recorders.go
index bd56dbb..7a21b54 100644
--- a/ably/ablytest/recorders.go
+++ b/ably/ablytest/recorders.go
@@ -4,13 +4,11 @@ import (
 	"bytes"
 	"context"
 	"errors"
-	"fmt"
 	"io"
 	"io/ioutil"
 	"net"
 	"net/http"
 	"net/url"
-	"reflect"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -125,168 +123,78 @@ func (rec *RoundTripRecorder) roundTrip(req *http.Request) (*http.Response, erro
 	return resp, err
 }
 
-// StateRecorder provides:
-//
-//   * send ably.State channel for recording state transitions
-//   * goroutine-safe access to recorded state enums
-//
-type StateRecorder struct {
-	Timeout time.Duration // times out waiting for states after this duration; 15s by default
-
+type ConnStatesRecorder struct {
 	mtx    sync.Mutex
-	wg     sync.WaitGroup
-	states []ably.State
-	ch     chan ably.State
-	done   chan struct{}
-	typ    ably.StateType
-}
-
-// NewStateRecorder gives new recorder which purpose is to record states via
-// (*clientOptions).Listener channel.
-//
-// If buffer is > 0, the recorder will use it as a buffer to ensure all states
-// transitions are received.
-// If buffer is <= 0, the recorder will not buffer any states, which can
-// result in some of them being dropped.
-func NewStateRecorder(buffer int) *StateRecorder {
-	return newStateRecorder(buffer, ably.StateChan|ably.StateConn)
+	states []ably.ConnectionState
 }
 
-// NewStateChanRecorder gives new recorder which records channel-related
-// state transitions only.
-func NewStateChanRecorder(buffer int) *StateRecorder {
-	return newStateRecorder(buffer, ably.StateChan)
+func (cs *ConnStatesRecorder) append(state ably.ConnectionState) {
+	cs.mtx.Lock()
+	defer cs.mtx.Unlock()
+	cs.states = append(cs.states, state)
 }
 
-// NewStateConnRecorder gives new recorder which records connection-related
-// state transitions only.
-func NewStateConnRecorder(buffer int) *StateRecorder {
-	return newStateRecorder(buffer, ably.StateConn)
-}
-
-func newStateRecorder(buffer int, typ ably.StateType) *StateRecorder {
-	if buffer < 0 {
-		buffer = 0
-	}
-	rec := &StateRecorder{
-		ch:     make(chan ably.State, buffer),
-		done:   make(chan struct{}),
-		states: make([]ably.State, 0, buffer),
-		typ:    typ,
-	}
-	rec.wg.Add(1)
-	go rec.processIncomingStates()
-	return rec
+func (cs *ConnStatesRecorder) Listen(r *ably.Realtime) (off func()) {
+	cs.append(r.Connection.State())
+	return r.Connection.OnAll(func(c ably.ConnectionStateChange) {
+		cs.append(c.Current)
+	})
 }
 
-func (rec *StateRecorder) processIncomingStates() {
-	defer rec.wg.Done()
-	for {
-		select {
-		case state, ok := <-rec.ch:
-			if !ok {
-				return
-			}
-			if state.Type != 0 && state.Type&rec.typ == 0 {
-				continue
-			}
-			rec.add(state)
-		case <-rec.done:
-			return
-		}
-	}
+func (cs *ConnStatesRecorder) States() []ably.ConnectionState {
+	cs.mtx.Lock()
+	defer cs.mtx.Unlock()
+	return cs.states
 }
 
-// Add appends state to the list of recorded ones, used to ensure ordering
-// of the states by injecting values at certain points of the test.
-func (rec *StateRecorder) Add(state ably.StateEnum) {
-	rec.ch <- ably.State{State: state}
+type ChanStatesRecorder struct {
+	mtx    sync.Mutex
+	states []ably.ChannelState
 }
 
-func (rec *StateRecorder) add(state ably.State) {
-	rec.mtx.Lock()
-	rec.states = append(rec.states, state)
-	rec.mtx.Unlock()
+func (cs *ChanStatesRecorder) append(state ably.ChannelState) {
+	cs.mtx.Lock()
+	defer cs.mtx.Unlock()
+	cs.states = append(cs.states, state)
 }
 
-func (rec *StateRecorder) Channel() chan<- ably.State {
-	return rec.ch
+func (cs *ChanStatesRecorder) Listen(channel *ably.RealtimeChannel) (off func()) {
+	cs.append(channel.State())
+	return channel.OnAll(func(c ably.ChannelStateChange) {
+		cs.append(c.Current)
+	})
 }
 
-// Stop stops separate recording goroutine and waits until it terminates.
-func (rec *StateRecorder) Stop() {
-	close(rec.done)
-	rec.wg.Wait()
-	// Drain listener channel.
-	for {
-		select {
-		case <-rec.ch:
-		default:
-			return
-		}
-	}
+func (cs *ChanStatesRecorder) States() []ably.ChannelState {
+	cs.mtx.Lock()
+	defer cs.mtx.Unlock()
+	return cs.states
 }
 
-// States gives copy of the recorded states, safe for use while the recorder
-// is still running.
-func (rec *StateRecorder) States() []ably.StateEnum {
-	rec.mtx.Lock()
-	defer rec.mtx.Unlock()
-	states := make([]ably.StateEnum, 0, len(rec.states))
-	for _, state := range rec.states {
-		states = append(states, state.State)
-	}
-	return states
+type ConnErrorsRecorder struct {
+	mtx    sync.Mutex
+	errors []*ably.ErrorInfo
 }
 
-// Errors gives copy of the error that recorded events hold. It returns only
-// non-nil errors. If none of the recorded states contained an error, the
-// method returns nil.
-func (rec *StateRecorder) Errors() []*ably.ErrorInfo {
-	var errors []*ably.ErrorInfo
-	rec.mtx.Lock()
-	defer rec.mtx.Unlock()
-	for _, state := range rec.states {
-		if state.Err != nil {
-			errors = append(errors, state.Err)
-		}
+func (ce *ConnErrorsRecorder) appendNonNil(err *ably.ErrorInfo) {
+	ce.mtx.Lock()
+	defer ce.mtx.Unlock()
+	if err != nil {
+		ce.errors = append(ce.errors, err)
 	}
-	return errors
 }
 
-// WaitFor blocks until we observe the given exact states were recorded.
-func (rec *StateRecorder) WaitFor(states []ably.StateEnum) error {
-	done := make(chan struct{})
-	stop := make(chan struct{})
-	go func() {
-		for {
-			select {
-			case <-stop:
-				return
-			default:
-				if recorded := rec.States(); reflect.DeepEqual(states, recorded) {
-					close(done)
-					return
-				}
-				time.Sleep(20 * time.Millisecond)
-			}
-		}
-	}()
-	select {
-	case <-done:
-		return nil
-	case <-time.After(rec.timeout()):
-		close(stop)
-		return fmt.Errorf("WaitFor(%v) has timed out after %v: recorded states were %v",
-			states, rec.timeout(), rec.States())
-	}
+func (ce *ConnErrorsRecorder) Listen(r *ably.Realtime) (off func()) {
+	ce.appendNonNil(r.Connection.ErrorReason())
+	return r.Connection.OnAll(func(c ably.ConnectionStateChange) {
+		ce.appendNonNil(c.Reason)
+	})
 }
 
-func (rec *StateRecorder) timeout() time.Duration {
-	if rec.Timeout != 0 {
-		return rec.Timeout
-	}
-	return 15 * time.Second
+func (ce *ConnErrorsRecorder) Errors() []*ably.ErrorInfo {
+	ce.mtx.Lock()
+	defer ce.mtx.Unlock()
+	return ce.errors
 }
 
 type MessagePipeOption func(*pipeConn)
diff --git a/ably/ablytest/resultgroup.go b/ably/ablytest/resultgroup.go
index 0b8495f..3946111 100644
--- a/ably/ablytest/resultgroup.go
+++ b/ably/ablytest/resultgroup.go
@@ -87,7 +87,18 @@ func ConnWaiter(client *ably.Realtime, do func(), expectedEvent ...ably.Connecti
 	off := client.Connection.OnAll(func(ev ably.ConnectionStateChange) {
 		change <- ev
 	})
-	do()
+	if do != nil {
+		do()
+	}
+	for _, ev := range expectedEvent {
+		if ev == ably.ConnectionEvent(client.Connection.State()) {
+			var err error
+			if errInfo := client.Connection.ErrorReason(); errInfo != nil {
+				err = errInfo
+			}
+			return ResultFunc(func() error { return err })
+		}
+	}
 	return ResultFunc(func() error {
 		defer off()
 		timer := time.After(Timeout)
diff --git a/ably/ablytest/timeout.go b/ably/ablytest/timeout.go
index aec6282..2c8e9eb 100644
--- a/ably/ablytest/timeout.go
+++ b/ably/ablytest/timeout.go
@@ -63,3 +63,20 @@ func (wt WithTimeout) recv(into interface{}, from interface{}) (ok, timeout bool
 	}
 	return ok, chosen == 1
 }
+
+func (wt WithTimeout) IsTrue(pred func() bool) bool {
+	ticker := time.NewTicker(10 * time.Millisecond)
+	defer ticker.Stop()
+	timeout := time.NewTimer(wt.before)
+	defer timeout.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			if pred() {
+				return true
+			}
+		case <-timeout.C:
+			return pred()
+		}
+	}
+}
diff --git a/ably/auth_test.go b/ably/auth_test.go
index f995386..b658f7c 100644
--- a/ably/auth_test.go
+++ b/ably/auth_test.go
@@ -663,8 +663,8 @@ func TestAuth_ClientID(t *testing.T) {
 	time.Sleep(time.Duration(params.TTL) * time.Millisecond)
 	tokenExpiredAt := time.Now()
 
-	failed := make(chan ably.State, 1)
-	client.Connection.OnState(failed, ably.StateConnFailed)
+	failed := make(ably.ConnStateChanges, 1)
+	client.Connection.On(ably.ConnectionEventFailed, failed.Receive)
 
 	// Allow some extra time for the server to reject our token.
 	err = nil
@@ -699,13 +699,11 @@ func TestAuth_ClientID(t *testing.T) {
 	if state := client.Connection.State(); state != ably.ConnectionStateFailed {
 		t.Fatalf("want state=%q; got %q", ably.ConnectionStateFailed, state)
 	}
-	select {
-	case state := <-failed:
-		if state.State != ably.StateConnFailed {
-			t.Fatalf("want state=%q; got %q", ably.StateConnFailed, state.State)
-		}
-	default:
-		t.Fatal("wanted to have StateConnFailed emitted; it wasn't")
+
+	var state ably.ConnectionStateChange
+	ablytest.Soon.Recv(t, &state, failed, t.Fatalf)
+	if state.Current != ably.ConnectionStateFailed {
+		t.Fatalf("want state=%q; got %q", ably.ConnectionStateFailed, state.Current)
 	}
 }
 
diff --git a/ably/export_test.go b/ably/export_test.go
index 69eb08b..189fbba 100644
--- a/ably/export_test.go
+++ b/ably/export_test.go
@@ -85,41 +85,18 @@ type EmitterString string
 func (EmitterString) isEmitterEvent() {}
 func (EmitterString) isEmitterData()  {}
 
-// TODO: Once channels have also an EventEmitter, refactor tests to use
-// EventEmitters for both connection and channels.
-
-func (c *Connection) OnState(ch chan<- State, states ...StateEnum) {
-	c.onState(ch, states...)
-}
-
-func (c *Connection) OffState(ch chan<- State, states ...StateEnum) {
-	c.offState(ch, states...)
-}
-
-func (c *Connection) StateEnum() StateEnum {
-	c.state.Lock()
-	defer c.state.Unlock()
-	return c.state.current
-}
-
 func (c *Connection) RemoveKey() {
-	c.state.Lock()
-	defer c.state.Unlock()
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
 	c.key = ""
 }
 
 func (c *Connection) MsgSerial() int64 {
-	c.state.Lock()
-	defer c.state.Unlock()
+	c.mtx.Lock()
+	defer c.mtx.Unlock()
 	return c.msgSerial
 }
 
-func (os ClientOptions) Listener(ch chan<- State) ClientOptions {
-	return append(os, func(os *clientOptions) {
-		os.Listener = ch
-	})
-}
-
 func (os ClientOptions) RealtimeRequestTimeout(d time.Duration) ClientOptions {
 	return append(os, func(os *clientOptions) {
 		os.RealtimeRequestTimeout = d
@@ -147,3 +124,7 @@ func (os ClientOptions) After(after func(context.Context, time.Duration) <-chan
 func (os ClientOptions) ApplyWithDefaults() *clientOptions {
 	return os.applyWithDefaults()
 }
+
+type ConnStateChanges = connStateChanges
+
+type ChannelStateChanges = channelStateChanges
diff --git a/ably/realtime_channel_spec_test.go b/ably/realtime_channel_spec_test.go
index 41967da..66108d5 100644
--- a/ably/realtime_channel_spec_test.go
+++ b/ably/realtime_channel_spec_test.go
@@ -14,6 +14,147 @@ import (
 	"github.com/ably/ably-go/ably/proto"
 )
 
+func TestRealtimeChannel_RTL2_ChannelEventForStateChange(t *testing.T) {
+	t.Parallel()
+
+	t.Run(fmt.Sprintf("on %s", ably.ChannelStateAttaching), func(t *testing.T) {
+		t.Parallel()
+
+		app, realtime := ablytest.NewRealtime(ably.ClientOptions{}.AutoConnect(false))
+		defer safeclose(t, ablytest.FullRealtimeCloser(realtime), app)
+
+		connectAndWait(t, realtime)
+
+		changes := make(chan ably.ChannelStateChange)
+		defer ablytest.Instantly.NoRecv(t, nil, changes, t.Errorf)
+
+		channel := realtime.Channels.Get("test")
+
+		channel.On(ably.ChannelEventAttaching, func(change ably.ChannelStateChange) {
+			changes <- change
+		})
+
+		_, err := channel.Attach()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+	})
+
+	t.Run(fmt.Sprintf("on %s", ably.ChannelStateAttached), func(t *testing.T) {
+		t.Parallel()
+
+		app, realtime := ablytest.NewRealtime(ably.ClientOptions{}.AutoConnect(false))
+		defer safeclose(t, ablytest.FullRealtimeCloser(realtime), app)
+
+		connectAndWait(t, realtime)
+
+		channel := realtime.Channels.Get("test")
+
+		attachAndWait(t, channel)
+	})
+
+	t.Run(fmt.Sprintf("on %s", ably.ChannelStateDetaching), func(t *testing.T) {
+		t.Parallel()
+
+		app, realtime := ablytest.NewRealtime(ably.ClientOptions{}.AutoConnect(false))
+		defer safeclose(t, ablytest.FullRealtimeCloser(realtime), app)
+
+		connectAndWait(t, realtime)
+
+		channel := realtime.Channels.Get("test")
+
+		attachAndWait(t, channel)
+
+		changes := make(chan ably.ChannelStateChange)
+		defer ablytest.Instantly.NoRecv(t, nil, changes, t.Errorf)
+
+		channel.On(ably.ChannelEventDetaching, func(change ably.ChannelStateChange) {
+			changes <- change
+		})
+
+		_, err := channel.Detach()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+	})
+
+	t.Run(fmt.Sprintf("on %s", ably.ChannelStateDetached), func(t *testing.T) {
+		t.Parallel()
+
+		app, realtime := ablytest.NewRealtime(ably.ClientOptions{}.AutoConnect(false))
+		defer safeclose(t, ablytest.FullRealtimeCloser(realtime), app)
+
+		connectAndWait(t, realtime)
+
+		channel := realtime.Channels.Get("test")
+
+		attachAndWait(t, channel)
+
+		changes := make(chan ably.ChannelStateChange)
+		defer ablytest.Instantly.NoRecv(t, nil, changes, t.Errorf)
+
+		channel.On(ably.ChannelEventDetached, func(change ably.ChannelStateChange) {
+			changes <- change
+		})
+
+		_, err := channel.Detach()
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+	})
+
+	t.Run(fmt.Sprintf("on %s", ably.ChannelStateSuspended), func(t *testing.T) {
+		t.Parallel()
+
+		t.Skip("SUSPENDED not yet implemented")
+	})
+
+	t.Run(fmt.Sprintf("on %s", ably.ChannelEventUpdated), func(t *testing.T) {
+		t.Parallel()
+
+		t.Skip("UPDATED not yet implemented")
+	})
+}
+
+func attachAndWait(t *testing.T, channel *ably.RealtimeChannel) {
+	t.Helper()
+
+	changes := make(chan ably.ChannelStateChange, 2)
+	defer ablytest.Instantly.NoRecv(t, nil, changes, t.Errorf)
+
+	{
+		off := channel.Once(ably.ChannelEventAttached, func(change ably.ChannelStateChange) {
+			changes <- change
+		})
+		defer off()
+	}
+
+	{
+		off := channel.Once(ably.ChannelEventFailed, func(change ably.ChannelStateChange) {
+			changes <- change
+		})
+		defer off()
+	}
+
+	_, err := channel.Attach()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var change ably.ChannelStateChange
+	ablytest.Soon.Recv(t, &change, changes, t.Fatalf)
+
+	if change.Current != ably.ChannelStateAttached {
+		t.Fatalf("unexpected FAILED event: %s", change.Reason)
+	}
+}
+
 func TestRealtimeChannel_RTL13_HandleDetached(t *testing.T) {
 	t.Parallel()
 
@@ -28,7 +169,7 @@ func TestRealtimeChannel_RTL13_HandleDetached(t *testing.T) {
 		in, out chan *proto.ProtocolMessage,
 		c *ably.Realtime,
 		channel *ably.RealtimeChannel,
-		stateChanges chan ably.State,
+		stateChanges ably.ChannelStateChanges,
 		afterCalls chan afterCall,
 	) {
 		in = make(chan *proto.ProtocolMessage, 1)
@@ -66,8 +207,8 @@ func TestRealtimeChannel_RTL13_HandleDetached(t *testing.T) {
 
 		ablytest.Instantly.Recv(t, nil, out, t.Fatalf) // Consume ATTACH
 
-		stateChanges = make(chan ably.State, 10)
-		channel.On(stateChanges)
+		stateChanges = make(ably.ChannelStateChanges, 10)
+		channel.OnAll(stateChanges.Receive)
 
 		return
 	}
@@ -84,10 +225,10 @@ func TestRealtimeChannel_RTL13_HandleDetached(t *testing.T) {
 			Channel: channel.Name,
 		}
 
-		var change ably.State
+		var change ably.ChannelStateChange
 
 		ablytest.Instantly.Recv(t, &change, stateChanges, t.Fatalf)
-		if expected, got := ably.StateChanAttached, change.State; expected != got {
+		if expected, got := ably.ChannelStateAttached, change.Current; expected != got {
 			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change)
 		}
 
@@ -106,11 +247,11 @@ func TestRealtimeChannel_RTL13_HandleDetached(t *testing.T) {
 		// Expect a transition to ATTACHING with the error.
 
 		ablytest.Instantly.Recv(t, &change, stateChanges, t.Fatalf)
-		if expected, got := ably.StateChanAttaching, change.State; expected != got {
+		if expected, got := ably.ChannelStateAttaching, change.Current; expected != got {
 			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change)
 		}
-		if got := fmt.Sprint(change.Err); !strings.Contains(got, errInfo.Message) {
-			t.Fatalf("expected %+v; got %v (error: %+v)", errInfo, got, change.Err)
+		if got := fmt.Sprint(change.Reason); !strings.Contains(got, errInfo.Message) {
+			t.Fatalf("expected %+v; got %v (error: %+v)", errInfo, got, change.Reason)
 		}
 
 		var msg *proto.ProtocolMessage
@@ -126,11 +267,11 @@ func TestRealtimeChannel_RTL13_HandleDetached(t *testing.T) {
 			Channel: channel.Name,
 		}
 		ablytest.Instantly.Recv(t, &change, stateChanges, t.Fatalf)
-		if expected, got := ably.StateChanAttached, change.State; expected != got {
+		if expected, got := ably.ChannelStateAttached, change.Current; expected != got {
 			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change)
 		}
-		if change.Err != nil {
-			t.Fatal(change.Err)
+		if change.Reason != nil {
+			t.Fatal(change.Reason)
 		}
 
 		// Expect the retry loop to be finished.
@@ -157,13 +298,13 @@ func TestRealtimeChannel_RTL13_HandleDetached(t *testing.T) {
 
 		// Expect a state change with the error.
 
-		var change ably.State
+		var change ably.ChannelStateChange
 		ablytest.Instantly.Recv(t, &change, stateChanges, t.Fatalf)
-		if expected, got := ably.StateChanDetached, change.State; expected != got {
+		if expected, got := ably.ChannelStateDetached, change.Current; expected != got {
 			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change)
 		}
-		if got := fmt.Sprint(change.Err); !strings.Contains(got, errInfo.Message) {
-			t.Fatalf("expected %+v; got %v (error: %+v)", errInfo, got, change.Err)
+		if got := fmt.Sprint(change.Reason); !strings.Contains(got, errInfo.Message) {
+			t.Fatalf("expected %+v; got %v (error: %+v)", errInfo, got, change.Reason)
 		}
 
 		// Expect an attempt to attach after channelRetryTimeout.
@@ -178,7 +319,7 @@ func TestRealtimeChannel_RTL13_HandleDetached(t *testing.T) {
 		// Expect a transition to ATTACHING, and an ATTACH message.
 
 		ablytest.Instantly.Recv(t, &change, stateChanges, t.Fatalf)
-		if expected, got := ably.StateChanAttaching, change.State; expected != got {
+		if expected, got := ably.ChannelStateAttaching, change.Current; expected != got {
 			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change)
 		}
 
@@ -195,11 +336,11 @@ func TestRealtimeChannel_RTL13_HandleDetached(t *testing.T) {
 			Channel: channel.Name,
 		}
 		ablytest.Instantly.Recv(t, &change, stateChanges, t.Fatalf)
-		if expected, got := ably.StateChanAttached, change.State; expected != got {
+		if expected, got := ably.ChannelStateAttached, change.Current; expected != got {
 			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change)
 		}
-		if change.Err != nil {
-			t.Fatal(change.Err)
+		if change.Reason != nil {
+			t.Fatal(change.Reason)
 		}
 
 		// Expect the retry loop to be finished.
@@ -226,13 +367,13 @@ func TestRealtimeChannel_RTL13_HandleDetached(t *testing.T) {
 
 		// Expect a state change with the error.
 
-		var change ably.State
+		var change ably.ChannelStateChange
 		ablytest.Instantly.Recv(t, &change, stateChanges, t.Fatalf)
-		if expected, got := ably.StateChanDetached, change.State; expected != got {
+		if expected, got := ably.ChannelStateDetached, change.Current; expected != got {
 			t.Fatalf("expected %v; got %v (event: %+v)", expected, got, change)
 		}
-		if got := fmt.Sprint(change.Err); !strings.Contains(got, errInfo.Message) {
-			t.Fatalf("expected %+v; got %v (error: %+v)", errInfo, got, change.Err)
+		if got := fmt.Sprint(change.Reason); !strings.Contains(got, errInfo.Message) {
+			t.Fatalf("expected %+v; got %v (error: %+v)", errInfo, got, change.Reason)
 		}
 
 		// Expect an attempt to attach after channelRetryTimeout.
diff --git a/ably/realtime_channel_test.go b/ably/realtime_channel_test.go
index e5a0793..01c7350 100644
--- a/ably/realtime_channel_test.go
+++ b/ably/realtime_channel_test.go
@@ -44,46 +44,6 @@ func TestRealtimeChannel_Publish(t *testing.T) {
 	}
 }
 
-func TestRealtimeChannel_Failed(t *testing.T) {
-	t.Parallel()
-	rec := ablytest.NewStateChanRecorder(5)
-	opts := ably.ClientOptions{}.
-		AutoConnect(false).
-		QueueMessages(false).
-		Listener(rec.Channel())
-	app, client := ablytest.NewRealtime(opts)
-	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
-
-	if err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
-		t.Fatalf("Connect()=%v", err)
-	}
-	channel := client.Channels.GetAndAttach("test")
-	if err := ablytest.FullRealtimeCloser(client).Close(); err != nil {
-		t.Fatalf("Close()=%v", err)
-	}
-	want := []ably.StateEnum{
-		ably.StateChanAttaching,
-		ably.StateChanAttached,
-		ably.StateChanClosed,
-		ably.StateChanFailed,
-	}
-	if err := rec.WaitFor(want[:3]); err != nil {
-		t.Fatal(err)
-	}
-	if err := checkError(80000, ablytest.Wait(channel.Publish("im", "closed"))); err != nil {
-		t.Fatal(err)
-	}
-	if err := rec.WaitFor(want); err != nil {
-		t.Fatal(err)
-	}
-	if err := checkError(80000, ablytest.Wait(channel.Detach())); err != nil {
-		t.Fatal(err)
-	}
-	if len(rec.Errors()) == 0 {
-		t.Fatal("want len(errors) != 0")
-	}
-}
-
 func TestRealtimeChannel_Subscribe(t *testing.T) {
 	t.Parallel()
 	app, client1 := ablytest.NewRealtime(nil)
@@ -136,30 +96,9 @@ func TestRealtimeChannel_Subscribe(t *testing.T) {
 	}
 }
 
-var chanCloseTransitions = [][]ably.StateEnum{{
-	ably.StateConnConnecting,
-	ably.StateChanAttaching,
-	ably.StateConnConnected,
-	ably.StateChanAttached,
-	ably.StateChanDetaching,
-	ably.StateChanDetached,
-	ably.StateConnClosing,
-	ably.StateConnClosed,
-}, {
-	ably.StateConnConnecting,
-	ably.StateConnConnected,
-	ably.StateChanAttaching,
-	ably.StateChanAttached,
-	ably.StateChanDetaching,
-	ably.StateChanDetached,
-	ably.StateConnClosing,
-	ably.StateConnClosed,
-}}
-
-func TestRealtimeChannel_Close(t *testing.T) {
+func TestRealtimeChannel_Detach(t *testing.T) {
 	t.Parallel()
-	rec := ablytest.NewStateRecorder(8)
-	app, client := ablytest.NewRealtime(ably.ClientOptions{}.Listener(rec.Channel()))
+	app, client := ablytest.NewRealtime(ably.ClientOptions{})
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
 	channel := client.Channels.Get("test")
@@ -180,16 +119,13 @@ func TestRealtimeChannel_Close(t *testing.T) {
 		if msg.Name != "hello" || !reflect.DeepEqual(msg.Data, "world") {
 			done <- fmt.Errorf(`want name="hello", data="world"; got %s, %v`, msg.Name, msg.Data)
 		}
-		if _, ok = <-sub.MessageChannel(); ok {
-			done <- fmt.Errorf("expected channel to be closed")
-		}
 		done <- nil
 	}()
-	if state := channel.State(); state != ably.StateChanAttached {
-		t.Fatalf("want state=%v; got %v", ably.StateChanAttached, state)
+	if state := channel.State(); state != ably.ChannelStateAttached {
+		t.Fatalf("want state=%v; got %v", ably.ChannelStateAttached, state)
 	}
-	if err := channel.Close(); err != nil {
-		t.Fatalf("channel.Close()=%v", err)
+	if err := ablytest.Wait(channel.Detach()); err != nil {
+		t.Fatalf("channel.Detach()=%v", err)
 	}
 	if err := ablytest.FullRealtimeCloser(client).Close(); err != nil {
 		t.Fatalf("ablytest.FullRealtimeCloser(client).Close()=%v", err)
@@ -202,11 +138,4 @@ func TestRealtimeChannel_Close(t *testing.T) {
 	case <-time.After(ablytest.Timeout):
 		t.Fatalf("waiting on subscribed channel close timed out after %v", ablytest.Timeout)
 	}
-	rec.Stop()
-	for _, expected := range chanCloseTransitions {
-		if err = rec.WaitFor(expected); err != nil {
-			return
-		}
-	}
-	t.Error(err)
 }
diff --git a/ably/realtime_client_test.go b/ably/realtime_client_test.go
index 32b5d1d..d6b01c9 100644
--- a/ably/realtime_client_test.go
+++ b/ably/realtime_client_test.go
@@ -24,15 +24,18 @@ func TestRealtime_RealtimeHost(t *testing.T) {
 		"localhost",
 		"::1",
 	}
-	stateRec := ablytest.NewStateRecorder(len(hosts))
+	var errorsRec ablytest.ConnErrorsRecorder
 	for _, host := range hosts {
 		opts := rec.Options(nil, host)
-		opts = opts.Listener(stateRec.Channel())
 		client, err := ably.NewRealtime(app.Options(opts))
 		if err != nil {
 			t.Errorf("NewRealtime=%s (host=%s)", err, host)
 			continue
 		}
+		{
+			off := errorsRec.Listen(client)
+			defer off()
+		}
 		if state := client.Connection.State(); state != ably.ConnectionStateInitialized {
 			t.Errorf("want state=%v; got %s", ably.ConnectionStateInitialized, state)
 			continue
@@ -41,10 +44,6 @@ func TestRealtime_RealtimeHost(t *testing.T) {
 			t.Errorf("%s (host=%s)", err, host)
 			continue
 		}
-		if state := client.Connection.State(); state != ably.ConnectionStateFailed {
-			t.Errorf("want state=%v; got %s", ably.ConnectionStateFailed, state)
-			continue
-		}
 		if err := checkError(80000, ablytest.FullRealtimeCloser(client).Close()); err != nil {
 			t.Errorf("%s (host=%s)", err, host)
 			continue
@@ -53,19 +52,12 @@ func TestRealtime_RealtimeHost(t *testing.T) {
 			t.Errorf("host %s was not recorded (recorded %v)", host, rec.Hosts)
 		}
 	}
-	want := []ably.StateEnum{
-		ably.StateConnConnecting,
-		ably.StateConnFailed,
-		ably.StateConnConnecting,
-		ably.StateConnFailed,
-		ably.StateConnConnecting,
-		ably.StateConnFailed,
-	}
-	if err := stateRec.WaitFor(want); err != nil {
-		t.Fatal(err)
-	}
-	errors := stateRec.Errors()
-	if len(errors) != len(hosts) {
+
+	var errors []*ably.ErrorInfo
+	if !ablytest.Soon.IsTrue(func() bool {
+		errors = errorsRec.Errors()
+		return len(errors) == len(hosts)
+	}) {
 		t.Fatalf("want len(errors)=%d; got %d", len(hosts), len(errors))
 	}
 	for i, err := range errors {
diff --git a/ably/realtime_conn_spec_test.go b/ably/realtime_conn_spec_test.go
index 6e1e7f8..7ad6766 100644
--- a/ably/realtime_conn_spec_test.go
+++ b/ably/realtime_conn_spec_test.go
@@ -467,8 +467,9 @@ func TestRealtimeConn_RTN15c1(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	chanStateChanges := make(chan ably.State)
-	channel.On(chanStateChanges)
+	chanStateChanges := make(ably.ChannelStateChanges)
+	off := channel.OnAll(chanStateChanges.Receive)
+	defer off()
 
 	stateChanges := make(chan ably.ConnectionStateChange, 16)
 	client.Connection.OnAll(func(c ably.ConnectionStateChange) {
@@ -515,24 +516,25 @@ func TestRealtimeConn_RTN15c1(t *testing.T) {
 		t.Fatal("expected message after connection recovery; got none")
 	}
 
-	select {
-	case <-chanStateChanges:
-		t.Fatal("expected no channel state changes")
-	case <-time.After(50 * time.Millisecond):
-		// (RTN15c1)
-		//
-		// - current connectionId == resume message connectionId
-		// - resume message has no error
-		// - no channel state changes happened.
-		if client.Connection.ID() != metaList[1].messages[0].ConnectionID {
-			t.Errorf("expected %q to equal %q", client.Connection.ID(), metaList[1].messages[0].ConnectionID)
-		}
-		if metaList[1].messages[0].Error != nil {
-			t.Error("expected resume error to be nil")
-		}
-	}
+	// (RTN15c1)
+	//
+	// - current connectionId == resume message connectionId
+	// - resume message has no error
+	// - no channel state changes happened.
 
+	var change ably.ChannelStateChange
+	ablytest.Soon.Recv(t, &change, chanStateChanges, t.Fatalf)
+	if change.Previous != change.Current {
+		t.Errorf("expected no state change; got %+v", change)
+	}
+	if client.Connection.ID() != metaList[1].messages[0].ConnectionID {
+		t.Errorf("expected %q to equal %q", client.Connection.ID(), metaList[1].messages[0].ConnectionID)
+	}
+	if metaList[1].messages[0].Error != nil {
+		t.Error("expected resume error to be nil")
+	}
 }
+
 func TestRealtimeConn_RTN15c2(t *testing.T) {
 	t.Parallel()
 
@@ -577,8 +579,9 @@ func TestRealtimeConn_RTN15c2(t *testing.T) {
 	if err := ablytest.Wait(channel.Attach()); err != nil {
 		t.Fatal(err)
 	}
-	chanStateChanges := make(chan ably.State)
-	channel.On(chanStateChanges)
+	chanStateChanges := make(ably.ChannelStateChanges)
+	off := channel.OnAll(chanStateChanges.Receive)
+	defer off()
 
 	sub, err := channel.Subscribe()
 	if err != nil {
@@ -631,33 +634,34 @@ func TestRealtimeConn_RTN15c2(t *testing.T) {
 		t.Fatal("expected message after connection recovery; got none")
 	}
 
-	<-stateChanges
-	select {
-	case <-chanStateChanges:
-		t.Fatal("expected no channel state changes")
-	case <-time.After(50 * time.Millisecond):
-		// (RTN15c2)
-		//
-		// - current connectionId == resume message connectionId
-		// - resume message has an error
-		// - Conn.Reqson == message resume error
-		// - no channel state changes happened.
-		if client.Connection.ID() != metaList[1].messages[0].ConnectionID {
-			t.Errorf("expected %q to equal %q", client.Connection.ID(), metaList[1].messages[0].ConnectionID)
-		}
-		if metaList[1].messages[0].Error == nil {
-			t.Error("expected resume error")
-		}
-		err = client.Connection.ErrorReason()
-		if err == nil {
-			t.Fatal("expected reason to be set")
-		}
-		reason := err.(*ably.ErrorInfo)
-		if reason.StatusCode != errInfo.StatusCode {
-			t.Errorf("expected %d got %d", errInfo.StatusCode, reason.StatusCode)
-		}
+	// (RTN15c2)
+	//
+	// - current connectionId == resume message connectionId
+	// - resume message has an error
+	// - Conn.Reqson == message resume error
+	// - no channel state changes happened.
+
+	var change ably.ChannelStateChange
+	ablytest.Soon.Recv(t, &change, chanStateChanges, t.Fatalf)
+	if change.Previous != change.Current {
+		t.Errorf("expected no state change; got %+v", change)
+	}
+	if client.Connection.ID() != metaList[1].messages[0].ConnectionID {
+		t.Errorf("expected %q to equal %q", client.Connection.ID(), metaList[1].messages[0].ConnectionID)
+	}
+	if metaList[1].messages[0].Error == nil {
+		t.Error("expected resume error")
+	}
+	err = client.Connection.ErrorReason()
+	if err == nil {
+		t.Fatal("expected reason to be set")
+	}
+	reason := err.(*ably.ErrorInfo)
+	if reason.StatusCode != errInfo.StatusCode {
+		t.Errorf("expected %d got %d", errInfo.StatusCode, reason.StatusCode)
 	}
 }
+
 func TestRealtimeConn_RTN15c3_attached(t *testing.T) {
 	t.Parallel()
 
@@ -703,8 +707,9 @@ func TestRealtimeConn_RTN15c3_attached(t *testing.T) {
 	if err := ablytest.Wait(channel.Attach()); err != nil {
 		t.Fatal(err)
 	}
-	chanStateChanges := make(chan ably.State, 18)
-	channel.On(chanStateChanges, ably.StateChanAttaching)
+	chanStateChanges := make(ably.ChannelStateChanges, 18)
+	off := channel.On(ably.ChannelEventAttaching, chanStateChanges.Receive)
+	defer off()
 
 	stateChanges := make(chan ably.ConnectionStateChange, 16)
 	client.Connection.OnAll(func(c ably.ConnectionStateChange) {
@@ -745,7 +750,7 @@ func TestRealtimeConn_RTN15c3_attached(t *testing.T) {
 
 	<-stateChanges
 
-	var chanState ably.State
+	var chanState ably.ChannelStateChange
 	select {
 	case chanState = <-chanStateChanges:
 	case <-time.After(50 * time.Millisecond):
@@ -753,7 +758,7 @@ func TestRealtimeConn_RTN15c3_attached(t *testing.T) {
 	}
 	// we are testing to make sure we have initiated a new attach for channels
 	// in ATTACHED state.
-	if expected, got := ably.StateChanAttaching, chanState; expected != got.State {
+	if expected, got := ably.ChannelStateAttaching, chanState; expected != got.Current {
 		t.Fatalf("expected transition to %v, got %v", expected, got)
 	}
 	reason := client.Connection.ErrorReason()
@@ -855,7 +860,7 @@ func TestRealtimeConn_RTN15c3_attaching(t *testing.T) {
 
 	// we are testing to make sure we have initiated a new attach for channels
 	// in ATTACHING  state.
-	if expected, got := ably.StateChanAttaching, channel.State(); expected != got {
+	if expected, got := ably.ChannelStateAttaching, channel.State(); expected != got {
 		t.Fatalf("expected transition to %v, got %v", expected, got)
 	}
 	reason := client.Connection.ErrorReason()
@@ -911,8 +916,9 @@ func TestRealtimeConn_RTN15c4(t *testing.T) {
 	if err := ablytest.Wait(channel.Attach()); err != nil {
 		t.Fatal(err)
 	}
-	chanStateChanges := make(chan ably.State, 1)
-	channel.On(chanStateChanges, ably.StateChanFailed)
+	chanStateChanges := make(ably.ChannelStateChanges, 1)
+	off := channel.On(ably.ChannelEventFailed, chanStateChanges.Receive)
+	defer off()
 
 	stateChanges := make(chan ably.ConnectionStateChange, 16)
 	client.Connection.OnAll(func(c ably.ConnectionStateChange) {
@@ -951,13 +957,13 @@ func TestRealtimeConn_RTN15c4(t *testing.T) {
 		t.Fatalf("expected transition to %v, got %v", expected, got)
 	}
 	<-stateChanges
-	var chanState ably.State
+	var chanState ably.ChannelStateChange
 	select {
 	case chanState = <-chanStateChanges:
 	case <-time.After(50 * time.Millisecond):
 		t.Fatal("didn't change state")
 	}
-	if expected, got := ably.StateChanFailed, chanState; expected != got.State {
+	if expected, got := ably.ChannelStateFailed, chanState; expected != got.Current {
 		t.Fatalf("expected transition to %v, got %v", expected, got)
 	}
 	reason := client.Connection.ErrorReason()
@@ -1559,8 +1565,9 @@ func TestRealtimeConn_RTN15i_OnErrorWhenConnected(t *testing.T) {
 
 	errorCode := 50123
 
-	channelFailed := make(chan ably.State, 1)
-	channel.On(channelFailed, ably.StateChanFailed)
+	channelFailed := make(ably.ChannelStateChanges, 1)
+	off := channel.On(ably.ChannelEventFailed, channelFailed.Receive)
+	defer off()
 
 	err = ablytest.ConnWaiter(c, func() {
 		in <- &proto.ProtocolMessage{
@@ -1589,7 +1596,7 @@ func TestRealtimeConn_RTN15i_OnErrorWhenConnected(t *testing.T) {
 
 	ablytest.Instantly.Recv(t, nil, channelFailed, t.Fatalf)
 
-	if expected, got := ably.StateChanFailed, channel.State(); expected != got {
+	if expected, got := ably.ChannelStateFailed, channel.State(); expected != got {
 		t.Fatalf("expected channel in state %v; got %v", expected, got)
 	}
 }
diff --git a/ably/realtime_conn_test.go b/ably/realtime_conn_test.go
index 08cad4e..c239bd9 100644
--- a/ably/realtime_conn_test.go
+++ b/ably/realtime_conn_test.go
@@ -1,7 +1,6 @@
 package ably_test
 
 import (
-	"fmt"
 	"testing"
 	"time"
 
@@ -10,35 +9,23 @@ import (
 	"github.com/ably/ably-go/ably/proto"
 )
 
-func await(fn func() ably.StateEnum, state ably.StateEnum) error {
-	t := time.After(ablytest.Timeout)
-	for {
-		select {
-		case <-t:
-			return fmt.Errorf("waiting for %s state has timed out after %v", state, ablytest.Timeout)
-		default:
-			if fn() == state {
-				return nil
-			}
-		}
-	}
-}
-
-var connTransitions = []ably.StateEnum{
-	ably.StateConnConnecting,
-	ably.StateConnConnected,
-	ably.StateConnClosing,
-	ably.StateConnClosed,
+var connTransitions = []ably.ConnectionState{
+	ably.ConnectionStateConnecting,
+	ably.ConnectionStateConnected,
+	ably.ConnectionStateClosing,
+	ably.ConnectionStateClosed,
 }
 
 func TestRealtimeConn_Connect(t *testing.T) {
 	t.Parallel()
-	rec := ablytest.NewStateRecorder(4)
-	app, client := ablytest.NewRealtime(ably.ClientOptions{}.Listener(rec.Channel()))
+	var rec ablytest.ConnStatesRecorder
+	app, client := ablytest.NewRealtime(ably.ClientOptions{})
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
+	off := rec.Listen(client)
+	defer off()
 
-	if err := await(client.Connection.StateEnum, ably.StateConnConnected); err != nil {
-		t.Fatal(err)
+	if err := ablytest.ConnWaiter(client, nil, ably.ConnectionEventConnected).Wait(); err != nil {
+		t.Fatalf("Connect()=%v", err)
 	}
 	if serial := client.Connection.Serial(); serial != -1 {
 		t.Fatalf("want serial=-1; got %d", serial)
@@ -46,21 +33,23 @@ func TestRealtimeConn_Connect(t *testing.T) {
 	if err := ablytest.FullRealtimeCloser(client).Close(); err != nil {
 		t.Fatalf("ablytest.FullRealtimeCloser(client).Close()=%v", err)
 	}
-	if err := rec.WaitFor(connTransitions); err != nil {
-		t.Fatal(err)
+	if !ablytest.Soon.IsTrue(func() bool {
+		return ablytest.Contains(rec.States(), connTransitions)
+	}) {
+		t.Fatalf("expected %+v, got %+v", connTransitions, rec.States())
 	}
 }
 
 func TestRealtimeConn_NoConnect(t *testing.T) {
 	t.Parallel()
-	rec := ablytest.NewStateRecorder(4)
+	var rec ablytest.ConnStatesRecorder
 	opts := ably.ClientOptions{}.
-		Listener(rec.Channel()).
 		AutoConnect(false)
 	app, client := ablytest.NewRealtime(opts)
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
+	off := rec.Listen(client)
+	defer off()
 
-	client.Connection.OnState(rec.Channel())
 	if err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
 		t.Fatalf("Connect()=%v", err)
 	}
@@ -70,37 +59,34 @@ func TestRealtimeConn_NoConnect(t *testing.T) {
 	if err := ablytest.FullRealtimeCloser(client).Close(); err != nil {
 		t.Fatalf("ablytest.FullRealtimeCloser(client).Close()=%v", err)
 	}
-	rec.Stop()
-	if err := rec.WaitFor(connTransitions); err != nil {
-		t.Error(err)
+	if !ablytest.Soon.IsTrue(func() bool {
+		return ablytest.Contains(rec.States(), connTransitions)
+	}) {
+		t.Fatalf("expected %+v, got %+v", connTransitions, rec.States())
 	}
 }
 
-var connCloseTransitions = []ably.StateEnum{
-	ably.StateConnConnecting,
-	ably.StateConnConnected,
-	ably.StateConnClosing,
-	ably.StateConnClosed,
-}
-
 func TestRealtimeConn_ConnectClose(t *testing.T) {
 	t.Parallel()
-	rec := ablytest.NewStateRecorder(4)
-	app, client := ablytest.NewRealtime(ably.ClientOptions{}.Listener(rec.Channel()))
+	var rec ablytest.ConnStatesRecorder
+	app, client := ablytest.NewRealtime(ably.ClientOptions{})
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
+	off := rec.Listen(client)
+	defer off()
 
-	if err := await(client.Connection.StateEnum, ably.StateConnConnected); err != nil {
+	if err := ablytest.ConnWaiter(client, nil, ably.ConnectionEventConnected).Wait(); err != nil {
 		t.Fatal(err)
 	}
 	if err := ablytest.FullRealtimeCloser(client).Close(); err != nil {
 		t.Fatalf("ablytest.FullRealtimeCloser(client).Close()=%v", err)
 	}
-	if err := await(client.Connection.StateEnum, ably.StateConnClosed); err != nil {
+	if err := ablytest.ConnWaiter(client, nil, ably.ConnectionEventClosed).Wait(); err != nil {
 		t.Fatal(err)
 	}
-	rec.Stop()
-	if err := rec.WaitFor(connCloseTransitions); err != nil {
-		t.Error(err)
+	if !ablytest.Soon.IsTrue(func() bool {
+		return ablytest.Contains(rec.States(), connTransitions)
+	}) {
+		t.Fatalf("expected %+v, got %+v", connTransitions, rec.States())
 	}
 }
 
@@ -135,7 +121,7 @@ func TestRealtimeConn_AuthError(t *testing.T) {
 		t.Fatal("Connect(): want err != nil")
 	}
 	if state := client.Connection.State(); state != ably.ConnectionStateFailed {
-		t.Fatalf("want state=%s; got %s", ably.StateConnFailed, state)
+		t.Fatalf("want state=%s; got %s", ably.ConnectionStateFailed, state)
 	}
 }
 
@@ -163,12 +149,19 @@ func TestRealtimeConn_ReceiveTimeout(t *testing.T) {
 		AutoConnect(false))
 	defer safeclose(t, app)
 
-	states := make(chan ably.State, 10)
-	client.Connection.OnState(states, ably.StateConnConnected, ably.StateConnDisconnected)
+	states := make(ably.ConnStateChanges, 10)
+	{
+		off := client.Connection.On(ably.ConnectionEventConnected, states.Receive)
+		defer off()
+	}
+	{
+		off := client.Connection.On(ably.ConnectionEventDisconnected, states.Receive)
+		defer off()
+	}
 
 	client.Connection.Connect()
 
-	var state ably.State
+	var state ably.ConnectionStateChange
 
 	select {
 	case state = <-states:
@@ -176,7 +169,7 @@ func TestRealtimeConn_ReceiveTimeout(t *testing.T) {
 		t.Fatal("didn't receive state change event")
 	}
 
-	if expected, got := ably.StateConnConnected, state.State; expected != got {
+	if expected, got := ably.ConnectionStateConnected, state.Current; expected != got {
 		t.Fatalf("expected %v, got %v", expected, got)
 	}
 
@@ -187,7 +180,7 @@ func TestRealtimeConn_ReceiveTimeout(t *testing.T) {
 		t.Fatal("didn't receive state change event")
 	}
 
-	if expected, got := ably.StateConnDisconnected, state.State; expected != got {
+	if expected, got := ably.ConnectionStateDisconnected, state.Current; expected != got {
 		t.Fatalf("expected %v, got %v", expected, got)
 	}
 }
diff --git a/ably/realtime_presence_test.go b/ably/realtime_presence_test.go
index a417a46..c7e9eb5 100644
--- a/ably/realtime_presence_test.go
+++ b/ably/realtime_presence_test.go
@@ -2,6 +2,7 @@ package ably_test
 
 import (
 	"fmt"
+	"reflect"
 	"strconv"
 	"testing"
 	"time"
@@ -104,33 +105,29 @@ func TestRealtimePresence_Sync250(t *testing.T) {
 
 func TestRealtimePresence_EnsureChannelIsAttached(t *testing.T) {
 	t.Parallel()
-	presTransitions := []ably.StateEnum{
-		ably.StateConnConnecting,
-		ably.StateConnConnected,
-		ably.StateChanAttaching,
-		ably.StateChanAttached,
+	presTransitions := []ably.ChannelState{
+		ably.ChannelStateInitialized,
+		ably.ChannelStateAttaching,
+		ably.ChannelStateAttached,
 	}
-	rec := ablytest.NewStateRecorder(4)
+	var rec ablytest.ChanStatesRecorder
 	opts := ably.ClientOptions{}.
-		Listener(rec.Channel()).
 		AutoConnect(false)
 	app, client := ablytest.NewRealtime(opts)
 	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 	channel := client.Channels.Get("persisted:presence_fixtures")
+	off := rec.Listen(channel)
+	defer off()
 	if err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
 		t.Fatal(err)
 	}
-	if err := rec.WaitFor(presTransitions[:2]); err != nil {
-		t.Fatal(err)
-	}
 	members, err := channel.Presence.Get(true)
 	if err != nil {
 		t.Fatal(err)
 	}
-	if err := rec.WaitFor(presTransitions); err != nil {
-		t.Fatal(err)
+	if expected, got := presTransitions, rec.States(); !reflect.DeepEqual(expected, got) {
+		t.Fatalf("expected %+v, got %+v", expected, got)
 	}
-	rec.Stop()
 	if err = contains(members, fixtureMembers...); err != nil {
 		t.Fatal(err)
 	}