Repository: ably/ably-go
Pull request #148 (2020-04-20 18:54:13)

RTN23a: Receive from WebSocket with a timeout.
> If a transport does not receive any indication of activity on a
> transport for a period greater than the sum of the maxIdleInterval
> (which will be sent in the connectionDetails of the most recent
> CONNECTED message received on that transport) and the
> realtimeRequestTimeout, that transport should be disconnected.


Patch:
diff --git a/ably/internal/ablyutil/websocket.go b/ably/internal/ablyutil/websocket.go
index 11fc6c4..f6c2dc3 100644
--- a/ably/internal/ablyutil/websocket.go
+++ b/ably/internal/ablyutil/websocket.go
@@ -3,6 +3,7 @@ package ablyutil
 import (
 	"errors"
 	"net/url"
+	"time"
 
 	"github.com/ably/ably-go/ably/proto"
 
@@ -18,8 +19,14 @@ func (ws *WebsocketConn) Send(msg *proto.ProtocolMessage) error {
 	return ws.codec.Send(ws.conn, msg)
 }
 
-func (ws *WebsocketConn) Receive() (*proto.ProtocolMessage, error) {
+func (ws *WebsocketConn) Receive(deadline time.Time) (*proto.ProtocolMessage, error) {
 	msg := &proto.ProtocolMessage{}
+	if !deadline.IsZero() {
+		err := ws.conn.SetReadDeadline(deadline)
+		if err != nil {
+			return nil, err
+		}
+	}
 	err := ws.codec.Receive(ws.conn, &msg)
 	if err != nil {
 		return nil, err
diff --git a/ably/options.go b/ably/options.go
index 0c66773..d5726d5 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",
+	TimeoutDisconnect:        30 * time.Second,
+	RealtimeRequestTimeout:   10 * time.Second, // DF1b
+	TimeoutSuspended:         2 * time.Minute,
+	FallbackRetryTimeout:     10 * time.Minute,
 	IdempotentRestPublishing: false,
 }
 
@@ -199,10 +199,18 @@ 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 is the time period after which connect request is failed.
+	//
+	// Deprecated: use RealtimeRequestTimeout instead.
+	TimeoutConnect    time.Duration
+	TimeoutDisconnect time.Duration // time period after which disconnect request is failed
+	TimeoutSuspended  time.Duration // time period after which no more reconnection attempts are performed
+
+	// RealtimeRequestTimeout is the timeout for realtime connection establishment
+	// and each subsequent operation.
+	RealtimeRequestTimeout time.Duration
 
 	// Dial specifies the dial function for creating message connections used
 	// by RealtimeClient.
@@ -236,7 +244,7 @@ func (opts *ClientOptions) timeoutConnect() time.Duration {
 	if opts.TimeoutConnect != 0 {
 		return opts.TimeoutConnect
 	}
-	return defaultOptions.TimeoutConnect
+	return defaultOptions.RealtimeRequestTimeout
 }
 
 func (opts *ClientOptions) timeoutDisconnect() time.Duration {
@@ -260,6 +268,13 @@ func (opts *ClientOptions) fallbackRetryTimeout() time.Duration {
 	return defaultOptions.FallbackRetryTimeout
 }
 
+func (opts *ClientOptions) realtimeRequestTimeout() time.Duration {
+	if opts.RealtimeRequestTimeout != 0 {
+		return opts.RealtimeRequestTimeout
+	}
+	return defaultOptions.RealtimeRequestTimeout
+}
+
 func (opts *ClientOptions) restURL() string {
 	host := opts.RestHost
 	if host == "" {
diff --git a/ably/proto/protocol_message.go b/ably/proto/protocol_message.go
index 698c8bc..55953ec 100644
--- a/ably/proto/protocol_message.go
+++ b/ably/proto/protocol_message.go
@@ -3,6 +3,7 @@ package proto
 import (
 	"encoding/json"
 	"fmt"
+	"time"
 )
 
 const (
@@ -23,6 +24,7 @@ type ConnectionDetails struct {
 	MaxFrameSize       int64  `json:"maxFrameSize,omitempty" codec:"maxFrameSize,omitempty"`
 	MaxInboundRate     int64  `json:"maxInboundRate,omitempty" codec:"maxInboundRate,omitempty"`
 	ConnectionStateTTL int64  `json:"connectionStateTtl,omitempty" codec:"connectionStateTtl,omitempty"`
+	MaxIdleInterval    int64  `json:"maxIdleInterval,omitempty" codec:"maxIdleInterval,omitempty"`
 }
 
 func (c *ConnectionDetails) FromMap(ctx map[string]interface{}) {
@@ -213,7 +215,10 @@ type Conn interface {
 
 	// Receive reads ProtocolMessage from the connection.
 	// It is expected to block until whole message is read.
-	Receive() (*ProtocolMessage, error)
+	//
+	// If the deadline is greater than zero and no message is received before
+	// then, a net.Error with Timeout() == true is returned.
+	Receive(deadline time.Time) (*ProtocolMessage, error)
 
 	// Close closes the connection.
 	Close() error
diff --git a/ably/realtime_conn.go b/ably/realtime_conn.go
index 94289bb..238e540 100644
--- a/ably/realtime_conn.go
+++ b/ably/realtime_conn.go
@@ -12,8 +12,7 @@ import (
 )
 
 var (
-	errQueueing      = errors.New("unable to send messages in current state with disabled queueing")
-	errCloseInactive = errors.New("attempted to close inactive connection")
+	errQueueing = errors.New("unable to send messages in current state with disabled queueing")
 )
 
 // Conn represents a single connection RealtimeClient instantiates for
@@ -155,10 +154,13 @@ func (c *Conn) close() (Result, error) {
 	c.state.Lock()
 	defer c.state.Unlock()
 	switch c.state.current {
-	case StateConnClosing, StateConnClosed:
+	case
+		StateConnClosing,
+		StateConnClosed,
+		StateConnInitialized,
+		StateConnFailed,
+		StateConnDisconnected:
 		return nopResult, nil
-	case StateConnInitialized, StateConnFailed, StateConnDisconnected:
-		return nil, stateError(c.state.current, errCloseInactive)
 	}
 	res := c.state.listenResult(closeResultStates...)
 	c.state.set(StateConnClosing, nil)
@@ -326,8 +328,14 @@ func (c *Conn) logger() *LoggerOptions {
 }
 
 func (c *Conn) eventloop() {
+	var receiveTimeout time.Duration
+
 	for {
-		msg, err := c.conn.Receive()
+		var deadline time.Time
+		if receiveTimeout != 0 {
+			deadline = time.Now().Add(receiveTimeout) // RTN23a
+		}
+		msg, err := c.conn.Receive(deadline)
 		if err != nil {
 			c.state.Lock()
 			if c.state.current == StateConnClosed {
@@ -371,6 +379,9 @@ func (c *Conn) eventloop() {
 				c.details = *msg.ConnectionDetails
 				// Spec RSA7b3, RSA7b4, RSA12a
 				c.auth.updateClientID(c.details.ClientID)
+
+				maxIdleInterval := time.Duration(msg.ConnectionDetails.MaxIdleInterval) * time.Millisecond
+				receiveTimeout = c.opts.realtimeRequestTimeout() + maxIdleInterval // RTN23a
 			}
 			c.serial = -1
 			c.msgSerial = 0
@@ -403,8 +414,8 @@ func (vc verboseConn) Send(msg *proto.ProtocolMessage) error {
 	return vc.conn.Send(msg)
 }
 
-func (vc verboseConn) Receive() (*proto.ProtocolMessage, error) {
-	msg, err := vc.conn.Receive()
+func (vc verboseConn) Receive(deadline time.Time) (*proto.ProtocolMessage, error) {
+	msg, err := vc.conn.Receive(deadline)
 	if err != nil {
 		return nil, err
 	}


Test patch:
diff --git a/ably/ablytest/recorders.go b/ably/ablytest/recorders.go
index e3005f4..28bb714 100644
--- a/ably/ablytest/recorders.go
+++ b/ably/ablytest/recorders.go
@@ -306,10 +306,23 @@ func (pc pipeConn) Send(msg *proto.ProtocolMessage) error {
 	return nil
 }
 
-func (pc pipeConn) Receive() (*proto.ProtocolMessage, error) {
-	return <-pc.in, nil
+func (pc pipeConn) Receive(deadline time.Time) (*proto.ProtocolMessage, error) {
+	select {
+	case m := <-pc.in:
+		return m, nil
+	case <-time.After(time.Until(deadline)):
+		return nil, errTimeout{}
+	}
 }
 
+type errTimeout struct{}
+
+func (errTimeout) Error() string   { return "timeout" }
+func (errTimeout) Temporary() bool { return true }
+func (errTimeout) Timeout() bool   { return true }
+
+var _ net.Error = errTimeout{}
+
 func (pc pipeConn) Close() error {
 	return nil
 }
@@ -387,8 +400,8 @@ func (c recConn) Send(msg *proto.ProtocolMessage) error {
 	return nil
 }
 
-func (c recConn) Receive() (*proto.ProtocolMessage, error) {
-	msg, err := c.conn.Receive()
+func (c recConn) Receive(deadline time.Time) (*proto.ProtocolMessage, error) {
+	msg, err := c.conn.Receive(deadline)
 	if err != nil {
 		return nil, err
 	}
diff --git a/ably/realtime_client_test.go b/ably/realtime_client_test.go
index e31b18f..c35f211 100644
--- a/ably/realtime_client_test.go
+++ b/ably/realtime_client_test.go
@@ -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 := client.Close(); err != nil {
 			t.Errorf("%s (host=%s)", err, host)
 			continue
 		}
@@ -163,9 +163,5 @@ func TestRealtimeClient_multiple(t *testing.T) {
 func TestRealtimeClient_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)
-	}
+	defer safeclose(t, app, client)
 }
diff --git a/ably/realtime_conn_test.go b/ably/realtime_conn_test.go
index 49b6b25..a34a524 100644
--- a/ably/realtime_conn_test.go
+++ b/ably/realtime_conn_test.go
@@ -7,6 +7,7 @@ import (
 
 	"github.com/ably/ably-go/ably"
 	"github.com/ably/ably-go/ably/ablytest"
+	"github.com/ably/ably-go/ably/proto"
 )
 
 func await(fn func() ably.StateEnum, state ably.StateEnum) error {
@@ -129,13 +130,67 @@ func TestRealtimeConn_AuthError(t *testing.T) {
 	if err != nil {
 		t.Fatalf("NewRealtimeClient()=%v", err)
 	}
+	defer safeclose(t, client)
 	if err = ablytest.Wait(client.Connection.Connect()); 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 {
-		t.Fatal("Close(): want err != nil")
+}
+
+func TestRealtimeConn_ReceiveTimeout(t *testing.T) {
+	t.Parallel()
+
+	const maxIdleInterval = 20
+	const realtimeRequestTimeout = 10 * time.Millisecond
+
+	in := make(chan *proto.ProtocolMessage, 16)
+	out := make(chan *proto.ProtocolMessage, 16)
+
+	connected := &proto.ProtocolMessage{
+		Action:       proto.ActionConnected,
+		ConnectionID: "connection-id",
+		ConnectionDetails: &proto.ConnectionDetails{
+			MaxIdleInterval: maxIdleInterval,
+		},
+	}
+	in <- connected
+
+	app, client := ablytest.NewRealtimeClient(&ably.ClientOptions{
+		Dial:                   ablytest.MessagePipe(in, out),
+		RealtimeRequestTimeout: 10 * time.Millisecond,
+		NoConnect:              true,
+	})
+	defer safeclose(t, app)
+
+	states := make(chan ably.State, 10)
+	client.Connection.On(states, ably.StateConnConnected, ably.StateConnFailed)
+
+	client.Connection.Connect()
+
+	var state ably.State
+
+	select {
+	case state = <-states:
+	case <-time.After(10 * time.Millisecond):
+		t.Fatal("didn't receive state change event")
+	}
+
+	if expected, got := ably.StateConnConnected, state.State; expected != got {
+		t.Fatalf("expected %v, got %v", expected, got)
+	}
+
+	leeway := 10 * time.Millisecond
+	select {
+	case state = <-states:
+	case <-time.After(realtimeRequestTimeout + time.Duration(maxIdleInterval)*time.Millisecond + leeway):
+		t.Fatal("didn't receive state change event")
+	}
+
+	// TODO: Should be Disconnected, not Failed
+	// Part of https://ably.atlassian.net/browse/FEA-391
+	if expected, got := ably.StateConnFailed, state.State; expected != got {
+		t.Fatalf("expected %v, got %v", expected, got)
 	}
 }