Repository: ably/ably-go
Pull request #179 (2020-07-31 18:09:17)

RTN15h*: Handle incoming DISCONNECTED while CONNECTED
Implements [RTN15h*](https://docs.ably.io/client-lib-development-guide/features/#RTN15h).


Patch:
diff --git a/ably/realtime_client.go b/ably/realtime_client.go
index 1909788..04d5fbd 100644
--- a/ably/realtime_client.go
+++ b/ably/realtime_client.go
@@ -29,7 +29,10 @@ func NewRealtime(options ClientOptions) (*Realtime, error) {
 	c.Auth = rest.Auth
 	c.Channels = newChannels(c)
 	conn, err := newConn(c.opts(), rest.Auth, connCallbacks{
-		c.onChannelMsg, c.onReconnectMsg, c.onConnStateChange,
+		c.onChannelMsg,
+		c.onReconnected,
+		c.onReconnectionFailed,
+		c.onConnStateChange,
 	})
 	if err != nil {
 		return nil, err
@@ -64,30 +67,27 @@ func (c *Realtime) onChannelMsg(msg *proto.ProtocolMessage) {
 	c.Channels.Get(msg.Channel).notify(msg)
 }
 
-func (c *Realtime) onReconnectMsg(msg *proto.ProtocolMessage, isNewID bool) {
-	switch msg.Action {
-	case proto.ActionConnected:
-		if msg.Error != nil || // RTN15c3
-			isNewID { // RTN15g3
-			for _, ch := range c.Channels.All() {
-				switch ch.State() {
-				// TODO: SUSPENDED
-				case StateChanAttaching, StateChanAttached:
-					ch.mayAttach(false, false)
-				}
-			}
-		}
-
-	case proto.ActionError:
-		// (RTN15c4)
+func (c *Realtime) onReconnected(err *proto.ErrorInfo, isNewID bool) {
+	if err == nil /* RTN15c3 */ && !isNewID /* RTN15g3 */ {
+		return
+	}
 
-		for _, ch := range c.Channels.All() {
-			ch.state.syncSet(StateChanFailed, newErrorProto(msg.Error))
+	for _, ch := range c.Channels.All() {
+		switch ch.State() {
+		// TODO: SUSPENDED
+		case StateChanAttaching, StateChanAttached:
+			ch.mayAttach(false, false)
 		}
 	}
 }
 
-func tokenError(err *proto.ErrorInfo) bool {
+func (c *Realtime) onReconnectionFailed(err *proto.ErrorInfo) {
+	for _, ch := range c.Channels.All() {
+		ch.state.syncSet(StateChanFailed, newErrorProto(err))
+	}
+}
+
+func isTokenError(err *proto.ErrorInfo) bool {
 	return err.StatusCode == http.StatusUnauthorized && (40140 <= err.Code && err.Code < 40150)
 }
 
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index 62c1fe8..fa02b5b 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -48,19 +48,22 @@ type Connection struct {
 	// with this set to true then its the first message/response after issuing the
 	// reconnection request.
 	reconnecting bool
+	// reauthorizing tracks if the current reconnection attempt is happening
+	// after a reauthorization, to avoid re-reauthorizing.
+	reauthorizing bool
 }
 
 type connCallbacks struct {
 	onChannelMsg func(*proto.ProtocolMessage)
-	// onReconnectMsg is called when we get a response from reconnect request. We
+	// onReconnected is called when we get a CONNECTED response from reconnect request. We
 	// move this up because some implementation details for (RTN15c) requires
 	// access to Channels and we dont have it here so we let RealtimeClient do the
 	// work.
-	onReconnectMsg func(_ *proto.ProtocolMessage, isNewID bool)
-	onStateChange  func(State)
-	// reconnecting tracks if we have issued a reconnection request. If we receive any message
-	// with this set to true then its the first message/response after issuing the
+	onReconnected func(_ *proto.ErrorInfo, isNewID bool)
+	// onReconnectionFailed is called when we get a FAILED response from a
 	// reconnection request.
+	onReconnectionFailed func(*proto.ErrorInfo)
+	onStateChange        func(State)
 }
 
 func newConn(opts *clientOptions, auth *Auth, callbacks connCallbacks) (*Connection, error) {
@@ -97,6 +100,12 @@ func (c *Connection) dial(proto string, u *url.URL) (proto.Conn, error) {
 // Connect attempts to move the connection to the CONNECTED state, if it
 // can and if it isn't already.
 func (c *Connection) Connect() {
+	c.state.Lock()
+	isActive := c.isActive()
+	c.state.Unlock()
+	if isActive {
+		return
+	}
 	c.connect(false)
 }
 
@@ -200,10 +209,9 @@ func (c *Connection) params(mode connectionMode) (url.Values, error) {
 func (c *Connection) connectWith(result bool, mode connectionMode) (Result, error) {
 	c.state.Lock()
 	defer c.state.Unlock()
-	if c.isActive() {
-		return nopResult, nil
+	if !c.isActive() {
+		c.setState(StateConnConnecting, nil)
 	}
-	c.setState(StateConnConnecting, nil)
 	u, err := url.Parse(c.opts.realtimeURL())
 	if err != nil {
 		return nil, c.setState(StateConnFailed, err)
@@ -537,58 +545,64 @@ func (c *Connection) eventloop() {
 				c.callbacks.onChannelMsg(msg)
 				break
 			}
+
 			c.state.Lock()
-			if c.reconnecting {
-				c.reconnecting = false
-				if tokenError(msg.Error) {
-					// (RTN15c5)
-					// TODO: (gernest) implement (RTN15h) This can be done as a separate task?
+			reauthorizing := c.reauthorizing
+			c.reauthorizing = false
+			if isTokenError(msg.Error) {
+				if reauthorizing {
+					c.lockedReauthorizationFailed(newErrorProto(msg.Error))
+					c.state.Unlock()
+					return
 				} else {
-					// (RTN15c4)
-					c.callbacks.onReconnectMsg(msg, false)
+					// TODO: RTN14b; may reuse c.reauthorize from RTN15h2.
 				}
 			}
-			c.setState(StateConnFailed, newErrorProto(msg.Error))
 			c.state.Unlock()
-			c.queue.Fail(newErrorProto(msg.Error))
-			if c.conn != nil {
-				c.conn.Close()
-			}
+
+			c.failedConnSideEffects(msg.Error)
 		case proto.ActionConnected:
 			c.state.Lock()
 			// we need to get this before we set c.key so as to be sure if we were
 			// resuming or recovering the connection.
 			mode := c.getMode()
-			c.state.Unlock()
 			if msg.ConnectionDetails != nil {
 				connDetails = msg.ConnectionDetails
-
-				c.state.Lock()
 				c.key = connDetails.ConnectionKey //(RTN15e) (RTN16d)
-				c.state.Unlock()
 
 				// Spec RSA7b3, RSA7b4, RSA12a
 				c.auth.updateClientID(connDetails.ClientID)
 			}
-			c.state.Lock()
 			reconnecting := c.reconnecting
 			if reconnecting {
 				// reset the mode
 				c.reconnecting = false
+				c.reauthorizing = false
 			}
-			id := c.id
+			previousID := c.id
+			c.id = msg.ConnectionID
+			c.msgSerial = 0
+			if reconnecting && mode == recoveryMode {
+				// we are setting msgSerial as per (RTN16f)
+				msgSerial, err := strconv.ParseInt(strings.Split(c.opts.Recover, ":")[2], 10, 64)
+				if err != nil {
+					//TODO: how to handle this? Panic?
+				}
+				c.msgSerial = msgSerial
+			}
+			c.setSerial(-1)
 			c.state.Unlock()
 			if reconnecting {
 				// (RTN15c1) (RTN15c2)
 				c.state.Lock()
 				c.setState(StateConnConnected, newErrorProto(msg.Error))
 				c.state.Unlock()
-				if id != msg.ConnectionID {
+				if previousID != msg.ConnectionID {
 					// (RTN15c3)
 					// we are calling this outside of locks to avoid deadlock because in the
 					// RealtimeClient client where this callback is implemented we do some ops
 					// with this Conn where we re acquire Conn.state.Lock again.
-					c.callbacks.onReconnectMsg(msg, true)
+					c.callbacks.onReconnected(msg.Error, true)
 				}
 			} else {
 				// preserve old behavior.
@@ -596,25 +610,24 @@ func (c *Connection) eventloop() {
 				c.setState(StateConnConnected, nil)
 				c.state.Unlock()
 			}
-			c.state.Lock()
-			c.id = msg.ConnectionID
-			c.msgSerial = 0
-			if reconnecting && mode == recoveryMode {
-				// we are setting msgSerial as per (RTN16f)
-				msgSerial, err := strconv.ParseInt(strings.Split(c.opts.Recover, ":")[2], 10, 64)
-				if err != nil {
-					//TODO: how to handle this? Panic?
-				}
-				c.msgSerial = msgSerial
-			}
-			c.setSerial(-1)
-			c.state.Unlock()
 			c.queue.Flush()
 		case proto.ActionDisconnected:
-			c.state.Lock()
-			c.id = ""
-			c.setState(StateConnDisconnected, nil)
-			c.state.Unlock()
+			if !isTokenError(msg.Error) {
+				// The spec doesn't say what to do in this case, so do nothing.
+				// Ably is supposed to then close the transport, which will
+				// trigger a transition to DISCONNECTED.
+				continue
+			}
+
+			if !c.auth.isTokenRenewable() {
+				// RTN15h1
+				c.failedConnSideEffects(msg.Error)
+				return
+			}
+
+			// RTN15h2
+			c.reauthorize(lastActivityAt, connDetails)
+			return
 		case proto.ActionClosed:
 			c.state.Lock()
 			c.id, c.key = "", "" //(RTN16c)
@@ -641,6 +654,42 @@ func (c *Connection) setState(state StateEnum, err error) error {
 	return c.state.set(state, err)
 }
 
+func (c *Connection) failedConnSideEffects(err *proto.ErrorInfo) {
+	c.state.Lock()
+	if c.reconnecting {
+		c.reconnecting = false
+		c.reauthorizing = false
+		c.callbacks.onReconnectionFailed(err)
+	}
+	c.setState(StateConnFailed, newErrorProto(err))
+	c.state.Unlock()
+	c.queue.Fail(newErrorProto(err))
+	if c.conn != nil {
+		c.conn.Close()
+	}
+}
+
+func (c *Connection) reauthorize(lastActivityAt time.Time, connDetails *proto.ConnectionDetails) {
+	c.state.Lock()
+	_, err := c.auth.reauthorize()
+	if err != nil {
+		c.lockedReauthorizationFailed(err)
+		c.state.Unlock()
+		return
+	}
+
+	// The reauthorize above will have set the new token in c.auth, so
+	// reconnecting will use the new token.
+	c.reauthorizing = true
+	c.state.Unlock()
+	c.reconnect(lastActivityAt, connDetails, false)
+}
+
+func (c *Connection) lockedReauthorizationFailed(err error) {
+	c.setState(StateConnDisconnected, err)
+	// TODO: RTN14d
+}
+
 type verboseConn struct {
 	conn   proto.Conn
 	logger *LoggerOptions


Test patch:
diff --git a/ably/ablytest/recorders.go b/ably/ablytest/recorders.go
index 4a3f471..e6c11c7 100644
--- a/ably/ablytest/recorders.go
+++ b/ably/ablytest/recorders.go
@@ -289,18 +289,35 @@ func (rec *StateRecorder) timeout() time.Duration {
 	return 15 * time.Second
 }
 
-func MessagePipe(in <-chan *proto.ProtocolMessage, out chan<- *proto.ProtocolMessage) func(string, *url.URL) (proto.Conn, error) {
+type MessagePipeOption func(*pipeConn)
+
+// MessagePipeWithNowFunc sets a function to get the current time. This time
+// will be used to determine whether a Receive times out.
+//
+// If not set, receives won't timeout.
+func MessagePipeWithNowFunc(now func() time.Time) MessagePipeOption {
+	return func(pc *pipeConn) {
+		pc.now = now
+	}
+}
+
+func MessagePipe(in <-chan *proto.ProtocolMessage, out chan<- *proto.ProtocolMessage, opts ...MessagePipeOption) func(string, *url.URL) (proto.Conn, error) {
 	return func(proto string, u *url.URL) (proto.Conn, error) {
-		return pipeConn{
+		pc := pipeConn{
 			in:  in,
 			out: out,
-		}, nil
+		}
+		for _, opt := range opts {
+			opt(&pc)
+		}
+		return pc, nil
 	}
 }
 
 type pipeConn struct {
 	in  <-chan *proto.ProtocolMessage
 	out chan<- *proto.ProtocolMessage
+	now func() time.Time
 }
 
 func (pc pipeConn) Send(msg *proto.ProtocolMessage) error {
@@ -309,13 +326,17 @@ func (pc pipeConn) Send(msg *proto.ProtocolMessage) error {
 }
 
 func (pc pipeConn) Receive(deadline time.Time) (*proto.ProtocolMessage, error) {
+	var timeout <-chan time.Time
+	if pc.now != nil {
+		timeout = time.After(deadline.Sub(pc.now()))
+	}
 	select {
 	case m, ok := <-pc.in:
 		if !ok {
 			return nil, io.EOF
 		}
 		return m, nil
-	case <-time.After(time.Until(deadline)):
+	case <-timeout:
 		return nil, errTimeout{}
 	}
 }
diff --git a/ably/realtime_conn_spec_test.go b/ably/realtime_conn_spec_test.go
index 419559f..8a78b39 100644
--- a/ably/realtime_conn_spec_test.go
+++ b/ably/realtime_conn_spec_test.go
@@ -1,6 +1,7 @@
 package ably_test
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io"
@@ -1135,6 +1136,268 @@ func TestRealtimeConn_RTN15g_NewConnectionOnStateLost(t *testing.T) {
 	ablytest.Instantly.NoRecv(t, nil, out, t.Fatalf)
 }
 
+func TestRealtimeConn_RTN15h1_OnDisconnectedCannotRenewToken(t *testing.T) {
+	t.Parallel()
+
+	in := make(chan *proto.ProtocolMessage, 1)
+	out := make(chan *proto.ProtocolMessage, 16)
+
+	c, _ := ably.NewRealtime(ably.ClientOptions{}.
+		AutoConnect(false).
+		Token("fake:token").
+		Dial(ablytest.MessagePipe(in, out)))
+
+	in <- &proto.ProtocolMessage{
+		Action:            proto.ActionConnected,
+		ConnectionID:      "connection-id",
+		ConnectionDetails: &proto.ConnectionDetails{},
+	}
+
+	err := ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected).Wait()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tokenErr := proto.ErrorInfo{
+		StatusCode: 401,
+		Code:       40141,
+		Message:    "fake token error",
+	}
+
+	err = ablytest.ConnWaiter(c, func() {
+		in <- &proto.ProtocolMessage{
+			Action: proto.ActionDisconnected,
+			Error:  &tokenErr,
+		}
+	}, ably.ConnectionEventFailed).Wait()
+
+	var errInfo *ably.ErrorInfo
+	if !errors.As(err, &errInfo) {
+		t.Fatal(err)
+	}
+
+	if errInfo.StatusCode != tokenErr.StatusCode || errInfo.Code != tokenErr.Code {
+		t.Fatalf("expected the token error as FAILED state change reason; got: %s", err)
+	}
+}
+
+func TestRealtimeConn_RTN15h2_ReauthFails(t *testing.T) {
+	t.Parallel()
+
+	authErr := fmt.Errorf("reauth error")
+
+	in := make(chan *proto.ProtocolMessage, 1)
+	out := make(chan *proto.ProtocolMessage, 16)
+
+	authCallbackCalled := false
+
+	c, _ := ably.NewRealtime(ably.ClientOptions{}.
+		AutoConnect(false).
+		Token("fake:token").
+		AuthCallback(func(context.Context, ably.TokenParams) (ably.Tokener, error) {
+			if authCallbackCalled {
+				t.Errorf("expected a single attempt to reauth")
+			}
+			authCallbackCalled = true
+			return nil, authErr
+		}).
+		Dial(ablytest.MessagePipe(in, out)))
+
+	in <- &proto.ProtocolMessage{
+		Action:            proto.ActionConnected,
+		ConnectionID:      "connection-id",
+		ConnectionDetails: &proto.ConnectionDetails{},
+	}
+
+	err := ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected).Wait()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	tokenErr := proto.ErrorInfo{
+		StatusCode: 401,
+		Code:       40141,
+		Message:    "fake token error",
+	}
+
+	err = ablytest.ConnWaiter(c, func() {
+		in <- &proto.ProtocolMessage{
+			Action: proto.ActionDisconnected,
+			Error:  &tokenErr,
+		}
+	}, ably.ConnectionEventDisconnected).Wait()
+
+	if !errors.Is(err, authErr) {
+		t.Fatalf("expected the auth error as DISCONNECTED state change reason; got: %s", err)
+	}
+}
+
+func TestRealtimeConn_RTN15h2_ReauthWithBadToken(t *testing.T) {
+	t.Parallel()
+
+	in := make(chan *proto.ProtocolMessage, 1)
+	out := make(chan *proto.ProtocolMessage, 16)
+
+	dials := make(chan *url.URL, 1)
+
+	c, _ := ably.NewRealtime(ably.ClientOptions{}.
+		Token("fake:token").
+		AutoConnect(false).
+		AuthCallback(func(context.Context, ably.TokenParams) (ably.Tokener, error) {
+			return ably.TokenString("bad:token"), nil
+		}).
+		Dial(func(proto string, u *url.URL) (proto.Conn, error) {
+			dials <- u
+			return ablytest.MessagePipe(in, out)(proto, u)
+		}))
+
+	in <- &proto.ProtocolMessage{
+		Action:            proto.ActionConnected,
+		ConnectionID:      "connection-id",
+		ConnectionDetails: &proto.ConnectionDetails{},
+	}
+
+	err := ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected).Wait()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ablytest.Instantly.Recv(t, nil, dials, t.Fatalf)
+
+	tokenErr := proto.ErrorInfo{
+		StatusCode: 401,
+		Code:       40141,
+		Message:    "fake token error",
+	}
+
+	stateChanges := make(chan ably.ConnectionStateChange, 1)
+
+	off := c.Connection.OnAll(func(change ably.ConnectionStateChange) {
+		stateChanges <- change
+	})
+	defer off()
+
+	// Remove the buffer from dials, so that we can make the library wait for
+	// us to receive it before a state change.
+	dials = make(chan *url.URL)
+
+	in <- &proto.ProtocolMessage{
+		Action: proto.ActionDisconnected,
+		Error:  &tokenErr,
+	}
+
+	// No state change expected before a reauthorization and reconnection
+	// attempt.
+	ablytest.Instantly.NoRecv(t, nil, stateChanges, t.Fatalf)
+
+	// The DISCONNECTED causes a reauth, and dial again with the new
+	// token.
+	var dialURL *url.URL
+	ablytest.Instantly.Recv(t, &dialURL, dials, t.Fatalf)
+
+	if expected, got := "bad:token", dialURL.Query().Get("access_token"); expected != got {
+		t.Errorf("expected reauthorization with token returned by the authCallback; got %q", got)
+	}
+
+	// After a token error response, we finally get to our expected
+	// DISCONNECTED state.
+
+	in <- &proto.ProtocolMessage{
+		Action: proto.ActionError,
+		Error:  &tokenErr,
+	}
+
+	var change ably.ConnectionStateChange
+	ablytest.Instantly.Recv(t, &change, stateChanges, t.Fatalf)
+
+	if change.Current != ably.ConnectionStateDisconnected {
+		t.Fatalf("expected DISCONNECTED event; got %v", change)
+	}
+
+	if change.Reason.StatusCode != tokenErr.StatusCode || change.Reason.Code != tokenErr.Code {
+		t.Fatalf("expected the token error as FAILED state change reason; got: %s", change.Reason)
+	}
+}
+
+func TestRealtimeConn_RTN15h2_Success(t *testing.T) {
+	t.Parallel()
+
+	in := make(chan *proto.ProtocolMessage, 1)
+	out := make(chan *proto.ProtocolMessage, 16)
+
+	dials := make(chan *url.URL, 1)
+
+	c, _ := ably.NewRealtime(ably.ClientOptions{}.
+		Token("fake:token").
+		AutoConnect(false).
+		AuthCallback(func(context.Context, ably.TokenParams) (ably.Tokener, error) {
+			return ably.TokenString("good:token"), nil
+		}).
+		Dial(func(proto string, u *url.URL) (proto.Conn, error) {
+			dials <- u
+			return ablytest.MessagePipe(in, out)(proto, u)
+		}))
+
+	in <- &proto.ProtocolMessage{
+		Action:            proto.ActionConnected,
+		ConnectionID:      "connection-id",
+		ConnectionDetails: &proto.ConnectionDetails{},
+	}
+
+	err := ablytest.ConnWaiter(c, c.Connect, ably.ConnectionEventConnected).Wait()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ablytest.Instantly.Recv(t, nil, dials, t.Fatalf)
+
+	tokenErr := proto.ErrorInfo{
+		StatusCode: 401,
+		Code:       40141,
+		Message:    "fake token error",
+	}
+
+	stateChanges := make(chan ably.ConnectionStateChange, 1)
+
+	off := c.Connection.OnAll(func(change ably.ConnectionStateChange) {
+		stateChanges <- change
+	})
+	defer off()
+
+	in <- &proto.ProtocolMessage{
+		Action: proto.ActionDisconnected,
+		Error:  &tokenErr,
+	}
+
+	// The DISCONNECTED causes a reauth, and dial again with the new
+	// token.
+	var dialURL *url.URL
+	ablytest.Instantly.Recv(t, &dialURL, dials, t.Fatalf)
+
+	if expected, got := "good:token", dialURL.Query().Get("access_token"); expected != got {
+		t.Errorf("expected reauthorization with token returned by the authCallback; got %q", got)
+	}
+
+	// Simulate a successful reconnection.
+	in <- &proto.ProtocolMessage{
+		Action:            proto.ActionConnected,
+		ConnectionID:      "new-connection-id",
+		ConnectionDetails: &proto.ConnectionDetails{},
+	}
+
+	// Expect a UPDATED event.
+
+	var change ably.ConnectionStateChange
+	ablytest.Instantly.Recv(t, &change, stateChanges, t.Fatalf)
+
+	if change.Event != ably.ConnectionEventUpdated {
+		t.Fatalf("expected UPDATED event; got %v", change)
+	}
+
+	// Expect no further events.break
+	ablytest.Instantly.NoRecv(t, nil, stateChanges, t.Fatalf)
+}
+
 func TestRealtimeConn_RTN15i_OnErrorWhenConnected(t *testing.T) {
 	t.Parallel()
 
@@ -1231,6 +1494,7 @@ func TestRealtimeConn_RTN16(t *testing.T) {
 	if err := ablytest.Wait(channel.Publish("name", "data")); err != nil {
 		t.Fatal(err)
 	}
+	prevMsgSerial := c.Connection.MsgSerial()
 
 	client := app.NewRealtime(ably.ClientOptions{}.
 		Recover(c.Connection.RecoveryKey()))
@@ -1245,8 +1509,8 @@ func TestRealtimeConn_RTN16(t *testing.T) {
 			t.Errorf("expected the same connection")
 		}
 
-		if client.Connection.MsgSerial() != c.Connection.MsgSerial() {
-			t.Errorf("expected %d got %d", c.Connection.MsgSerial(), client.Connection.MsgSerial())
+		if expected, got := prevMsgSerial, client.Connection.MsgSerial(); expected != got {
+			t.Errorf("expected %d got %d", expected, got)
 		}
 		if true {
 			return
diff --git a/ably/realtime_conn_test.go b/ably/realtime_conn_test.go
index 7bf044f..c13252f 100644
--- a/ably/realtime_conn_test.go
+++ b/ably/realtime_conn_test.go
@@ -158,7 +158,7 @@ func TestRealtimeConn_ReceiveTimeout(t *testing.T) {
 	in <- connected
 
 	app, client := ablytest.NewRealtime(ably.ClientOptions{}.
-		Dial(ablytest.MessagePipe(in, out)).
+		Dial(ablytest.MessagePipe(in, out, ablytest.MessagePipeWithNowFunc(time.Now))).
 		RealtimeRequestTimeout(10 * time.Millisecond).
 		AutoConnect(false))
 	defer safeclose(t, app)