Repository: ably/ably-go
Pull request #144 (2020-04-03 10:29:32)

v1.2 event emitter for connection (RTN4)
And associated API fixes and test helpers.

As discussed, the spec-compliant API is being appended alongside the old API. We identify the spec-compliant API by appending a `V12` to names. This way, we can work without having to refactor existing, working code, and when we're finished we can make the old API private and remove the `V12` prefixes in a single automated swoop.

The RTN4a test for SUSPENDED is left failing until connection suspension is implemented.


Patch:
diff --git a/README.md b/README.md
index 73164e9..c9e0b7f 100644
--- a/README.md
+++ b/README.md
@@ -50,7 +50,7 @@ specification. If there are features that are currently missing that are a high
 ### Creating a client
 
 ```go
-client, err := ably.NewRealtimeClient(ably.NewClientOptions("xxx:xxx"))
+client, err := ably.NewRealtime(ably.NewClientOptions("xxx:xxx"))
 if err != nil {
 	panic(err)
 }
diff --git a/ably/auth.go b/ably/auth.go
index 9c9ccb4..cac13b0 100644
--- a/ably/auth.go
+++ b/ably/auth.go
@@ -421,9 +421,7 @@ func (a *Auth) isTokenRenewable() bool {
 }
 
 func (a *Auth) newError(code int, err error) error {
-	e := newError(code, err)
-	e.Server = a.host
-	return e
+	return newError(code, err)
 }
 
 func (a *Auth) authReq(req *http.Request) error {
diff --git a/ably/doc.go b/ably/doc.go
new file mode 100644
index 0000000..59113f4
--- /dev/null
+++ b/ably/doc.go
@@ -0,0 +1,33 @@
+// Package ably is the official Ably client library for Go.
+//
+// Get started at https://github.com/ably/ably-go#using-the-realtime-api.
+//
+// Event Emitters
+//
+// An event emitter pattern appears in multiple places in the library.
+//
+// The On method takes an event type identifier and a handler function to be
+// called with the event's associated data whenever an event of that type is
+// emitted in the event emitter. It also returns an "off" function to undo this
+// operation, so that the handler function isn't called anymore.
+//
+// The OnAll method is like On, but for events of all types.
+//
+// The Once method works like On, except the handler is just called once, for
+// the first matching event.
+//
+// OnceAll is like OnAll in the same way Once is like On.
+//
+// The Off method is like calling the "off" function returned by calls to On and
+// Once methods with a matching event type identifier.
+//
+// The OffAll method is like Off, except it is like calling all the "off"
+// functions.
+//
+// Calls to handlers for a single or for subsequent events may be concurrent,
+// but are always ordered. That is, if an event is fired after another, the same
+// handler will be called in the same order that the events were fired.
+//
+// Calling any of these methods an "off" function inside a handler will only
+// have effect for subsequent events.
+package ably
diff --git a/ably/error.go b/ably/error.go
index 132e84d..ffdd77c 100644
--- a/ably/error.go
+++ b/ably/error.go
@@ -8,6 +8,7 @@ import (
 	"mime"
 	"net"
 	"net/http"
+	"strings"
 
 	"github.com/ably/ably-go/ably/proto"
 )
@@ -21,69 +22,84 @@ func toStatusCode(code int) int {
 	}
 }
 
-// Error describes error returned from Ably API. It always has non-zero error
+// ErrorInfo describes error returned from Ably API. It always has non-zero error
 // code. It may contain underlying error value which caused the failure
 // condition.
-type Error struct {
-	Code       int    // internal error code
-	StatusCode int    // HTTP status code
-	Err        error  // underlying error responsible for the failure; may be nil
-	Server     string // non-empty ID of the Ably server which the error was received from
+type ErrorInfo struct {
+	StatusCode int
+	Code       int
+	HRef       string
+	Cause      *ErrorInfo
+
+	// err is the application-level error we're wrapping, or just a message.
+	// If Cause is non-nil, err == *Cause.
+	err error
+}
+
+// Error implements the builtin error interface.
+func (e ErrorInfo) Error() string {
+	errorHref := e.HRef
+	if errorHref == "" && e.Code != 0 {
+		errorHref = fmt.Sprintf("https://help.ably.io/error/%d", e.Code)
+	}
+	msg := e.Message()
+	var see string
+	if !strings.Contains(msg, errorHref) {
+		see = "See " + errorHref
+	}
+	return fmt.Sprintf("[ErrorInfo :%s code=%d statusCode=%d] %s", msg, e.Code, e.StatusCode, see)
 }
 
-// Error implements builtin error interface.
-func (err *Error) Error() string {
-	if err.Err != nil {
-		return fmt.Sprintf("%s (status=%d, internal=%d)", err.Err, err.StatusCode, err.Code)
+// Unwrap implements the implicit interface that errors.Unwrap understands.
+func (e ErrorInfo) Unwrap() error {
+	return e.err
+}
+
+// Message returns the undecorated error message.
+func (e ErrorInfo) Message() string {
+	if e.err == nil {
+		return ""
 	}
-	return errCodeText[err.Code]
+	return e.err.Error()
 }
 
-func newError(code int, err error) *Error {
+func newError(defaultCode int, err error) *ErrorInfo {
 	switch err := err.(type) {
-	case *Error:
-		if _, ok := err.Err.(genericError); ok {
-			// If err was returned from http.Response but we were unable to
-			// parse the internal error code, we overwrite it here.
-			err.Code = code
-			return err
-		}
+	case nil:
+		return nil
+	case *ErrorInfo:
 		return err
 	case net.Error:
 		if err.Timeout() {
-			return &Error{Code: ErrTimeoutError, StatusCode: 500, Err: err}
+			return &ErrorInfo{Code: ErrTimeoutError, StatusCode: 500, err: err}
 		}
 	}
-	return &Error{
-		Code:       code,
-		StatusCode: toStatusCode(code),
-		Err:        err,
+	return &ErrorInfo{
+		Code:       defaultCode,
+		StatusCode: toStatusCode(defaultCode),
+
+		err: err,
 	}
 }
 
-func newErrorf(code int, format string, v ...interface{}) *Error {
-	return &Error{
-		Code:       code,
-		StatusCode: toStatusCode(code),
-		Err:        fmt.Errorf(format, v...),
-	}
+func newErrorf(code int, format string, v ...interface{}) *ErrorInfo {
+	return newError(code, fmt.Errorf(format, v...))
 }
 
-func newErrorProto(err *proto.ErrorInfo) *Error {
+func newErrorProto(err *proto.ErrorInfo) *ErrorInfo {
 	if err == nil {
 		return nil
 	}
-	return &Error{
+	return &ErrorInfo{
 		Code:       err.Code,
 		StatusCode: err.StatusCode,
-		Err:        errors.New(err.Message),
+		HRef:       err.HRef,
+		err:        errors.New(err.Message),
 	}
 }
 
-type genericError error
-
 func code(err error) int {
-	if e, ok := err.(*Error); ok {
+	if e, ok := err.(*ErrorInfo); ok {
 		return e.Code
 	}
 	return 0
@@ -115,20 +131,15 @@ func checkValidHTTPResponse(resp *http.Response) error {
 
 	body := &errorBody{}
 	if err := decode(typ, resp.Body, &body); err != nil {
-		return &Error{
-			Code:       50000,
-			StatusCode: resp.StatusCode,
-			Err:        genericError(errors.New(http.StatusText(resp.StatusCode))),
-		}
+		return newError(resp.StatusCode*100, errors.New(http.StatusText(resp.StatusCode)))
 	}
 
-	err := &Error{
+	err := &ErrorInfo{
 		Code:       body.Error.Code,
 		StatusCode: body.Error.StatusCode,
-		Server:     body.Error.Server,
 	}
 	if body.Error.Message != "" {
-		err.Err = errors.New(body.Error.Message)
+		err.err = errors.New(body.Error.Message)
 	}
 	if err.Code == 0 && err.StatusCode == 0 {
 		err.Code, err.StatusCode = resp.StatusCode*100, resp.StatusCode
diff --git a/ably/event_emitter.go b/ably/event_emitter.go
new file mode 100644
index 0000000..70b5a94
--- /dev/null
+++ b/ably/event_emitter.go
@@ -0,0 +1,199 @@
+package ably
+
+import (
+	"runtime/debug"
+	"sync"
+)
+
+type eventEmitter struct {
+	sync.Mutex
+	listeners listenersForEvent
+	logger    Logger
+}
+
+type emitterEvent interface {
+	isEmitterEvent()
+}
+
+type emitterData interface {
+	isEmitterData()
+}
+
+type listenersForEvent map[emitterEvent]listenerSet
+type listenerSet map[*eventListener]struct{}
+
+type eventListener struct {
+	handler func(emitterData)
+	once    bool
+
+	queueMtx sync.Mutex
+	queue    []emitterData
+}
+
+func (l *eventListener) handle(e emitterData, log Logger) {
+	// The goroutine that finds the queue empty launches a goroutine for emptying it,
+	// ie. process its emitterData and afterwards any emitterData that concurrent
+	// goroutines may have left in the queue in the meantime.
+	//
+	// Other goroutines just enqueue their emitterData.
+	//
+	// This ensures our guarantee that a single handler runs concurrently with
+	// other handlers, but calls to the same handler are sequential and ordered,
+	// while emitting an event doesn't block the emitting goroutine.
+
+	var isBusy bool
+
+	l.queueMtx.Lock()
+	isBusy = len(l.queue) > 0
+	l.queue = append(l.queue, e)
+	l.queueMtx.Unlock()
+
+	if isBusy {
+		return
+	}
+
+	go func() {
+		done := false
+		for !done {
+			l.queueMtx.Lock()
+			e := l.queue[0]
+			l.queueMtx.Unlock()
+
+			safeHandle(e, l.handler, log)
+
+			l.queueMtx.Lock()
+			l.queue = l.queue[1:]
+			done = len(l.queue) == 0
+			l.queueMtx.Unlock()
+		}
+	}()
+}
+
+func safeHandle(e emitterData, handle func(emitterData), log Logger) {
+	defer func() {
+		r := recover()
+		if r == nil {
+			return
+		}
+		log.Printf(LogError, "EventEmitter: panic in event handler: %v\n%s", r, debug.Stack())
+	}()
+
+	handle(e)
+}
+
+func newEventEmitter(logger Logger) *eventEmitter {
+	return &eventEmitter{
+		listeners: listenersForEvent{
+			nil: listenerSet{},
+		},
+		logger: logger,
+	}
+}
+
+// On registers an event listener. The event must be comparable to the
+// eventEmitter's event type, and only events equal to it will trigger the
+// listener.
+//
+// It returns a function to deregister the listener.
+func (em *eventEmitter) On(event emitterEvent, handle func(emitterData)) (off func()) {
+	return em.on(event, handle, false)
+}
+
+// OnAll is like On, except the listener is triggered by all events.
+func (em *eventEmitter) OnAll(handle func(emitterData)) (off func()) {
+	return em.on(nil, handle, false)
+}
+
+// Once is like On, except the listener is deregistered once first triggered.
+func (em *eventEmitter) Once(event emitterEvent, handle func(emitterData)) (off func()) {
+	return em.on(event, handle, true)
+}
+
+// OnceAll is like Once, except the listener is triggered by all events.
+func (em *eventEmitter) OnceAll(handle func(emitterData)) (off func()) {
+	return em.on(nil, handle, true)
+}
+
+func (em *eventEmitter) on(event emitterEvent, handle func(emitterData), once bool) (off func()) {
+	em.Lock()
+	defer em.Unlock()
+
+	l := &eventListener{
+		handler: handle,
+		once:    once,
+	}
+
+	listeners := em.listeners[event]
+	if listeners == nil {
+		listeners = listenerSet{}
+		em.listeners[event] = listeners
+	}
+
+	listeners[l] = struct{}{}
+
+	return func() {
+		em.Lock()
+		defer em.Unlock()
+
+		listeners := em.listeners[event]
+		if listeners != nil {
+			delete(listeners, l)
+		}
+	}
+}
+
+// Off deregisters event listeners. The event must be comparable to the
+// eventEmitter's event type, and only listeners that were associated with that
+// event will be removed.
+func (em *eventEmitter) Off(event emitterEvent) {
+	em.off(event)
+}
+
+// OffAll is like Off, except is deregisters all event listeners.
+func (em *eventEmitter) OffAll() {
+	em.off(nil)
+}
+
+func (em *eventEmitter) off(event emitterEvent) {
+	em.Lock()
+	defer em.Unlock()
+
+	if event != nil {
+		delete(em.listeners, event)
+	} else {
+		em.listeners = listenersForEvent{
+			nil: listenerSet{},
+		}
+	}
+}
+
+func (em *eventEmitter) Emit(event emitterEvent, data emitterData) {
+	// Let's first collect the handlers, and then call them outside the lock.
+	// This allows the handler functions to call again into the event emitter,
+	// which would otherwise deadlock.
+
+	for _, handle := range em.handlersForEvent(event) {
+		handle(data, em.logger)
+	}
+}
+
+func (em *eventEmitter) handlersForEvent(event emitterEvent) (handlers []func(emitterData, Logger)) {
+	em.Lock()
+	defer em.Unlock()
+
+	sets := []listenerSet{em.listeners[nil]}
+	if event != nil {
+		sets = append(sets, em.listeners[event])
+	}
+
+	for _, listeners := range sets {
+		for l, _ := range listeners {
+			if l.once {
+				delete(listeners, l)
+			}
+			handlers = append(handlers, l.handle)
+		}
+	}
+
+	return handlers
+}
diff --git a/ably/options.go b/ably/options.go
index 0c66773..9b00150 100644
--- a/ably/options.go
+++ b/ably/options.go
@@ -23,14 +23,14 @@ const (
 )
 
 var defaultOptions = &ClientOptions{
-	RestHost:             RestHost,
-	FallbackHosts:        DefaultFallbackHosts(),
-	HTTPMaxRetryCount:    3,
-	RealtimeHost:         "realtime.ably.io",
-	TimeoutConnect:       15 * time.Second,
-	TimeoutDisconnect:    30 * time.Second,
-	TimeoutSuspended:     2 * time.Minute,
-	FallbackRetryTimeout: 10 * time.Minute,
+	RestHost:                 RestHost,
+	FallbackHosts:            DefaultFallbackHosts(),
+	HTTPMaxRetryCount:        3,
+	RealtimeHost:             "realtime.ably.io",
+	TimeoutConnect:           15 * time.Second,
+	TimeoutDisconnect:        30 * time.Second,
+	TimeoutSuspended:         2 * time.Minute,
+	FallbackRetryTimeout:     10 * time.Minute,
 	IdempotentRestPublishing: false,
 }
 
@@ -199,13 +199,13 @@ type ClientOptions struct {
 
 	// When true idempotent rest publishing will be enabled.
 	// Spec TO3n
-	IdempotentRestPublishing   bool
-	TimeoutConnect             time.Duration // time period after which connect request is failed
-	TimeoutDisconnect          time.Duration // time period after which disconnect request is failed
-	TimeoutSuspended           time.Duration // time period after which no more reconnection attempts are performed
+	IdempotentRestPublishing bool
+	TimeoutConnect           time.Duration // time period after which connect request is failed
+	TimeoutDisconnect        time.Duration // time period after which disconnect request is failed
+	TimeoutSuspended         time.Duration // time period after which no more reconnection attempts are performed
 
 	// Dial specifies the dial function for creating message connections used
-	// by RealtimeClient.
+	// by Realtime.
 	//
 	// If Dial is nil, the default websocket connection is used.
 	Dial func(protocol string, u *url.URL) (proto.Conn, error)
diff --git a/ably/proto/error.go b/ably/proto/error.go
index 9bf7cf8..0d29825 100644
--- a/ably/proto/error.go
+++ b/ably/proto/error.go
@@ -1,10 +1,5 @@
 package proto
 
-import (
-	"fmt"
-	"strings"
-)
-
 // ErrorInfo describes an error returned via ProtocolMessage.
 type ErrorInfo struct {
 	StatusCode int    `json:"statusCode,omitempty" codec:"statusCode,omitempty"`
@@ -31,16 +26,3 @@ func (e *ErrorInfo) FromMap(ctx map[string]interface{}) {
 		e.Server = v.(string)
 	}
 }
-
-// Error implements the builtin error interface.
-func (e *ErrorInfo) Error() string {
-	errorHref := e.HRef
-	if errorHref == "" && e.Code != 0 {
-		errorHref = fmt.Sprintf("https://help.ably.io/error/%d", e.Code)
-	}
-	var see string
-	if !strings.Contains(e.Message, errorHref) {
-		see = "See " + errorHref
-	}
-	return fmt.Sprintf("[ErrorInfo :%s code=%d statusCode=%d] %s", e.Message, e.Code, e.StatusCode, see)
-}
diff --git a/ably/realtime_channel.go b/ably/realtime_channel.go
index 04fa952..83764e0 100644
--- a/ably/realtime_channel.go
+++ b/ably/realtime_channel.go
@@ -25,11 +25,11 @@ func (ch chanSlice) Sort()              { sort.Sort(ch) }
 // for creating, deleting and iterating over existing channels.
 type Channels struct {
 	mtx    sync.Mutex
-	client *RealtimeClient
+	client *Realtime
 	chans  map[string]*RealtimeChannel
 }
 
-func newChannels(client *RealtimeClient) *Channels {
+func newChannels(client *Realtime) *Channels {
 	return &Channels{
 		client: client,
 		chans:  make(map[string]*RealtimeChannel),
@@ -90,14 +90,14 @@ type RealtimeChannel struct {
 	Name     string            // name used to create the channel
 	Presence *RealtimePresence //
 
-	client *RealtimeClient
+	client *Realtime
 	state  *stateEmitter
 	subs   *subscriptions
 	queue  *msgQueue
 	listen chan State
 }
 
-func newRealtimeChannel(name string, client *RealtimeClient) *RealtimeChannel {
+func newRealtimeChannel(name string, client *Realtime) *RealtimeChannel {
 	c := &RealtimeChannel{
 		Name:   name,
 		client: client,
@@ -110,7 +110,7 @@ func newRealtimeChannel(name string, client *RealtimeClient) *RealtimeChannel {
 	if c.opts().Listener != nil {
 		c.On(c.opts().Listener)
 	}
-	c.client.Connection.On(c.listen, StateConnFailed, StateConnClosed)
+	c.client.Connection.onState(c.listen, StateConnFailed, StateConnClosed)
 	go c.listenLoop()
 	return c
 }
@@ -327,7 +327,7 @@ func (c *RealtimeChannel) send(msg *proto.ProtocolMessage) (Result, error) {
 		return res, nil
 	case StateChanAttached:
 	default:
-		return nil, &Error{Code: 90001}
+		return nil, &ErrorInfo{Code: 90001}
 	}
 	if err := c.client.Connection.send(msg, listen); err != nil {
 		return nil, err
diff --git a/ably/realtime_client.go b/ably/realtime_client.go
index 848bdd2..652574a 100644
--- a/ably/realtime_client.go
+++ b/ably/realtime_client.go
@@ -5,13 +5,13 @@ import (
 	"time"
 )
 
-// The RealtimeClient libraries establish and maintain a persistent connection
+// The Realtime libraries establish and maintain a persistent connection
 // to Ably enabling extremely low latency broadcasting of messages and presence
 // state.
-type RealtimeClient struct {
+type Realtime struct {
 	Auth       *Auth
 	Channels   *Channels
-	Connection *Conn
+	Connection *Connection
 
 	chansMtx sync.RWMutex
 	chans    map[string]*RealtimeChannel
@@ -19,12 +19,12 @@ type RealtimeClient struct {
 	err      chan error
 }
 
-// NewRealtimeClient
-func NewRealtimeClient(opts *ClientOptions) (*RealtimeClient, error) {
+// NewRealtime constructs a new Realtime.
+func NewRealtime(opts *ClientOptions) (*Realtime, error) {
 	if opts == nil {
-		panic("called NewRealtimeClient with nil ClientOptions")
+		panic("called NewRealtime with nil ClientOptions")
 	}
-	c := &RealtimeClient{
+	c := &Realtime{
 		err:   make(chan error),
 		chans: make(map[string]*RealtimeChannel),
 	}
@@ -44,33 +44,38 @@ func NewRealtimeClient(opts *ClientOptions) (*RealtimeClient, error) {
 	return c, nil
 }
 
-// Close
-func (c *RealtimeClient) Close() error {
-	return c.Connection.Close()
+// Connect is the same as Connection.Connect.
+func (c *Realtime) Connect() {
+	c.Connection.Connect()
+}
+
+// Close is the same as Connection.Close.
+func (c *Realtime) Close() {
+	c.Connection.Close()
 }
 
 // Stats gives the clients metrics according to the given parameters. The
 // returned result can be inspected for the statistics via the Stats()
 // method.
-func (c *RealtimeClient) Stats(params *PaginateParams) (*PaginatedResult, error) {
+func (c *Realtime) Stats(params *PaginateParams) (*PaginatedResult, error) {
 	return c.rest.Stats(params)
 }
 
 // Time
-func (c *RealtimeClient) Time() (time.Time, error) {
+func (c *Realtime) Time() (time.Time, error) {
 	return c.rest.Time()
 }
 
-func (c *RealtimeClient) dispatchloop() {
+func (c *Realtime) dispatchloop() {
 	for msg := range c.Connection.msgCh {
 		c.Channels.Get(msg.Channel).notify(msg)
 	}
 }
 
-func (c *RealtimeClient) opts() *ClientOptions {
+func (c *Realtime) opts() *ClientOptions {
 	return &c.rest.opts
 }
 
-func (c *RealtimeClient) logger() *LoggerOptions {
+func (c *Realtime) logger() *LoggerOptions {
 	return c.rest.logger()
 }
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index 94289bb..e047056 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -16,9 +16,9 @@ var (
 	errCloseInactive = errors.New("attempted to close inactive connection")
 )
 
-// Conn represents a single connection RealtimeClient instantiates for
+// Connection represents a single connection Realtime instantiates for
 // communication with Ably servers.
-type Conn struct {
+type Connection struct {
 	details   proto.ConnectionDetails
 	id        string
 	serial    int64
@@ -34,8 +34,8 @@ type Conn struct {
 	auth      *Auth
 }
 
-func newConn(opts *ClientOptions, auth *Auth) (*Conn, error) {
-	c := &Conn{
+func newConn(opts *ClientOptions, auth *Auth) (*Connection, error) {
+	c := &Connection{
 		opts:    opts,
 		msgCh:   make(chan *proto.ProtocolMessage),
 		state:   newStateEmitter(StateConn, StateConnInitialized, "", auth.logger()),
@@ -44,7 +44,7 @@ func newConn(opts *ClientOptions, auth *Auth) (*Conn, error) {
 	}
 	c.queue = newMsgQueue(c)
 	if opts.Listener != nil {
-		c.On(opts.Listener)
+		c.onState(opts.Listener)
 	}
 	if !opts.NoConnect {
 		if _, err := c.connect(false); err != nil {
@@ -54,24 +54,23 @@ func newConn(opts *ClientOptions, auth *Auth) (*Conn, error) {
 	return c, nil
 }
 
-func (c *Conn) dial(proto string, u *url.URL) (proto.Conn, error) {
+func (c *Connection) dial(proto string, u *url.URL) (proto.Conn, error) {
 	if c.opts.Dial != nil {
 		return c.opts.Dial(proto, u)
 	}
 	return ablyutil.DialWebsocket(proto, u)
 }
 
-// Connect is used to connect to Ably servers manually, when the client owning
-// the connection was created with NoConnect option. The connect request is
-// being processed on a separate goroutine.
-//
-// If client is already connected, this method is a nop.
-// If connecting fail due to authorization error, the returned error value
-// is non-nil.
-// If authorization succeeds, the returned Result value can be used to wait
-// until connection confirmation is received from a server.
-func (c *Conn) Connect() (Result, error) {
-	return c.connect(true)
+// Connect attempts to move the connection to the CONNECTED state, if it
+// can and if it isn't already.
+func (c *Connection) Connect() {
+	c.connect(false)
+}
+
+// Close attempts to move the connection to the CLOSED state, if it can and if
+// it isn't already.
+func (c *Connection) Close() {
+	c.close()
 }
 
 var connectResultStates = []StateEnum{
@@ -80,7 +79,7 @@ var connectResultStates = []StateEnum{
 	StateConnDisconnected,
 }
 
-func (c *Conn) connect(result bool) (Result, error) {
+func (c *Connection) connect(result bool) (Result, error) {
 	c.state.Lock()
 	defer c.state.Unlock()
 	if c.isActive() {
@@ -130,47 +129,42 @@ func (c *Conn) connect(result bool) (Result, error) {
 	return res, nil
 }
 
-// Close initiates closing sequence for the connection; it waits until the
-// operation is complete.
-//
-// If connection is already closed, this method is a nop.
-func (c *Conn) Close() error {
-	err := wait(c.close())
-	if c.conn != nil {
-		c.conn.Close()
-	}
-	if err != nil {
-		return c.state.syncSet(StateConnFailed, err)
-	}
-	return nil
-}
-
 var closeResultStates = []StateEnum{
 	StateConnClosed, // expected state
 	StateConnFailed,
 	StateConnDisconnected,
 }
 
-func (c *Conn) close() (Result, error) {
+func (c *Connection) close() {
 	c.state.Lock()
 	defer c.state.Unlock()
 	switch c.state.current {
-	case StateConnClosing, StateConnClosed:
-		return nopResult, nil
-	case StateConnInitialized, StateConnFailed, StateConnDisconnected:
-		return nil, stateError(c.state.current, errCloseInactive)
+	case
+		StateConnClosing,
+		StateConnClosed,
+		StateConnInitialized,
+		StateConnFailed,
+		StateConnDisconnected:
+
+		return
 	}
-	res := c.state.listenResult(closeResultStates...)
 	c.state.set(StateConnClosing, nil)
 	msg := &proto.ProtocolMessage{Action: proto.ActionClose}
 	c.updateSerial(msg, nil)
-	return res, c.conn.Send(msg)
+
+	// TODO: handle error. If you can't send a message, the fail-fast way to
+	// deal with it is to discard the WebSocket and perform a normal
+	// reconnection. We could also have a retry loop, but in any case, it should
+	// be dealt with centrally, so Send shouldn't return the error but handle
+	// it in some way. The caller isn't responsible for recovering from realtime
+	// connection transient errors.
+	_ = c.conn.Send(msg)
 }
 
 // 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.
-func (c *Conn) ID() string {
+func (c *Connection) ID() string {
 	c.state.Lock()
 	defer c.state.Unlock()
 	return c.id
@@ -179,7 +173,7 @@ func (c *Conn) ID() string {
 // Key gives unique key string obtained from Ably upon successful connection.
 // The key may change due to reconnection and recovery; on every received
 // StatConnConnected event previously obtained Key is no longer valid.
-func (c *Conn) Key() string {
+func (c *Connection) Key() string {
 	c.state.Lock()
 	defer c.state.Unlock()
 	return c.details.ConnectionKey
@@ -190,13 +184,13 @@ func (c *Conn) Key() string {
 //
 // Ping returns non-nil error without any attempt of communication with Ably
 // if the connection state is StateConnClosed or StateConnFailed.
-func (c *Conn) Ping() (ping, pong time.Duration, err error) {
+func (c *Connection) Ping() (ping, pong time.Duration, err error) {
 	return 0, 0, errors.New("TODO")
 }
 
-// Reason gives last known error that caused connection transit to
+// ErrorReason gives last known error that caused connection transit to
 // StateConnFailed state.
-func (c *Conn) Reason() error {
+func (c *Connection) ErrorReason() error {
 	c.state.Lock()
 	defer c.state.Unlock()
 	return c.state.err
@@ -204,40 +198,90 @@ func (c *Conn) Reason() error {
 
 // Serial gives serial number of a message received most recently. Last known
 // serial number is used when recovering connection state.
-func (c *Conn) Serial() int64 {
+func (c *Connection) Serial() int64 {
 	c.state.Lock()
 	defer c.state.Unlock()
 	return c.serial
 }
 
 // State returns current state of the connection.
-func (c *Conn) State() StateEnum {
+func (c *Connection) State() StateEnum {
 	c.state.Lock()
 	defer c.state.Unlock()
 	return c.state.current
 }
 
-// On relays request connection states to the given channel; on state transition
+// 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 *Conn) On(ch chan<- State, states ...StateEnum) {
+func (c *Connection) onState(ch chan<- State, states ...StateEnum) {
 	c.state.on(ch, states...)
 }
 
-// Off removes c from listening on the given connection state transitions.
+// 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) {
+		handle(change.(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) {
+		handle(change.(ConnectionStateChange))
+	})
+}
+
+// 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) {
+		handle(change.(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) {
+		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 *Conn) Off(ch chan<- State, states ...StateEnum) {
+func (c *Connection) offState(ch chan<- State, states ...StateEnum) {
 	c.state.off(ch, states...)
 }
 
-func (c *Conn) updateSerial(msg *proto.ProtocolMessage, listen chan<- error) {
+// 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)
+}
+
+// Off deregisters all event handlers.
+//
+// See package-level documentation on Event Emitter for details.
+func (c *Connection) OffAll() {
+	c.state.eventEmitter.OffAll()
+}
+
+func (c *Connection) updateSerial(msg *proto.ProtocolMessage, listen chan<- error) {
 	const maxint64 = 1<<63 - 1
 	msg.MsgSerial = c.msgSerial
 	c.msgSerial = (c.msgSerial + 1) % maxint64
@@ -246,7 +290,7 @@ func (c *Conn) updateSerial(msg *proto.ProtocolMessage, listen chan<- error) {
 	}
 }
 
-func (c *Conn) send(msg *proto.ProtocolMessage, listen chan<- error) error {
+func (c *Connection) send(msg *proto.ProtocolMessage, listen chan<- error) error {
 	c.state.Lock()
 	switch state := c.state.current; state {
 	case StateConnInitialized, StateConnConnecting, StateConnDisconnected:
@@ -276,7 +320,7 @@ func (c *Conn) send(msg *proto.ProtocolMessage, listen chan<- error) error {
 //
 // If both user was not authenticated with a wildcard ClientID and the one
 // being sent does not match it, the method return non-nil error.
-func (c *Conn) verifyAndUpdateMessages(msg *proto.ProtocolMessage) (err error) {
+func (c *Connection) verifyAndUpdateMessages(msg *proto.ProtocolMessage) (err error) {
 	clientID := c.auth.clientIDForCheck()
 	connectionID := c.id
 	switch msg.Action {
@@ -306,26 +350,26 @@ func (c *Conn) verifyAndUpdateMessages(msg *proto.ProtocolMessage) (err error) {
 	return nil
 }
 
-func (c *Conn) isActive() bool {
+func (c *Connection) isActive() bool {
 	return c.state.current == StateConnConnecting || c.state.current == StateConnConnected
 }
 
-func (c *Conn) lockIsActive() bool {
+func (c *Connection) lockIsActive() bool {
 	c.state.Lock()
 	defer c.state.Unlock()
 	return c.isActive()
 }
 
-func (c *Conn) setConn(conn proto.Conn) {
+func (c *Connection) setConn(conn proto.Conn) {
 	c.conn = conn
 	go c.eventloop()
 }
 
-func (c *Conn) logger() *LoggerOptions {
+func (c *Connection) logger() *LoggerOptions {
 	return c.auth.logger()
 }
 
-func (c *Conn) eventloop() {
+func (c *Connection) eventloop() {
 	for {
 		msg, err := c.conn.Receive()
 		if err != nil {
@@ -363,6 +407,9 @@ func (c *Conn) eventloop() {
 			c.state.set(StateConnFailed, newErrorProto(msg.Error))
 			c.state.Unlock()
 			c.queue.Fail(newErrorProto(msg.Error))
+			if c.conn != nil {
+				c.conn.Close()
+			}
 		case proto.ActionConnected:
 			c.auth.updateClientID(msg.ConnectionDetails.ClientID)
 			c.state.Lock()
@@ -387,6 +434,9 @@ func (c *Conn) eventloop() {
 			c.id = ""
 			c.state.set(StateConnClosed, nil)
 			c.state.Unlock()
+			if c.conn != nil {
+				c.conn.Close()
+			}
 		default:
 			c.msgCh <- msg
 		}
diff --git a/ably/rest_client.go b/ably/rest_client.go
index 72d79c0..f94da37 100644
--- a/ably/rest_client.go
+++ b/ably/rest_client.go
@@ -135,7 +135,7 @@ type RestClient struct {
 
 func NewRestClient(opts *ClientOptions) (*RestClient, error) {
 	if opts == nil {
-		panic("called NewRealtimeClient with nil ClientOptions")
+		panic("called NewRealtime with nil ClientOptions")
 	}
 	c := &RestClient{
 		opts: *opts,
@@ -211,11 +211,11 @@ func (c *RestClient) Request(method string, path string, params *PaginateParams,
 			})
 		}, c.logger())
 	default:
-		return nil, &proto.ErrorInfo{
+		return nil, newErrorProto(&proto.ErrorInfo{
 			Message:    fmt.Sprintf("%s method is not supported", method),
 			Code:       ErrMethodNotAllowed,
 			StatusCode: http.StatusMethodNotAllowed,
-		}
+		})
 	}
 }
 
@@ -327,7 +327,7 @@ func (c *RestClient) doWithHandle(r *Request, handle func(*http.Response, interf
 	}
 	resp, err = handle(resp, r.Out)
 	if err != nil {
-		if e, ok := err.(*Error); ok {
+		if e, ok := err.(*ErrorInfo); ok {
 			if canFallBack(e.StatusCode) &&
 				(c.opts.FallbackHostsUseDefault ||
 					strings.HasPrefix(req.URL.Host, defaultOptions.RestHost) ||
@@ -376,7 +376,7 @@ func (c *RestClient) doWithHandle(r *Request, handle func(*http.Response, interf
 							if iteration == maxLimit-1 {
 								return nil, err
 							}
-							if ev, ok := err.(*Error); ok {
+							if ev, ok := err.(*ErrorInfo); ok {
 								if canFallBack(ev.StatusCode) {
 									iteration++
 									continue
diff --git a/ably/state.go b/ably/state.go
index 25e96e0..7a0ed77 100644
--- a/ably/state.go
+++ b/ably/state.go
@@ -141,37 +141,33 @@ var stateMasks = map[StateType]StateEnum{
 }
 
 var (
-	errClosed         = newErrorf(10000, "Connection closed by client")
 	errDisconnected   = newErrorf(80003, "Connection temporarily unavailable")
 	errSuspended      = newErrorf(80002, "Connection unavailable")
 	errFailed         = newErrorf(80000, "Connection failed")
 	errNeverConnected = newErrorf(80002, "Unable to establish connection")
 )
 
-var stateErrors = map[StateEnum]Error{
+var stateErrors = map[StateEnum]ErrorInfo{
 	StateConnInitialized:  *errNeverConnected,
-	StateConnClosed:       *errClosed,
-	StateConnClosing:      *errClosed,
 	StateConnDisconnected: *errDisconnected,
 	StateConnFailed:       *errFailed,
 	StateConnSuspended:    *errSuspended,
-	StateChanClosed:       *errClosed,
 	StateChanFailed:       *errFailed,
 }
 
-func stateError(state StateEnum, err error) error {
+func stateError(state StateEnum, err error) *ErrorInfo {
 	// Set default error information associated with the target state.
-	e, ok := err.(*Error)
+	e, ok := err.(*ErrorInfo)
 	if ok {
 		return e
 	}
 	if e, ok := stateErrors[state]; ok {
 		if err != nil {
-			e.Err = err
+			e.err = err
 		}
 		err = &e
 	}
-	return err
+	return newError(0, err)
 }
 
 // State describes a single state transition of either realtime connection or channel
@@ -183,10 +179,10 @@ func stateError(state StateEnum, err error) error {
 // 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     error     // 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
+	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 {
@@ -194,10 +190,12 @@ type stateEmitter struct {
 	channel   string
 	listeners map[StateEnum]map[chan<- State]struct{}
 	onetime   map[StateEnum]map[chan<- State]struct{}
-	err       error
+	err       *ErrorInfo
 	current   StateEnum
 	typ       StateType
 	logger    *LoggerOptions
+
+	eventEmitter *eventEmitter
 }
 
 func newStateEmitter(typ StateType, startState StateEnum, channel string, log *LoggerOptions) *stateEmitter {
@@ -211,14 +209,17 @@ func newStateEmitter(typ StateType, startState StateEnum, channel string, log *L
 		current:   startState,
 		typ:       typ,
 		logger:    log,
+
+		eventEmitter: newEventEmitter(log),
 	}
 }
 
 func (s *stateEmitter) set(state StateEnum, err error) error {
-	doemit := s.current != state
+	previous := s.current
+	changed := s.current != state
 	s.current = state
 	s.err = stateError(state, err)
-	if doemit {
+	if changed {
 		s.emit(State{
 			Channel: s.channel,
 			Err:     s.err,
@@ -226,6 +227,22 @@ func (s *stateEmitter) set(state StateEnum, err error) error {
 			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
 }
 
@@ -422,10 +439,10 @@ type msgch struct {
 type msgQueue struct {
 	mtx   sync.Mutex
 	queue []msgch
-	conn  *Conn
+	conn  *Connection
 }
 
-func newMsgQueue(conn *Conn) *msgQueue {
+func newMsgQueue(conn *Connection) *msgQueue {
 	return &msgQueue{
 		conn: conn,
 	}
@@ -523,12 +540,91 @@ func (res *stateResult) Wait() error {
 			if state.Type == StateConn {
 				code = 50002
 			}
-			res.err = &Error{
-				Code: code,
-				Err:  fmt.Errorf("failed %s state: %s", state.Type, state.State),
-			}
+			res.err = newError(code, fmt.Errorf("failed %s state: %s", state.Type, state.State))
 		}
 		res.listen = nil
 	}
 	return res.err
 }
+
+// A ConnectionState identifies the state of an Ably realtime connection.
+type ConnectionState struct {
+	name string
+}
+
+var (
+	ConnectionStateInitialized  = ConnectionState{name: "INITIALIZED"}
+	ConnectionStateConnecting   = ConnectionState{name: "CONNECTING"}
+	ConnectionStateConnected    = ConnectionState{name: "CONNECTED"}
+	ConnectionStateDisconnected = ConnectionState{name: "DISCONNECTED"}
+	ConnectionStateSuspended    = ConnectionState{name: "SUSPENDED"}
+	ConnectionStateClosing      = ConnectionState{name: "CLOSING"}
+	ConnectionStateClosed       = ConnectionState{name: "CLOSED"}
+	ConnectionStateFailed       = ConnectionState{name: "FAILED"}
+)
+
+func (e ConnectionState) String() string {
+	return e.name
+}
+
+// A ConnectionEvent identifies an event in the lifetime of an Ably realtime
+// connection.
+type ConnectionEvent struct {
+	name string
+}
+
+func (ConnectionEvent) isEmitterEvent() {}
+
+var (
+	ConnectionEventInitialized  = ConnectionEvent(ConnectionStateInitialized)
+	ConnectionEventConnecting   = ConnectionEvent(ConnectionStateConnecting)
+	ConnectionEventConnected    = ConnectionEvent(ConnectionStateConnected)
+	ConnectionEventDisconnected = ConnectionEvent(ConnectionStateDisconnected)
+	ConnectionEventSuspended    = ConnectionEvent(ConnectionStateSuspended)
+	ConnectionEventClosing      = ConnectionEvent(ConnectionStateClosing)
+	ConnectionEventClosed       = ConnectionEvent(ConnectionStateClosed)
+	ConnectionEventFailed       = ConnectionEvent(ConnectionStateFailed)
+	ConnectionEventUpdated      = ConnectionEvent{name: "UPDATED"}
+)
+
+func (e ConnectionEvent) String() string {
+	return e.name
+}
+
+// A ConnectionStateChange is the data associated with a ConnectionEvent.
+//
+// If the Event is a ConnectionEventUpdated, Current and Previous are the
+// the same. Otherwise, the event is a state transition from Previous to
+// Current.
+type ConnectionStateChange struct {
+	Current  ConnectionState
+	Event    ConnectionEvent
+	Previous ConnectionState
+	// Reason, if any, is an error that caused the state change.
+	Reason *ErrorInfo
+}
+
+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))
+	}
+}


Test patch:
diff --git a/ably/ably_test.go b/ably/ably_test.go
index 96a1ca1..de01bd1 100644
--- a/ably/ably_test.go
+++ b/ably/ably_test.go
@@ -34,14 +34,13 @@ func safeclose(t *testing.T, closers ...io.Closer) {
 		for _, err := range errors {
 			t.Logf("safeclose %d: failed to close %T: %s", err.i, err.c, err.err)
 		}
-		t.FailNow()
 	}
 }
 
 func checkError(code int, err error) error {
-	switch e, ok := err.(*ably.Error); {
+	switch e, ok := err.(*ably.ErrorInfo); {
 	case !ok:
-		return fmt.Errorf("want err to be *ably.Error; was %T: %v", err, err)
+		return fmt.Errorf("want err to be *ably.ErrorInfo; was %T: %v", err, err)
 	case e.Code != code:
 		return fmt.Errorf("want e.Code=%d; got %d: %s", code, e.Code, err)
 	default:
diff --git a/ably/ablytest/ablytest.go b/ably/ablytest/ablytest.go
index 1fa84ec..6243952 100644
--- a/ably/ablytest/ablytest.go
+++ b/ably/ablytest/ablytest.go
@@ -65,11 +65,7 @@ func encode(typ string, in interface{}) ([]byte, error) {
 	case "text/plain":
 		return []byte(fmt.Sprintf("%v", in)), nil
 	default:
-		return nil, &ably.Error{
-			StatusCode: 400,
-			Code:       40000,
-			Err:        fmt.Errorf("encoding error: unrecognized Content-Type: %q", typ),
-		}
+		return nil, fmt.Errorf("encoding error: unrecognized Content-Type: %q", typ)
 	}
 }
 
diff --git a/ably/ablytest/fmt.go b/ably/ablytest/fmt.go
new file mode 100644
index 0000000..bc6e30f
--- /dev/null
+++ b/ably/ablytest/fmt.go
@@ -0,0 +1,25 @@
+package ablytest
+
+import (
+	"fmt"
+	"testing"
+)
+
+// A FmtFunc is a non-failing function analogous to fmt.Printf.
+type FmtFunc func(format string, args ...interface{})
+
+// Wrap wraps a FmtFunc in another. The wrapper FmtFunc first uses the fixed
+// format and args, plus the format it's called with, to create a new format
+// string by calling fmt.Sprintf. Then, it calls the wrapped FmtFunc with this
+// format string and the args the wrapper is called with.
+//
+// It's useful to have some fixed context on everything that is formatted with
+// a FmtFunc, e. g. test scope context.
+func (f FmtFunc) Wrap(t *testing.T, format string, args ...interface{}) FmtFunc {
+	return func(wrappedFormat string, wrappedArgs ...interface{}) {
+		if t != nil {
+			t.Helper()
+		}
+		f(fmt.Sprintf(format, append(args, wrappedFormat)...), wrappedArgs...)
+	}
+}
diff --git a/ably/ablytest/fmt_test.go b/ably/ablytest/fmt_test.go
new file mode 100644
index 0000000..00f0f90
--- /dev/null
+++ b/ably/ablytest/fmt_test.go
@@ -0,0 +1,25 @@
+package ablytest_test
+
+import (
+	"fmt"
+
+	"github.com/ably/ably-go/ably/ablytest"
+)
+
+func ExampleFmtFunc_Wrap() {
+	println := ablytest.FmtFunc(func(format string, args ...interface{}) {
+		fmt.Println(fmt.Sprintf(format, args...))
+	})
+
+	id := 42
+	wrapped := println.Wrap(nil, "for ID %d: %s", id)
+
+	wrapped("everything's OK")
+
+	errMsg := "all hell broke loose"
+	wrapped("something failed: %s", errMsg)
+
+	// Output:
+	// for ID 42: everything's OK
+	// for ID 42: something failed: all hell broke loose
+}
diff --git a/ably/ablytest/logger.go b/ably/ablytest/logger.go
new file mode 100644
index 0000000..ccaa0a8
--- /dev/null
+++ b/ably/ablytest/logger.go
@@ -0,0 +1,39 @@
+package ablytest
+
+import (
+	"fmt"
+
+	"github.com/ably/ably-go/ably"
+)
+
+type LogMessage struct {
+	Level   ably.LogLevel
+	Message string
+}
+
+func NewLogger(messages chan<- LogMessage) ably.Logger {
+	return testLogger{messages: messages}
+}
+
+type testLogger struct {
+	messages chan<- LogMessage
+}
+
+func (l testLogger) Print(level ably.LogLevel, v ...interface{}) {
+	l.Printf(level, "%s", v...)
+}
+
+func (l testLogger) Printf(level ably.LogLevel, format string, v ...interface{}) {
+	l.messages <- LogMessage{
+		Level:   level,
+		Message: fmt.Sprintf(format, v...),
+	}
+}
+
+var DiscardLogger ably.Logger = discardLogger{}
+
+type discardLogger struct{}
+
+func (discardLogger) Print(level ably.LogLevel, v ...interface{}) {}
+
+func (discardLogger) Printf(level ably.LogLevel, format string, v ...interface{}) {}
diff --git a/ably/ablytest/recorders.go b/ably/ablytest/recorders.go
index e3005f4..bb978ca 100644
--- a/ably/ablytest/recorders.go
+++ b/ably/ablytest/recorders.go
@@ -2,6 +2,8 @@ package ablytest
 
 import (
 	"bytes"
+	"context"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -240,8 +242,8 @@ func (rec *StateRecorder) States() []ably.StateEnum {
 // 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() []error {
-	var errors []error
+func (rec *StateRecorder) Errors() []*ably.ErrorInfo {
+	var errors []*ably.ErrorInfo
 	rec.mtx.Lock()
 	defer rec.mtx.Unlock()
 	for _, state := range rec.states {
@@ -450,3 +452,158 @@ func (hr *HostRecorder) addHost(host string) {
 func body(p []byte) io.ReadCloser {
 	return ioutil.NopCloser(bytes.NewReader(p))
 }
+
+// SetFakeDisconnect wraps the Dial option so that calling the returned
+// disconnect function forcibly closes the connection to the server and fakes
+// a DISCONNECT message from the server to the client.
+//
+// Only a single connection gets the disconnect signal, so don't reuse the
+// ClientOptions.
+func SetFakeDisconnect(opts *ably.ClientOptions) (disconnect func() error) {
+	dial := opts.Dial
+	if dial == nil {
+		dial = func(proto string, url *url.URL) (proto.Conn, error) {
+			return ablyutil.DialWebsocket(proto, url)
+		}
+	}
+
+	disconnectReq := make(chan chan<- error, 1)
+
+	opts.Dial = func(proto string, url *url.URL) (proto.Conn, error) {
+		conn, err := dial(proto, url)
+		if err != nil {
+			return nil, err
+		}
+
+		return connWithFakeDisconnect{
+			conn:          conn,
+			disconnectReq: disconnectReq,
+			closed:        make(chan struct{}),
+		}, nil
+	}
+
+	return func() error {
+		err := make(chan error)
+		disconnectReq <- err
+		return <-err
+	}
+}
+
+type connWithFakeDisconnect struct {
+	conn          proto.Conn
+	disconnectReq <-chan chan<- error
+	closed        chan struct{}
+}
+
+func (c connWithFakeDisconnect) Send(m *proto.ProtocolMessage) error {
+	return c.conn.Send(m)
+}
+
+func (c connWithFakeDisconnect) Receive() (*proto.ProtocolMessage, error) {
+	// Call the real Receive while waiting for a fake disconnection request.
+	// The first wins. After a disconnection request, the connection is closed,
+	// the ongoing real Receive is ignored and subsequent calls to Receive
+	// fail.
+
+	select {
+	case <-c.closed:
+		return nil, errors.New("called Receive on closed connection")
+	default:
+	}
+
+	type receiveResult struct {
+		m   *proto.ProtocolMessage
+		err error
+	}
+	realReceive := make(chan receiveResult, 1)
+	go func() {
+		m, err := c.conn.Receive()
+		select {
+		case <-c.closed:
+		case realReceive <- receiveResult{m: m, err: err}:
+		}
+	}()
+
+	select {
+	case r := <-realReceive:
+		return r.m, r.err
+
+	case errCh := <-c.disconnectReq:
+		err := c.Close()
+		errCh <- err
+		if err != nil {
+			return nil, err
+		}
+
+		return &proto.ProtocolMessage{
+			Action: proto.ActionDisconnected,
+			Error:  &proto.ErrorInfo{Message: "fake disconnection"},
+		}, nil
+	}
+}
+
+func (c connWithFakeDisconnect) Close() error {
+	select {
+	case <-c.closed:
+		// Already closed.
+		return nil
+	default:
+	}
+
+	close(c.closed)
+	return c.conn.Close()
+}
+
+// FullRealtimeCloser returns an io.Closer that, on Close, calls Close on the
+// Realtime instance and waits for its effects.
+func FullRealtimeCloser(c *ably.Realtime) io.Closer {
+	return realtimeIOCloser{c: c}
+}
+
+type realtimeIOCloser struct {
+	c *ably.Realtime
+}
+
+func (c realtimeIOCloser) Close() error {
+	switch c.c.Connection.State() {
+	case
+		ably.StateConnInitialized,
+		ably.StateConnClosed,
+		ably.StateConnFailed:
+
+		return c.c.Connection.ErrorReason()
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), Timeout)
+	defer cancel()
+
+	errCh := make(chan error, 1)
+
+	off := make(chan func(), 1)
+	off <- c.c.Connection.OnAll(func(c ably.ConnectionStateChange) {
+		switch c.Current {
+		default:
+			return
+		case
+			ably.ConnectionStateClosed,
+			ably.ConnectionStateFailed:
+		}
+
+		(<-off)()
+
+		var err error
+		if c.Reason != nil {
+			err = *c.Reason
+		}
+		errCh <- err
+	})
+
+	c.c.Close()
+
+	select {
+	case err := <-errCh:
+		return err
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
diff --git a/ably/ablytest/resultgroup.go b/ably/ablytest/resultgroup.go
index 44698c1..0b8495f 100644
--- a/ably/ablytest/resultgroup.go
+++ b/ably/ablytest/resultgroup.go
@@ -2,6 +2,7 @@ package ablytest
 
 import (
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 
@@ -80,3 +81,44 @@ func (rg *ResultGroup) Wait() error {
 		return err
 	}
 }
+
+func ConnWaiter(client *ably.Realtime, do func(), expectedEvent ...ably.ConnectionEvent) ably.Result {
+	change := make(chan ably.ConnectionStateChange, 1)
+	off := client.Connection.OnAll(func(ev ably.ConnectionStateChange) {
+		change <- ev
+	})
+	do()
+	return ResultFunc(func() error {
+		defer off()
+		timer := time.After(Timeout)
+
+		for {
+			select {
+			case <-timer:
+				return fmt.Errorf("timeout waiting for event %v", expectedEvent)
+
+			case change := <-change:
+				if change.Reason != nil {
+					return change.Reason
+				}
+
+				if len(expectedEvent) > 0 {
+					for _, ev := range expectedEvent {
+						if ev == change.Event {
+							return nil
+						}
+					}
+					continue
+				}
+
+				return nil
+			}
+		}
+	})
+}
+
+type ResultFunc func() error
+
+func (f ResultFunc) Wait() error {
+	return f()
+}
diff --git a/ably/ablytest/sandbox.go b/ably/ablytest/sandbox.go
index 62ec976..81d0a40 100644
--- a/ably/ablytest/sandbox.go
+++ b/ably/ablytest/sandbox.go
@@ -101,9 +101,9 @@ type Sandbox struct {
 	client *http.Client
 }
 
-func NewRealtimeClient(opts *ably.ClientOptions) (*Sandbox, *ably.RealtimeClient) {
+func NewRealtime(opts *ably.ClientOptions) (*Sandbox, *ably.Realtime) {
 	app := MustSandbox(nil)
-	client, err := ably.NewRealtimeClient(app.Options(opts))
+	client, err := ably.NewRealtime(app.Options(opts))
 	if err != nil {
 		panic(nonil(err, app.Close()))
 	}
@@ -185,10 +185,10 @@ func (app *Sandbox) Close() error {
 	return nil
 }
 
-func (app *Sandbox) NewRealtimeClient(opts ...*ably.ClientOptions) *ably.RealtimeClient {
-	client, err := ably.NewRealtimeClient(app.Options(opts...))
+func (app *Sandbox) NewRealtime(opts ...*ably.ClientOptions) *ably.Realtime {
+	client, err := ably.NewRealtime(app.Options(opts...))
 	if err != nil {
-		panic("ably.NewRealtimeClient failed: " + err.Error())
+		panic("ably.NewRealtime failed: " + err.Error())
 	}
 	return client
 }
diff --git a/ably/ablytest/timeout.go b/ably/ablytest/timeout.go
new file mode 100644
index 0000000..aec6282
--- /dev/null
+++ b/ably/ablytest/timeout.go
@@ -0,0 +1,65 @@
+package ablytest
+
+import (
+	"reflect"
+	"testing"
+	"time"
+)
+
+var Instantly = Before(10 * time.Millisecond)
+
+var Soon = Before(Timeout)
+
+// Before returns a WithTimeout with the given timeout duration.
+func Before(d time.Duration) WithTimeout {
+	return WithTimeout{before: d}
+}
+
+// WithTimeout configures test helpers with a timeout.
+type WithTimeout struct {
+	before time.Duration
+}
+
+// Recv asserts that a value is received through channel from before the
+// timeout. If it isn't, the fail function is called.
+//
+// If into is non-nil, it must be a pointer to a variable of the same type as
+// from's element type and it will be set to the received value, if any.
+//
+// It returns the second, boolean value returned by the receive operation,
+// or false if the operation times out.
+func (wt WithTimeout) Recv(t *testing.T, into interface{}, from interface{}, fail func(fmt string, args ...interface{})) (ok bool) {
+	t.Helper()
+	ok, timeout := wt.recv(into, from)
+	if timeout {
+		fail("timed out waiting for channel receive")
+	}
+	return ok
+}
+
+// NoRecv is like Recv, except it asserts no value is received.
+func (wt WithTimeout) NoRecv(t *testing.T, into interface{}, from interface{}, fail func(fmt string, args ...interface{})) (ok bool) {
+	t.Helper()
+	if into == nil {
+		into = &into
+	}
+	ok, timeout := wt.recv(into, from)
+	if !timeout {
+		fail("unexpectedly received in channel: %v", into)
+	}
+	return ok
+}
+
+func (wt WithTimeout) recv(into interface{}, from interface{}) (ok, timeout bool) {
+	chosen, recv, ok := reflect.Select([]reflect.SelectCase{{
+		Dir:  reflect.SelectRecv,
+		Chan: reflect.ValueOf(from),
+	}, {
+		Dir:  reflect.SelectRecv,
+		Chan: reflect.ValueOf(time.After(time.Duration(wt.before))),
+	}})
+	if chosen == 0 && ok && into != nil {
+		reflect.ValueOf(into).Elem().Set(recv)
+	}
+	return ok, chosen == 1
+}
diff --git a/ably/auth_test.go b/ably/auth_test.go
index 03e660a..5104594 100644
--- a/ably/auth_test.go
+++ b/ably/auth_test.go
@@ -485,7 +485,7 @@ func TestAuth_RequestToken(t *testing.T) {
 		optsURL.AuthOptions = ably.AuthOptions{Token: tokURL.Token}
 		c, err := ably.NewRestClient(optsURL)
 		if err != nil {
-			t.Errorf("NewRealtimeClient()=%v", err)
+			t.Errorf("NewRealtime()=%v", err)
 			continue
 		}
 		if _, err = c.Stats(single()); err != nil {
@@ -503,7 +503,7 @@ func TestAuth_ClientID_Error(t *testing.T) {
 			UseTokenAuth: true,
 		},
 	}
-	_, err := ably.NewRealtimeClient(opts)
+	_, err := ably.NewRealtime(opts)
 	if err := checkError(40102, err); err != nil {
 		t.Fatal(err)
 	}
@@ -585,8 +585,8 @@ func TestAuth_RequestToken_PublishClientID(t *testing.T) {
 		if i == 4 {
 			copts.ClientID = cas.clientID
 		}
-		client := app.NewRealtimeClient(copts)
-		defer safeclose(t, client)
+		client := app.NewRealtime(copts)
+		defer safeclose(t, ablytest.FullRealtimeCloser(client))
 		if id := client.Auth.ClientID(); id != cas.clientID {
 			t.Errorf("%d: want ClientID to be %q; got %s", i, cas.clientID, id)
 			continue
@@ -649,7 +649,7 @@ func TestAuth_ClientID(t *testing.T) {
 		Dial:      ablytest.MessagePipe(in, out),
 		NoConnect: true,
 	}
-	client := app.NewRealtimeClient(opts) // no client.Close as the connection is mocked
+	client := app.NewRealtime(opts) // no client.Close as the connection is mocked
 
 	tok, err := client.Auth.RequestToken(params, nil)
 	if err != nil {
@@ -673,7 +673,7 @@ func TestAuth_ClientID(t *testing.T) {
 	if id := client.Auth.ClientID(); id != "" {
 		t.Fatalf("want clientID to be empty; got %q", id)
 	}
-	if err := ablytest.Wait(client.Connection.Connect()); err != nil {
+	if err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
 		t.Fatalf("Connect()=%v", err)
 	}
 	if id := client.Auth.ClientID(); id != connected.ConnectionDetails.ClientID {
@@ -696,24 +696,34 @@ func TestAuth_ClientID(t *testing.T) {
 	tokenExpiredAt := time.Now()
 
 	failed := make(chan ably.State, 1)
-	client.Connection.On(failed, ably.StateConnFailed)
+	client.Connection.OnState(failed, ably.StateConnFailed)
 
 	// Allow some extra time for the server to reject our token.
 	err = nil
 	tokenErrorDeadline := tokenExpiredAt.Add(5 * time.Second)
 	for err == nil && time.Now().Before(tokenErrorDeadline) {
 		time.Sleep(100 * time.Millisecond)
-		closed := &proto.ProtocolMessage{
-			Action: proto.ActionClosed,
-		}
-		in <- closed
-		if err := client.Close(); err != nil {
+
+		err = ablytest.Wait(ablytest.ConnWaiter(client, client.Close, ably.ConnectionEventClosing), nil)
+		if err != nil {
 			t.Fatalf("Close()=%v", err)
 		}
 
-		in <- connected
+		err = ablytest.Wait(ablytest.ConnWaiter(client, func() {
+			closed := &proto.ProtocolMessage{
+				Action: proto.ActionClosed,
+			}
+			in <- closed
+		}, ably.ConnectionEventClosed), nil)
+		if err != nil {
+			t.Fatalf("waiting for close: %v", err)
+		}
+
 		proxy.TokenQueue = append(proxy.TokenQueue, tok)
-		err = ablytest.Wait(client.Connection.Connect())
+		err = ablytest.Wait(ablytest.ConnWaiter(client, client.Connect,
+			ably.ConnectionEventConnected,
+			ably.ConnectionEventFailed,
+		), nil)
 	}
 	if err = checkError(40012, err); err != nil {
 		t.Fatal(err)
@@ -767,7 +777,7 @@ func TestAuth_CreateTokenRequest(t *testing.T) {
 			if err == nil {
 				ts.Fatal("expected an error")
 			}
-			e := err.(*ably.Error)
+			e := err.(*ably.ErrorInfo)
 			if e.Code != ably.ErrInvalidCredentials {
 				ts.Errorf("expected error code %d got %d", ably.ErrInvalidCredentials, e.Code)
 			}
@@ -778,7 +788,7 @@ func TestAuth_CreateTokenRequest(t *testing.T) {
 			if err == nil {
 				ts.Fatal("expected an error")
 			}
-			e = err.(*ably.Error)
+			e = err.(*ably.ErrorInfo)
 			if e.Code != ably.ErrIncompatibleCredentials {
 				ts.Errorf("expected error code %d got %d", ably.ErrIncompatibleCredentials, e.Code)
 			}
@@ -813,10 +823,10 @@ func TestAuth_RealtimeAccessToken(t *testing.T) {
 		Dial:      rec.Dial,
 	}
 	opts.UseTokenAuth = true
-	app, client := ablytest.NewRealtimeClient(opts)
+	app, client := ablytest.NewRealtime(opts)
 	defer safeclose(t, app)
 
-	if err := ablytest.Wait(client.Connection.Connect()); err != nil {
+	if err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
 		t.Fatalf("Connect()=%v", err)
 	}
 	if err := ablytest.Wait(client.Channels.Get("test").Publish("name", "value")); err != nil {
@@ -825,7 +835,7 @@ func TestAuth_RealtimeAccessToken(t *testing.T) {
 	if clientID := client.Auth.ClientID(); clientID != opts.ClientID {
 		t.Fatalf("want ClientID=%q; got %q", opts.ClientID, clientID)
 	}
-	if err := client.Close(); err != nil {
+	if err := ablytest.FullRealtimeCloser(client).Close(); err != nil {
 		t.Fatalf("Close()=%v", err)
 	}
 	urls := rec.URL()
diff --git a/ably/error_test.go b/ably/error_test.go
index a3ee1ff..97ad399 100644
--- a/ably/error_test.go
+++ b/ably/error_test.go
@@ -1,11 +1,13 @@
 package ably_test
 
 import (
+	"errors"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"net/url"
 	"strconv"
+	"strings"
 	"testing"
 
 	"github.com/ably/ably-go/ably"
@@ -18,11 +20,11 @@ func TestErrorResponseWithInvalidKey(t *testing.T) {
 	if e == nil {
 		t.Fatal("NewRestClient(): expected err != nil")
 	}
-	err, ok := e.(*ably.Error)
+	err, ok := e.(*ably.ErrorInfo)
 	assert.True(t, ok, fmt.Sprintf("want e be *ably.Error; was %T", e))
 	assert.Equal(t, 400, err.StatusCode, fmt.Sprintf("want StatusCode=400; got %d", err.StatusCode))
 	assert.Equal(t, 40005, err.Code, fmt.Sprintf("want Code=40005; got %d", err.Code))
-	assert.NotNil(t, err.Err)
+	assert.NotNil(t, err.Unwrap())
 }
 
 func TestIssue127ErrorResponse(t *testing.T) {
@@ -54,3 +56,69 @@ func TestIssue127ErrorResponse(t *testing.T) {
 	assert.NotNil(t, err)
 	assert.Contains(t, err.Error(), errMsg)
 }
+
+func TestErrorInfo(t *testing.T) {
+	t.Run("without an error code", func(ts *testing.T) {
+		e := &ably.ErrorInfo{
+			StatusCode: 401,
+		}
+		h := "help.ably.io"
+		if strings.Contains(e.Error(), h) {
+			ts.Errorf("expected error message not to contain %s", h)
+		}
+	})
+	t.Run("with an error code", func(ts *testing.T) {
+		e := &ably.ErrorInfo{
+			Code: 44444,
+		}
+		h := "https://help.ably.io/error/44444"
+		if !strings.Contains(e.Error(), h) {
+			ts.Errorf("expected error message %s  to contain %s", e.Error(), h)
+		}
+	})
+	t.Run("with an error code and an href attribute", func(ts *testing.T) {
+		href := "http://foo.bar.com/"
+		e := &ably.ErrorInfo{
+			Code: 44444,
+			HRef: href,
+		}
+		h := "https://help.ably.io/error/44444"
+		if strings.Contains(e.Error(), h) {
+			ts.Errorf("expected error message %s not to contain %s", e.Error(), h)
+		}
+		if !strings.Contains(e.Error(), href) {
+			ts.Errorf("expected error message %s  to contain %s", e.Error(), href)
+		}
+	})
+
+	t.Run("with an error code and a message with the same error URL", func(ts *testing.T) {
+		e := ably.NewErrorInfo(44444, errors.New("error https://help.ably.io/error/44444"))
+		h := "https://help.ably.io/error/44444"
+		if !strings.Contains(e.Error(), h) {
+			ts.Errorf("expected error message %s  to contain %s", e.Error(), h)
+		}
+		n := strings.Count(e.Error(), h)
+		if n != 1 {
+			ts.Errorf("expected 1 occupance of %s got %d", h, n)
+		}
+	})
+	t.Run("with an error code and a message with a different error URL", func(ts *testing.T) {
+		e := ably.NewErrorInfo(44444, errors.New("error https://help.ably.io/error/123123"))
+		h := "https://help.ably.io/error/44444"
+		if !strings.Contains(e.Error(), h) {
+			ts.Errorf("expected error message %s  to contain %s", e.Error(), h)
+		}
+		n := strings.Count(e.Error(), "help.ably.io")
+		if n != 2 {
+			ts.Errorf("expected 2 got %d", n)
+		}
+		n = strings.Count(e.Error(), "/123123")
+		if n != 1 {
+			ts.Errorf("expected 1 got %d", n)
+		}
+		n = strings.Count(e.Error(), "/44444")
+		if n != 1 {
+			ts.Errorf("expected 1 got %d", n)
+		}
+	})
+}
diff --git a/ably/event_emitter_spec_test.go b/ably/event_emitter_spec_test.go
new file mode 100644
index 0000000..c2bde8b
--- /dev/null
+++ b/ably/event_emitter_spec_test.go
@@ -0,0 +1,326 @@
+package ably_test
+
+import (
+	"fmt"
+	"math/rand"
+	"reflect"
+	"testing"
+
+	"github.com/ably/ably-go/ably"
+	"github.com/ably/ably-go/ably/ablytest"
+)
+
+func Test_RTE3_EventEmitter_On(t *testing.T) {
+	for _, tc := range []struct {
+		event      ably.EmitterString
+		data       ably.EmitterData
+		expectRecv []ably.EmitterString
+	}{{
+		"qux", ably.EmitterString("42"),
+		[]ably.EmitterString{"*"},
+	}, {
+		"foo", ably.EmitterString("42"),
+		[]ably.EmitterString{"*", "foo"},
+	}, {
+		"bar", ably.EmitterString("42"),
+		[]ably.EmitterString{"*", "bar", "bar"},
+	}} {
+		t.Run(fmt.Sprintf("event: %s, data: %v", tc.event, tc.data), func(t *testing.T) {
+			t.Parallel()
+
+			em := ably.NewEventEmitter(ablytest.DiscardLogger)
+
+			received := map[ably.EmitterString]chan ably.EmitterData{
+				"*":   make(chan ably.EmitterData, 999),
+				"foo": make(chan ably.EmitterData, 999),
+				"bar": make(chan ably.EmitterData, 999),
+			}
+
+			em.OnAll(func(v ably.EmitterData) {
+				received["*"] <- v
+			})
+
+			em.On(ably.EmitterString("foo"), func(v ably.EmitterData) {
+				received["foo"] <- v
+			})
+
+			em.On(ably.EmitterString("bar"), func(v ably.EmitterData) {
+				received["bar"] <- v
+			})
+
+			em.On(ably.EmitterString("bar"), func(v ably.EmitterData) {
+				received["bar"] <- v
+			})
+
+			em.Emit(tc.event, tc.data)
+
+			for _, name := range tc.expectRecv {
+				fatalf := ablytest.FmtFunc.Wrap(t.Fatalf, t, "received to %q: %s", name)
+
+				var recv ably.EmitterData
+				ablytest.Instantly.Recv(t, &recv, received[name], fatalf)
+
+				if !reflect.DeepEqual(tc.data, recv) {
+					fatalf("expected %q, got %q")
+				}
+			}
+
+			assertNoFurtherReceives(t, received)
+		})
+	}
+}
+
+func Test_RTE4_EventEmitter_Once(t *testing.T) {
+	for _, tc := range []struct {
+		event      ably.EmitterString
+		data       ably.EmitterData
+		expectRecv []ably.EmitterString
+	}{{
+		"qux", ably.EmitterString("42"),
+		[]ably.EmitterString{"*"},
+	}, {
+		"foo", ably.EmitterString("42"),
+		[]ably.EmitterString{"*", "foo"},
+	}, {
+		"bar", ably.EmitterString("42"),
+		[]ably.EmitterString{"*", "bar", "bar"},
+	}} {
+		t.Run(fmt.Sprintf("event: %s, data: %v", tc.event, tc.data), func(t *testing.T) {
+			t.Parallel()
+
+			em := ably.NewEventEmitter(ablytest.DiscardLogger)
+
+			received := map[ably.EmitterString]chan ably.EmitterData{
+				"*":   make(chan ably.EmitterData, 999),
+				"foo": make(chan ably.EmitterData, 999),
+				"bar": make(chan ably.EmitterData, 999),
+			}
+
+			em.Once(nil, func(v ably.EmitterData) {
+				received["*"] <- v
+			})
+
+			em.Once(ably.EmitterString("foo"), func(v ably.EmitterData) {
+				received["foo"] <- v
+			})
+
+			em.Once(ably.EmitterString("bar"), func(v ably.EmitterData) {
+				received["bar"] <- v
+			})
+
+			em.Once(ably.EmitterString("bar"), func(v ably.EmitterData) {
+				received["bar"] <- v
+			})
+
+			em.Emit(tc.event, tc.data)
+
+			for _, name := range tc.expectRecv {
+				fatalf := ablytest.FmtFunc.Wrap(t.Fatalf, t, "received to %q: %s", name)
+
+				var recv ably.EmitterData
+				ablytest.Instantly.Recv(t, &recv, received[name], fatalf)
+
+				if !reflect.DeepEqual(tc.data, recv) {
+					fatalf("expected %q, got %q")
+				}
+			}
+
+			assertNoFurtherReceives(t, received)
+
+			for name, ch := range received {
+				fatalf := ablytest.FmtFunc.Wrap(t.Fatalf, t, "received to %q: %s", name)
+				ablytest.Instantly.NoRecv(t, nil, ch, fatalf)
+			}
+
+			// Emit again; check that nothing is received because the listeners
+			// got removed after the first trigger.
+
+			em.Emit(tc.event, tc.data)
+
+			assertNoFurtherReceives(t, received)
+		})
+	}
+}
+
+func Test_RTE5_EventEmitter_Off(t *testing.T) {
+	setUp := func() (
+		received map[ably.EmitterString]chan ably.EmitterData,
+		em *ably.EventEmitter,
+		allOff, fooOff, barOff1, barOff2 func(),
+	) {
+		received = map[ably.EmitterString]chan ably.EmitterData{
+			"*":   make(chan ably.EmitterData, 999),
+			"foo": make(chan ably.EmitterData, 999),
+			"bar": make(chan ably.EmitterData, 999),
+		}
+
+		em = ably.NewEventEmitter(ablytest.DiscardLogger)
+
+		allOff = em.OnAll(func(v ably.EmitterData) {
+			received["*"] <- v
+		})
+
+		fooOff = em.On(ably.EmitterString("foo"), func(v ably.EmitterData) {
+			received["foo"] <- v
+		})
+
+		barOff1 = em.On(ably.EmitterString("bar"), func(v ably.EmitterData) {
+			received["bar"] <- v
+		})
+
+		barOff2 = em.On(ably.EmitterString("bar"), func(v ably.EmitterData) {
+			received["bar"] <- v
+		})
+
+		return
+	}
+
+	t.Run("specific listener", func(t *testing.T) {
+		t.Parallel()
+
+		received, em, _, fooOff, barOff1, _ := setUp()
+		defer assertNoFurtherReceives(t, received)
+
+		barOff1()
+		em.Emit(ably.EmitterString("bar"), nil)
+
+		// The other listeners work.
+		ablytest.Instantly.Recv(t, nil, received["bar"], t.Fatalf)
+		ablytest.Instantly.Recv(t, nil, received["*"], t.Fatalf)
+
+		// Other specific listeners unaffected.
+		em.Emit(ably.EmitterString("foo"), nil)
+		ablytest.Instantly.Recv(t, nil, received["foo"], t.Fatalf)
+		ablytest.Instantly.Recv(t, nil, received["*"], t.Fatalf)
+
+		fooOff()
+		em.Emit(ably.EmitterString("foo"), nil)
+
+		// Only matching listener affected.
+		ablytest.Instantly.NoRecv(t, nil, received["foo"], t.Fatalf)
+		ablytest.Instantly.Recv(t, nil, received["*"], t.Fatalf)
+	})
+
+	t.Run("specific event", func(t *testing.T) {
+		t.Parallel()
+
+		received, em, _, _, _, _ := setUp()
+		defer assertNoFurtherReceives(t, received)
+
+		em.Off(ably.EmitterString("bar"))
+		em.Emit(ably.EmitterString("bar"), nil)
+
+		// Event-specific listeners affected.
+		ablytest.Instantly.NoRecv(t, nil, received["bar"], t.Fatalf)
+
+		// The other listeners work.
+		ablytest.Instantly.Recv(t, nil, received["*"], t.Fatalf)
+
+		// Other specific listeners unaffected.
+		em.Emit(ably.EmitterString("foo"), nil)
+		ablytest.Instantly.Recv(t, nil, received["foo"], t.Fatalf)
+		ablytest.Instantly.Recv(t, nil, received["*"], t.Fatalf)
+
+		em.Off(ably.EmitterString("foo"))
+		em.Emit(ably.EmitterString("foo"), nil)
+
+		// Only matching listener affected.
+		ablytest.Instantly.NoRecv(t, nil, received["foo"], t.Fatalf)
+		ablytest.Instantly.Recv(t, nil, received["*"], t.Fatalf)
+	})
+
+	t.Run("all", func(t *testing.T) {
+		t.Parallel()
+
+		received, em, _, _, _, _ := setUp()
+
+		em.Off(nil)
+
+		em.Emit(ably.EmitterString("foo"), nil)
+		em.Emit(ably.EmitterString("bar"), nil)
+		em.Emit(ably.EmitterString("qux"), nil)
+
+		// Nothing gets delivered.
+		assertNoFurtherReceives(t, received)
+	})
+}
+
+func Test_RTE6_EventEmitter_EmitPanic(t *testing.T) {
+	t.Parallel()
+
+	logs := make(chan ablytest.LogMessage, 999)
+	em := ably.NewEventEmitter(ablytest.NewLogger(logs))
+
+	shouldPanic := true
+	called := make(chan struct{}, 999)
+	em.OnAll(func(ably.EmitterData) {
+		if shouldPanic {
+			panic("aaah")
+		}
+		called <- struct{}{}
+	})
+
+	em.Emit(nil, nil)
+
+	ablytest.Instantly.Recv(t, nil, logs, t.Fatalf)
+	ablytest.Instantly.NoRecv(t, nil, called, t.Fatalf)
+
+	shouldPanic = false
+
+	// Panic didn't break anything.
+
+	em.Emit(nil, nil)
+
+	ablytest.Instantly.NoRecv(t, nil, logs, t.Fatalf)
+	ablytest.Instantly.Recv(t, nil, called, t.Fatalf)
+}
+
+func Test_RTE6a_EventEmitter_EmitToFixedListenersCollection(t *testing.T) {
+	t.Parallel()
+
+	// Give it a few tries; the order in which listeners are triggered isn't
+	// deterministic.
+	for i := 0; i < 10; i++ {
+		em := ably.NewEventEmitter(ablytest.DiscardLogger)
+
+		addedCalled := make(chan struct{}, 1)
+		removedCalled := make(chan struct{}, 1)
+
+		var off func()
+
+		// Shuffle the order in which we register the listeners for extra
+		// exhaustiveness.
+		ons := []func(){func() {
+			off = em.OnAll(func(ably.EmitterData) {
+				close(removedCalled)
+			})
+		}, func() {
+			em.OnAll(func(ably.EmitterData) {
+				off()
+
+				em.OnAll(func(ably.EmitterData) {
+					close(addedCalled)
+				})
+			})
+		}}
+		rand.Shuffle(len(ons), func(i, j int) {
+			ons[i], ons[j] = ons[j], ons[i]
+		})
+		for _, on := range ons {
+			on()
+		}
+
+		em.Emit(nil, nil)
+
+		ablytest.Instantly.NoRecv(t, nil, addedCalled, t.Errorf)
+		ablytest.Instantly.Recv(t, nil, removedCalled, t.Errorf)
+	}
+}
+
+func assertNoFurtherReceives(t *testing.T, received map[ably.EmitterString]chan ably.EmitterData) {
+	t.Helper()
+	for name, ch := range received {
+		fatalf := ablytest.FmtFunc.Wrap(t.Fatalf, t, "received to %q: %s", name)
+		ablytest.Instantly.NoRecv(t, nil, ch, fatalf)
+	}
+}
diff --git a/ably/event_emitter_test.go b/ably/event_emitter_test.go
new file mode 100644
index 0000000..945b8a7
--- /dev/null
+++ b/ably/event_emitter_test.go
@@ -0,0 +1,77 @@
+package ably_test
+
+import (
+	"testing"
+
+	"github.com/ably/ably-go/ably"
+	"github.com/ably/ably-go/ably/ablytest"
+)
+
+func TestEventEmitterConcurrency(t *testing.T) {
+	em := ably.NewEventEmitter(ablytest.DiscardLogger)
+
+	type called struct {
+		i    int
+		goOn chan struct{}
+	}
+	calls := make(chan called)
+
+	for i := 0; i < 2; i++ {
+		i := i
+		em.OnAll(func(ably.EmitterData) {
+			c := called{i: i, goOn: make(chan struct{})}
+			calls <- c
+			<-c.goOn
+		})
+	}
+
+	// Emit, and, since handlers are concurrent, expect a call per handler to
+	// be initiated.
+
+	em.Emit(ably.EmitterString("foo"), nil)
+
+	var ongoingCalls []called
+
+	for i := 0; i < 2; i++ {
+		var call called
+		ablytest.Instantly.Recv(t, &call, calls, t.Fatalf)
+		ongoingCalls = append(ongoingCalls, call)
+	}
+	ablytest.Instantly.NoRecv(t, nil, calls, t.Fatalf)
+
+	// While the last event is still being handled by each handler, a new event
+	// should be enqueued.
+
+	em.Emit(ably.EmitterString("foo"), nil)
+
+	ablytest.Instantly.NoRecv(t, nil, calls, t.Fatalf)
+
+	// Allow the first ongoing call to finish, which should then process the
+	// enqueued event for that handler.
+
+	close(ongoingCalls[0].goOn)
+	var call called
+	ablytest.Instantly.Recv(t, &call, calls, t.Fatalf)
+	if expected, got := ongoingCalls[0].i, call.i; expected != got {
+		t.Fatalf("expected to unblock handler %d, got %d", expected, got)
+	}
+	close(call.goOn)
+
+	// Unblock the other handler too.
+	close(ongoingCalls[1].goOn)
+	ablytest.Instantly.Recv(t, &call, calls, t.Fatalf)
+	if expected, got := ongoingCalls[1].i, call.i; expected != got {
+		t.Fatalf("expected to unblock handler %d, got %d", expected, got)
+	}
+	close(call.goOn)
+
+	// Make sure things still work after emptying the queues.
+
+	em.Emit(ably.EmitterString("foo"), nil)
+	for i := 0; i < 2; i++ {
+		ablytest.Instantly.Recv(t, &call, calls, t.Fatalf)
+		close(call.goOn)
+	}
+
+	ablytest.Instantly.NoRecv(t, nil, calls, t.Fatalf)
+}
diff --git a/ably/export_test.go b/ably/export_test.go
index 6feca6d..d3833c0 100644
--- a/ably/export_test.go
+++ b/ably/export_test.go
@@ -38,11 +38,11 @@ func ErrorCode(err error) int {
 	return code(err)
 }
 
-// MustRealtimeClient is like NewRealtimeClient, but panics on error.
-func MustRealtimeClient(opts *ClientOptions) *RealtimeClient {
-	client, err := NewRealtimeClient(opts)
+// MustRealtime is like NewRealtime, but panics on error.
+func MustRealtime(opts *ClientOptions) *Realtime {
+	client, err := NewRealtime(opts)
 	if err != nil {
-		panic("ably.NewRealtimeClient failed: " + err.Error())
+		panic("ably.NewRealtime failed: " + err.Error())
 	}
 	return client
 }
@@ -80,3 +80,29 @@ func (c *RestClient) GetCachedFallbackHost() string {
 func (opts *ClientOptions) GetFallbackRetryTimeout() time.Duration {
 	return opts.fallbackRetryTimeout()
 }
+
+func NewErrorInfo(code int, err error) *ErrorInfo {
+	return newError(code, err)
+}
+
+var NewEventEmitter = newEventEmitter
+
+type EventEmitter = eventEmitter
+type EmitterEvent = emitterEvent
+type EmitterData = emitterData
+
+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...)
+}
diff --git a/ably/proto/error_test.go b/ably/proto/error_test.go
deleted file mode 100644
index a162bbc..0000000
--- a/ably/proto/error_test.go
+++ /dev/null
@@ -1,80 +0,0 @@
-package proto_test
-
-import (
-	"strings"
-	"testing"
-
-	"github.com/ably/ably-go/ably/proto"
-)
-
-func TestErrorInfo(t *testing.T) {
-	t.Run("without an error code", func(ts *testing.T) {
-		e := &proto.ErrorInfo{
-			StatusCode: 401,
-		}
-		h := "help.ably.io"
-		if strings.Contains(e.Error(), h) {
-			ts.Errorf("expected error message not to contain %s", h)
-		}
-	})
-	t.Run("with an error code", func(ts *testing.T) {
-		e := &proto.ErrorInfo{
-			Code: 44444,
-		}
-		h := "https://help.ably.io/error/44444"
-		if !strings.Contains(e.Error(), h) {
-			ts.Errorf("expected error message %s  to contain %s", e.Error(), h)
-		}
-	})
-	t.Run("with an error code and an href attribute", func(ts *testing.T) {
-		href := "http://foo.bar.com/"
-		e := &proto.ErrorInfo{
-			Code: 44444,
-			HRef: href,
-		}
-		h := "https://help.ably.io/error/44444"
-		if strings.Contains(e.Error(), h) {
-			ts.Errorf("expected error message %s not to contain %s", e.Error(), h)
-		}
-		if !strings.Contains(e.Error(), href) {
-			ts.Errorf("expected error message %s  to contain %s", e.Error(), href)
-		}
-	})
-
-	t.Run("with an error code and a message with the same error URL", func(ts *testing.T) {
-		e := &proto.ErrorInfo{
-			Code:    44444,
-			Message: "error https://help.ably.io/error/44444",
-		}
-		h := "https://help.ably.io/error/44444"
-		if !strings.Contains(e.Error(), h) {
-			ts.Errorf("expected error message %s  to contain %s", e.Error(), h)
-		}
-		n := strings.Count(e.Error(), h)
-		if n != 1 {
-			ts.Errorf("expected 1 occupance of %s got %d", h, n)
-		}
-	})
-	t.Run("with an error code and a message with a different error URL", func(ts *testing.T) {
-		e := &proto.ErrorInfo{
-			Code:    44444,
-			Message: "error https://help.ably.io/error/123123",
-		}
-		h := "https://help.ably.io/error/44444"
-		if !strings.Contains(e.Error(), h) {
-			ts.Errorf("expected error message %s  to contain %s", e.Error(), h)
-		}
-		n := strings.Count(e.Error(), "help.ably.io")
-		if n != 2 {
-			ts.Errorf("expected 2 got %d", n)
-		}
-		n = strings.Count(e.Error(), "/123123")
-		if n != 1 {
-			ts.Errorf("expected 1 got %d", n)
-		}
-		n = strings.Count(e.Error(), "/44444")
-		if n != 1 {
-			ts.Errorf("expected 1 got %d", n)
-		}
-	})
-}
diff --git a/ably/realtime_channel_test.go b/ably/realtime_channel_test.go
index 085e1e6..4a3dba1 100644
--- a/ably/realtime_channel_test.go
+++ b/ably/realtime_channel_test.go
@@ -35,8 +35,8 @@ func expectMsg(ch <-chan *proto.Message, name string, data interface{}, t time.D
 
 func TestRealtimeChannel_Publish(t *testing.T) {
 	t.Parallel()
-	app, client := ablytest.NewRealtimeClient(nil)
-	defer safeclose(t, client, app)
+	app, client := ablytest.NewRealtime(nil)
+	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
 	channel := client.Channels.Get("test")
 	if err := ablytest.Wait(channel.Publish("hello", "world")); err != nil {
@@ -52,14 +52,14 @@ func TestRealtimeChannel_Failed(t *testing.T) {
 		NoQueueing: true,
 		Listener:   rec.Channel(),
 	}
-	app, client := ablytest.NewRealtimeClient(opts)
-	defer safeclose(t, client, app)
+	app, client := ablytest.NewRealtime(opts)
+	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
-	if err := ablytest.Wait(client.Connection.Connect()); err != nil {
+	if err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
 		t.Fatalf("Connect()=%v", err)
 	}
 	channel := client.Channels.GetAndAttach("test")
-	if err := client.Close(); err != nil {
+	if err := ablytest.FullRealtimeCloser(client).Close(); err != nil {
 		t.Fatalf("Close()=%v", err)
 	}
 	want := []ably.StateEnum{
@@ -87,10 +87,10 @@ func TestRealtimeChannel_Failed(t *testing.T) {
 
 func TestRealtimeChannel_Subscribe(t *testing.T) {
 	t.Parallel()
-	app, client1 := ablytest.NewRealtimeClient(nil)
-	defer safeclose(t, client1, app)
-	client2 := app.NewRealtimeClient(&ably.ClientOptions{NoEcho: true})
-	defer safeclose(t, client2)
+	app, client1 := ablytest.NewRealtime(nil)
+	defer safeclose(t, ablytest.FullRealtimeCloser(client1), app)
+	client2 := app.NewRealtime(&ably.ClientOptions{NoEcho: true})
+	defer safeclose(t, ablytest.FullRealtimeCloser(client2))
 
 	channel1 := client1.Channels.Get("test")
 	channel2 := client2.Channels.Get("test")
@@ -160,8 +160,8 @@ var chanCloseTransitions = [][]ably.StateEnum{{
 func TestRealtimeChannel_Close(t *testing.T) {
 	t.Parallel()
 	rec := ablytest.NewStateRecorder(8)
-	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{Listener: rec.Channel()})
-	defer safeclose(t, client, app)
+	app, client := ablytest.NewRealtime(&ably.ClientOptions{Listener: rec.Channel()})
+	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
 	channel := client.Channels.Get("test")
 	sub, err := channel.Subscribe()
@@ -192,8 +192,8 @@ func TestRealtimeChannel_Close(t *testing.T) {
 	if err := channel.Close(); err != nil {
 		t.Fatalf("channel.Close()=%v", err)
 	}
-	if err := client.Close(); err != nil {
-		t.Fatalf("client.Close()=%v", err)
+	if err := ablytest.FullRealtimeCloser(client).Close(); err != nil {
+		t.Fatalf("ablytest.FullRealtimeCloser(client).Close()=%v", err)
 	}
 	select {
 	case err := <-done:
diff --git a/ably/realtime_client_test.go b/ably/realtime_client_test.go
index e31b18f..538d710 100644
--- a/ably/realtime_client_test.go
+++ b/ably/realtime_client_test.go
@@ -9,7 +9,7 @@ import (
 	"github.com/ably/ably-go/ably/ablytest"
 )
 
-func TestRealtimeClient_RealtimeHost(t *testing.T) {
+func TestRealtime_RealtimeHost(t *testing.T) {
 	t.Parallel()
 	httpClient := ablytest.NewHTTPClient()
 	app, err := ablytest.NewSandbox(nil)
@@ -28,16 +28,16 @@ func TestRealtimeClient_RealtimeHost(t *testing.T) {
 	for _, host := range hosts {
 		opts := rec.Options(host)
 		opts.Listener = stateRec.Channel()
-		client, err := ably.NewRealtimeClient(app.Options(opts))
+		client, err := ably.NewRealtime(app.Options(opts))
 		if err != nil {
-			t.Errorf("NewRealtimeClient=%s (host=%s)", err, host)
+			t.Errorf("NewRealtime=%s (host=%s)", err, host)
 			continue
 		}
 		if state := client.Connection.State(); state != ably.StateConnInitialized {
 			t.Errorf("want state=%v; got %s", ably.StateConnInitialized, state)
 			continue
 		}
-		if err := checkError(80000, ablytest.Wait(client.Connection.Connect())); err != nil {
+		if err := checkError(80000, ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait()); err != nil {
 			t.Errorf("%s (host=%s)", err, host)
 			continue
 		}
@@ -45,7 +45,7 @@ func TestRealtimeClient_RealtimeHost(t *testing.T) {
 			t.Errorf("want state=%v; got %s", ably.StateConnFailed, state)
 			continue
 		}
-		if err := checkError(80000, client.Close()); err != nil {
+		if err := checkError(80000, ablytest.FullRealtimeCloser(client).Close()); err != nil {
 			t.Errorf("%s (host=%s)", err, host)
 			continue
 		}
@@ -90,7 +90,7 @@ func checkUnique(ch chan string, typ string, n int) error {
 	return nil
 }
 
-func TestRealtimeClient_multiple(t *testing.T) {
+func TestRealtime_multiple(t *testing.T) {
 	t.Parallel()
 	const N = 3
 	var all ablytest.ResultGroup
@@ -109,13 +109,13 @@ func TestRealtimeClient_multiple(t *testing.T) {
 			opts := app.Options(nil)
 			opts.ClientID = fmt.Sprintf("client/%d", i)
 			opts.NoConnect = true
-			c, err := ably.NewRealtimeClient(opts)
+			c, err := ably.NewRealtime(opts)
 			if err != nil {
 				all.Add(nil, err)
 				return
 			}
 			var rg ablytest.ResultGroup
-			rg.Add(c.Connection.Connect())
+			rg.Add(ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected), nil)
 			for j := 0; j < 10; j++ {
 				channel := c.Channels.Get(fmt.Sprintf("client/%d/channel/%d", i, j))
 				rg.Add(channel.Attach())
@@ -143,7 +143,7 @@ func TestRealtimeClient_multiple(t *testing.T) {
 			}
 			idch <- c.Connection.ID()
 			keych <- c.Connection.Key()
-			if err := c.Close(); err != nil {
+			if err := ablytest.FullRealtimeCloser(c).Close(); err != nil {
 				all.Add(nil, err)
 			}
 		}(i)
@@ -160,12 +160,8 @@ func TestRealtimeClient_multiple(t *testing.T) {
 	}
 }
 
-func TestRealtimeClient_DontCrashOnCloseWhenEchoOff(t *testing.T) {
+func TestRealtime_DontCrashOnCloseWhenEchoOff(t *testing.T) {
 	t.Parallel()
-	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{NoConnect: true})
-	defer safeclose(t, app)
-
-	if err := checkError(80002, client.Close()); err != nil {
-		t.Fatal(err)
-	}
+	app, client := ablytest.NewRealtime(&ably.ClientOptions{NoConnect: true})
+	defer safeclose(t, app, ablytest.FullRealtimeCloser(client))
 }
diff --git a/ably/realtime_conn_spec_test.go b/ably/realtime_conn_spec_test.go
new file mode 100644
index 0000000..eab2eea
--- /dev/null
+++ b/ably/realtime_conn_spec_test.go
@@ -0,0 +1,147 @@
+package ably_test
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/ably/ably-go/ably"
+	"github.com/ably/ably-go/ably/ablytest"
+)
+
+func Test_RTN4a_ConnectionEventForStateChange(t *testing.T) {
+	t.Run(fmt.Sprintf("on %s", ably.ConnectionStateConnecting), func(t *testing.T) {
+		t.Parallel()
+
+		app, realtime := ablytest.NewRealtime(&ably.ClientOptions{NoConnect: true})
+		defer safeclose(t, ablytest.FullRealtimeCloser(realtime), app)
+
+		changes := make(chan ably.ConnectionStateChange)
+		defer ablytest.Instantly.NoRecv(t, nil, changes, t.Errorf)
+
+		realtime.Connection.On(ably.ConnectionEventConnecting, func(change ably.ConnectionStateChange) {
+			changes <- change
+		})
+
+		realtime.Connect()
+
+		ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+	})
+
+	t.Run(fmt.Sprintf("on %s", ably.ConnectionStateConnected), func(t *testing.T) {
+		t.Parallel()
+
+		app, realtime := ablytest.NewRealtime(&ably.ClientOptions{NoConnect: true})
+		defer safeclose(t, ablytest.FullRealtimeCloser(realtime), app)
+
+		connectAndWait(t, realtime)
+	})
+
+	t.Run(fmt.Sprintf("on %s", ably.ConnectionStateDisconnected), func(t *testing.T) {
+		t.Parallel()
+
+		options := &ably.ClientOptions{NoConnect: true}
+		disconnect := ablytest.SetFakeDisconnect(options)
+		app, realtime := ablytest.NewRealtime(options)
+		defer safeclose(t, app)
+		defer realtime.Close()
+
+		connectAndWait(t, realtime)
+
+		changes := make(chan ably.ConnectionStateChange)
+		defer ablytest.Instantly.NoRecv(t, nil, changes, t.Errorf)
+
+		realtime.Connection.On(ably.ConnectionEventDisconnected, func(change ably.ConnectionStateChange) {
+			changes <- change
+		})
+
+		err := disconnect()
+		if err != nil {
+			t.Fatalf("fake disconnection failed: %v", err)
+		}
+
+		ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+
+	})
+
+	t.Run(fmt.Sprintf("on %s", ably.ConnectionStateSuspended), func(t *testing.T) {
+		t.Parallel()
+
+		t.Skip("SUSPENDED not yet implemented")
+	})
+
+	t.Run(fmt.Sprintf("on %s", ably.ConnectionStateClosing), func(t *testing.T) {
+		t.Parallel()
+
+		app, realtime := ablytest.NewRealtime(&ably.ClientOptions{NoConnect: true})
+		defer safeclose(t, ablytest.FullRealtimeCloser(realtime), app)
+
+		connectAndWait(t, realtime)
+
+		changes := make(chan ably.ConnectionStateChange)
+		defer ablytest.Instantly.NoRecv(t, nil, changes, t.Errorf)
+
+		realtime.Connection.On(ably.ConnectionEventClosing, func(change ably.ConnectionStateChange) {
+			changes <- change
+		})
+
+		realtime.Close()
+		ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+	})
+
+	t.Run(fmt.Sprintf("on %s", ably.ConnectionStateClosed), func(t *testing.T) {
+		t.Parallel()
+
+		app, realtime := ablytest.NewRealtime(&ably.ClientOptions{NoConnect: true})
+		defer safeclose(t, ablytest.FullRealtimeCloser(realtime), app)
+
+		connectAndWait(t, realtime)
+
+		changes := make(chan ably.ConnectionStateChange)
+		defer ablytest.Instantly.NoRecv(t, nil, changes, t.Errorf)
+
+		realtime.Connection.On(ably.ConnectionEventClosed, func(change ably.ConnectionStateChange) {
+			changes <- change
+		})
+
+		realtime.Close()
+		ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+	})
+
+	t.Run(fmt.Sprintf("on %s", ably.ConnectionStateFailed), func(t *testing.T) {
+		t.Parallel()
+
+		var options ably.ClientOptions
+		options.Environment = "sandbox"
+		options.NoConnect = true
+		options.Key = "made:up"
+
+		realtime, err := ably.NewRealtime(&options)
+		if err != nil {
+			t.Fatalf("unexpected err: %s", err)
+		}
+
+		changes := make(chan ably.ConnectionStateChange)
+		defer ablytest.Instantly.NoRecv(t, nil, changes, t.Errorf)
+
+		realtime.Connection.On(ably.ConnectionEventFailed, func(change ably.ConnectionStateChange) {
+			changes <- change
+		})
+
+		realtime.Connect()
+		ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+	})
+}
+
+func connectAndWait(t *testing.T, realtime *ably.Realtime) {
+	t.Helper()
+
+	changes := make(chan ably.ConnectionStateChange)
+	defer ablytest.Instantly.NoRecv(t, nil, changes, t.Errorf)
+
+	realtime.Connection.Once(ably.ConnectionEventConnected, func(change ably.ConnectionStateChange) {
+		changes <- change
+	})
+
+	realtime.Connect()
+	ablytest.Soon.Recv(t, nil, changes, t.Fatalf)
+}
diff --git a/ably/realtime_conn_test.go b/ably/realtime_conn_test.go
index 49b6b25..c86a42e 100644
--- a/ably/realtime_conn_test.go
+++ b/ably/realtime_conn_test.go
@@ -33,8 +33,8 @@ var connTransitions = []ably.StateEnum{
 func TestRealtimeConn_Connect(t *testing.T) {
 	t.Parallel()
 	rec := ablytest.NewStateRecorder(4)
-	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{Listener: rec.Channel()})
-	defer safeclose(t, client, app)
+	app, client := ablytest.NewRealtime(&ably.ClientOptions{Listener: rec.Channel()})
+	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
 	if err := await(client.Connection.State, ably.StateConnConnected); err != nil {
 		t.Fatal(err)
@@ -42,8 +42,8 @@ func TestRealtimeConn_Connect(t *testing.T) {
 	if serial := client.Connection.Serial(); serial != -1 {
 		t.Fatalf("want serial=-1; got %d", serial)
 	}
-	if err := client.Close(); err != nil {
-		t.Fatalf("client.Close()=%v", err)
+	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)
@@ -57,18 +57,18 @@ func TestRealtimeConn_NoConnect(t *testing.T) {
 		Listener:  rec.Channel(),
 		NoConnect: true,
 	}
-	app, client := ablytest.NewRealtimeClient(opts)
-	defer safeclose(t, client, app)
+	app, client := ablytest.NewRealtime(opts)
+	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
-	client.Connection.On(rec.Channel())
-	if err := ablytest.Wait(client.Connection.Connect()); err != nil {
+	client.Connection.OnState(rec.Channel())
+	if err := ablytest.ConnWaiter(client, client.Connect, 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)
 	}
-	if err := client.Close(); err != nil {
-		t.Fatalf("client.Close()=%v", err)
+	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 {
@@ -86,14 +86,14 @@ var connCloseTransitions = []ably.StateEnum{
 func TestRealtimeConn_ConnectClose(t *testing.T) {
 	t.Parallel()
 	rec := ablytest.NewStateRecorder(4)
-	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{Listener: rec.Channel()})
-	defer safeclose(t, client, app)
+	app, client := ablytest.NewRealtime(&ably.ClientOptions{Listener: rec.Channel()})
+	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
 	if err := await(client.Connection.State, ably.StateConnConnected); err != nil {
 		t.Fatal(err)
 	}
-	if err := client.Close(); err != nil {
-		t.Fatalf("client.Close()=%v", err)
+	if err := ablytest.FullRealtimeCloser(client).Close(); err != nil {
+		t.Fatalf("ablytest.FullRealtimeCloser(client).Close()=%v", err)
 	}
 	if err := await(client.Connection.State, ably.StateConnClosed); err != nil {
 		t.Fatal(err)
@@ -106,15 +106,20 @@ func TestRealtimeConn_ConnectClose(t *testing.T) {
 
 func TestRealtimeConn_AlreadyConnected(t *testing.T) {
 	t.Parallel()
-	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{NoConnect: true})
-	defer safeclose(t, client, app)
+	app, client := ablytest.NewRealtime(&ably.ClientOptions{NoConnect: true})
+	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
-	if err := ablytest.Wait(client.Connection.Connect()); err != nil {
-		t.Fatalf("Connect=%s", err)
-	}
-	if err := ablytest.Wait(client.Connection.Connect()); err != nil {
+	if err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
 		t.Fatalf("Connect=%s", err)
 	}
+
+	changes := make(chan ably.ConnectionStateChange, 1)
+	off := client.Connection.OnceAll(func(change ably.ConnectionStateChange) {
+		changes <- change
+	})
+	defer off()
+
+	ablytest.Before(100*time.Millisecond).NoRecv(t, nil, changes, t.Fatalf)
 }
 
 func TestRealtimeConn_AuthError(t *testing.T) {
@@ -125,17 +130,17 @@ func TestRealtimeConn_AuthError(t *testing.T) {
 		},
 		NoConnect: true,
 	}
-	client, err := ably.NewRealtimeClient(opts)
+	client, err := ably.NewRealtime(opts)
 	if err != nil {
-		t.Fatalf("NewRealtimeClient()=%v", err)
+		t.Fatalf("NewRealtime()=%v", err)
 	}
-	if err = ablytest.Wait(client.Connection.Connect()); err == nil {
+	if err = ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err == nil {
 		t.Fatal("Connect(): want err != nil")
 	}
 	if state := client.Connection.State(); state != ably.StateConnFailed {
 		t.Fatalf("want state=%s; got %s", ably.StateConnFailed, state)
 	}
-	if err = client.Close(); err == nil {
+	if err = ablytest.FullRealtimeCloser(client).Close(); err == nil {
 		t.Fatal("Close(): want err != nil")
 	}
 }
diff --git a/ably/realtime_presence_test.go b/ably/realtime_presence_test.go
index 9277eda..c732b8b 100644
--- a/ably/realtime_presence_test.go
+++ b/ably/realtime_presence_test.go
@@ -41,8 +41,8 @@ var fixtureMembers = []string{
 
 func TestRealtimePresence_Sync(t *testing.T) {
 	t.Parallel()
-	app, client := ablytest.NewRealtimeClient(nil)
-	defer safeclose(t, client, app)
+	app, client := ablytest.NewRealtime(nil)
+	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 
 	members, err := client.Channels.GetAndAttach("persisted:presence_fixtures").Presence.Get(true)
 	if err != nil {
@@ -55,11 +55,11 @@ func TestRealtimePresence_Sync(t *testing.T) {
 
 func TestRealtimePresence_Sync250(t *testing.T) {
 	t.Parallel()
-	app, client1 := ablytest.NewRealtimeClient(nil)
-	defer safeclose(t, client1, app)
-	client2 := app.NewRealtimeClient()
-	client3 := app.NewRealtimeClient()
-	defer safeclose(t, client2, client3)
+	app, client1 := ablytest.NewRealtime(nil)
+	defer safeclose(t, ablytest.FullRealtimeCloser(client1), app)
+	client2 := app.NewRealtime()
+	client3 := app.NewRealtime()
+	defer safeclose(t, ablytest.FullRealtimeCloser(client2), ablytest.FullRealtimeCloser(client3))
 
 	channel1 := client1.Channels.GetAndAttach("sync250")
 	channel2 := client2.Channels.GetAndAttach("sync250")
@@ -115,10 +115,10 @@ func TestRealtimePresence_EnsureChannelIsAttached(t *testing.T) {
 		Listener:  rec.Channel(),
 		NoConnect: true,
 	}
-	app, client := ablytest.NewRealtimeClient(opts)
-	defer safeclose(t, client, app)
+	app, client := ablytest.NewRealtime(opts)
+	defer safeclose(t, ablytest.FullRealtimeCloser(client), app)
 	channel := client.Channels.Get("persisted:presence_fixtures")
-	if err := ablytest.Wait(client.Connection.Connect()); err != nil {
+	if err := ablytest.ConnWaiter(client, client.Connect, ably.ConnectionEventConnected).Wait(); err != nil {
 		t.Fatal(err)
 	}
 	if err := rec.WaitFor(presTransitions[:2]); err != nil {